Building a Port Scanner

import argparse

def parse_args():
    p = argparse.ArgumentParser(description="pyscan — Python Port Scanner")
    p.add_argument("host",                  help="Target hostname or IP")
    p.add_argument("-p", "--ports",         default="1-1024",
                   help="Port range: 1-1024 or comma list 22,80,443")
    p.add_argument("-t", "--threads",       type=int, default=100)
    p.add_argument("-T", "--timeout",       type=float, default=1.0)
    p.add_argument("-o", "--output",        help="Output file (json or csv)")
    p.add_argument("--banner",              action="store_true",
                   help="Grab service banners")
    return p.parse_args()

def parse_ports(port_str):
    ports = set()
    for part in port_str.split(","):
        part = part.strip()
        if "-" in part:
            start, end = part.split("-")
            ports.update(range(int(start), int(end) + 1))
        else:
            ports.add(int(part))
    return sorted(ports)

import socket
from concurrent.futures import ThreadPoolExecutor, as_completed

def scan_port(host, port, timeout, grab_banner):
    result = {"port": port, "status": "closed", "banner": ""}
    try:
        s = socket.socket()
        s.settimeout(timeout)
        s.connect((host, port))
        result["status"] = "open"
        if grab_banner:
            try:
                banner = s.recv(1024).decode(errors="replace").strip()
                result["banner"] = banner[:100]
            except socket.timeout:
                pass
        s.close()
    except OSError:
        pass
    return result

def run_scan(host, ports, threads, timeout, grab_banner):
    open_ports = []
    with ThreadPoolExecutor(max_workers=threads) as pool:
        futures = {
            pool.submit(scan_port, host, p, timeout, grab_banner): p
            for p in ports
        }
        for future in as_completed(futures):
            res = future.result()
            if res["status"] == "open":
                open_ports.append(res)
                print(f"  OPEN  {res['port']:5d}  {res['banner'][:50]}")
    return sorted(open_ports, key=lambda x: x["port"])

import json, csv

def write_json(results, host, path):
    data = {"host": host, "open_ports": results}
    with open(path, "w") as f:
        json.dump(data, f, indent=2)
    print(f"JSON report → {path}")

def write_csv(results, path):
    with open(path, "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=["port", "status", "banner"])
        w.writeheader()
        w.writerows(results)
    print(f"CSV report  → {path}")

#!/usr/bin/env python3
"""pyscan — multi-threaded Python port scanner"""
import argparse, socket, json, csv, time
from concurrent.futures import ThreadPoolExecutor, as_completed

def parse_args():
    p = argparse.ArgumentParser(description="pyscan")
    p.add_argument("host")
    p.add_argument("-p", "--ports",   default="1-1024")
    p.add_argument("-t", "--threads", type=int,   default=100)
    p.add_argument("-T", "--timeout", type=float, default=1.0)
    p.add_argument("-o", "--output")
    p.add_argument("--banner", action="store_true")
    return p.parse_args()

def parse_ports(s):
    ports = set()
    for part in s.split(","):
        part = part.strip()
        if "-" in part:
            a, b = part.split("-")
            ports.update(range(int(a), int(b)+1))
        else:
            ports.add(int(part))
    return sorted(ports)

def scan_one(host, port, timeout, banner):
    r = {"port": port, "status": "closed", "banner": ""}
    try:
        s = socket.socket()
        s.settimeout(timeout)
        s.connect((host, port))
        r["status"] = "open"
        if banner:
            try:
                r["banner"] = s.recv(1024).decode(errors="replace").strip()[:80]
            except Exception:
                pass
        s.close()
    except OSError:
        pass
    return r

def main():
    args  = parse_args()
    ports = parse_ports(args.ports)
    host  = socket.gethostbyname(args.host)
    print(f"Scanning {host} ({args.host}) — {len(ports)} ports — {args.threads} threads")
    t0 = time.time()

    results = []
    with ThreadPoolExecutor(max_workers=args.threads) as pool:
        futures = {pool.submit(scan_one, host, p, args.timeout, args.banner): p for p in ports}
        for fut in as_completed(futures):
            r = fut.result()
            if r["status"] == "open":
                results.append(r)
                print(f"  OPEN  {r['port']:5d}  {r['banner'][:40]}")

    results.sort(key=lambda x: x["port"])
    elapsed = time.time() - t0
    print(f"\nDone in {elapsed:.1f}s — {len(results)} open ports")

    if args.output:
        if args.output.endswith(".csv"):
            with open(args.output, "w", newline="") as f:
                w = csv.DictWriter(f, fieldnames=["port","status","banner"])
                w.writeheader(); w.writerows(results)
        else:
            with open(args.output, "w") as f:
                json.dump({"host": host, "ports": results}, f, indent=2)

if __name__ == "__main__":
    main()

python3 pyscan.py localhost -p 1-1024 --banner -o results.json
python3 pyscan.py scanme.nmap.org -p 22,80,443 --banner