Project: Build a Port Scanner

Build scanner.py, an ethical, useful port scanner:

• Authorization allow-list — refuses any host not explicitly permitted (Lesson 19).
• Concurrent scanning — a thread pool to scan many ports fast.
• Rate limiting — bounded concurrency so it's not aggressive.
• Banner grabbing + a self-audit report — what's open, what's there, and "should it be?"

You already have the atom (Lesson 19's port_state) and the gate (assert_authorized). Scanning 1000 ports one at a time with a 1-second timeout would take ~16 minutes — too slow. The fix is concurrency: scan many ports in parallel with a thread pool. But unbounded concurrency is aggressive (and looks like an attack), so we cap it — fast and polite.

The authorization gate (first, always)

import socket

# ONLY hosts you own or that explicitly permit scanning:
AUTHORISED = {
    "127.0.0.1", "localhost",
    "scanme.nmap.org",          # nmap's official, scan-permitted test host
    # add your own lab VM IPs here, e.g. "192.168.56.10"
}

def assert_authorized(host: str) -> None:
    # resolve to compare both name and IP forms
    try:
        ip = socket.gethostbyname(host)
    except socket.gaierror:
        ip = host
    if host not in AUTHORISED and ip not in AUTHORISED:
        raise PermissionError(
            f"Refusing to scan '{host}'. Only scan hosts you OWN or that "
            f"explicitly permit it. Add it to AUTHORISED only if you're sure.")

Scanning one port (with banner grab)

def scan_port(host: str, port: int, timeout: float = 1.0) -> dict | None:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.settimeout(timeout)
        if s.connect_ex((host, port)) != 0:
            return None                       # not open — skip
        banner = ""
        try:
            banner = s.recv(128).decode(errors="replace").strip()
        except (socket.timeout, OSError):
            pass
        return {"port": port, "service": _guess_service(port), "banner": banner}

COMMON = {21: "ftp", 22: "ssh", 23: "telnet", 25: "smtp", 53: "dns",
          80: "http", 443: "https", 3306: "mysql", 5432: "postgres",
          6379: "redis", 8080: "http-alt", 27017: "mongodb"}
def _guess_service(port: int) -> str:
    return COMMON.get(port, "unknown")

Concurrent + rate-limited

from concurrent.futures import ThreadPoolExecutor, as_completed

def scan(host: str, ports, max_workers: int = 100) -> list[dict]:
    assert_authorized(host)                   # GATE — before any packet
    open_ports = []
    # bounded thread pool = concurrency cap = built-in rate limiting
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {pool.submit(scan_port, host, p): p for p in ports}
        for fut in as_completed(futures):
            result = fut.result()
            if result:
                open_ports.append(result)
    return sorted(open_ports, key=lambda r: r["port"])

The ThreadPoolExecutor with a max_workers cap is both the speedup and the rate limit: at most 100 connections in flight, so we're fast but never floods the target. The gate runs first, so an unauthorized host fails before a single packet is sent.

Wrap it in a CLI that produces a defensive report — not just "open ports" but "should this be open?" — pulling in Level 7 reporting.

import argparse, json, logging
from datetime import datetime

logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("scanner")

# ports that are usually a RED FLAG if exposed beyond localhost
SHOULD_NOT_EXPOSE = {3306, 5432, 6379, 27017, 23, 21}

def audit(open_ports: list[dict], host: str) -> dict:
    findings = []
    for p in open_ports:
        note = "review: is this meant to be reachable?"
        if p["port"] in SHOULD_NOT_EXPOSE:
            note = "⚠️ HIGH: database/admin/plaintext service exposed — firewall it"
        findings.append({**p, "note": note})
    return {"host": host, "scanned_at": datetime.now().isoformat(),
            "open_ports": findings}

def main():
    p = argparse.ArgumentParser(description="Ethical port scanner (own hosts only).")
    p.add_argument("host")
    p.add_argument("--start", type=int, default=1)
    p.add_argument("--end", type=int, default=1024)
    p.add_argument("--out", default="scan.json")
    a = p.parse_args()

    log.info("⚠️  Authorised targets only. Scanning %s ports %d-%d…",
             a.host, a.start, a.end)
    try:
        results = scan(a.host, range(a.start, a.end + 1))
    except PermissionError as e:
        log.error("%s", e); return

    report = audit(results, a.host)
    log.info("\n%d open port(s) on %s:", len(results), a.host)
    for f in report["open_ports"]:
        line = f"  {f['port']:5} {f['service']:10}"
        if f["banner"]:
            line += f" [{f['banner'][:40]}]"
        log.info(line + f"\n        → {f['note']}")
    with open(a.out, "w") as fp:
        json.dump(report, fp, indent=2)
    log.info("\nreport → %s", a.out)

if __name__ == "__main__":
    main()

⚠️  Authorised targets only. Scanning 127.0.0.1 ports 1-1024…

3 open port(s) on 127.0.0.1:
    22    ssh        [SSH-2.0-OpenSSH_9.6]
        → review: is this meant to be reachable?
   3000   unknown
        → review: is this meant to be reachable?
   5432   postgres
        → ⚠️ HIGH: database/admin/plaintext service exposed — firewall it
report → scan.json

# (point it at an unauthorised host → PermissionError, no packets sent)

Read the result

This is a tool you'd actually use to harden your own systems. The authorization gate runs first — try an unlisted host and it refuses before touching the network. The thread pool makes a 1024-port scan finish in seconds while capping concurrency for courtesy. Banner grabbing (Lesson 7) identifies services, and the audit step adds the defender's judgement: an exposed PostgreSQL is flagged HIGH, because a database reachable beyond localhost is a classic misconfiguration (Lesson 34). The JSON report feeds your evidence/reporting workflow (Lesson 46).

Scan 127.0.0.1 (your machine) or scanme.nmap.org (explicitly permitted) only.

import time
t = time.perf_counter()
scan("127.0.0.1", range(1, 1025), max_workers=1)
print("serial:", time.perf_counter()-t)
t = time.perf_counter()
scan("127.0.0.1", range(1, 1025), max_workers=100)
print("pool:  ", time.perf_counter()-t)

A responsible scanner is fast, polite, and safe. Concurrency (a ThreadPoolExecutor) makes it fast; a bounded max_workers doubles as rate limiting so it's not aggressive; and an authorization allow-list checked before any packet makes it physically refuse unauthorized targets. Banner grabbing identifies services, and an audit step turns "open ports" into defensive findings ("this database shouldn't be exposed"). The whole point is the self-audit: scanning your own systems to find and fix misconfigurations before attackers do. Mature tools like nmap go further — but you now understand exactly what they're doing and why the ethics are non-negotiable.