Deterministic network hardening checks — self-contained Python, config-driven (plain-HTTP admin UI, WPA3, legacy TLS, weak ciphers, certs, security headers, HTTP methods, mgmt APIs, unknown LAN devices, firmware via SSDP)

pentest-check.py

#!/usr/bin/env python3
"""Deterministic network hardening checks — self-contained.

Usage:
  python3 pentest-check.py            # run checks with built-in CONFIG
  python3 pentest-check.py --report   # also regenerate the HTML report
  python3 pentest-check.py -c cfg.json  # use an external config instead
  python3 pentest-check.py --init     # write the built-in CONFIG to pentest-check.json

Edit the CONFIG dict at the top to match your network. All checks are
read-only probes (HTTP GET/OPTIONS, TCP connect, TLS handshake, SSDP,
LAN TCP sweep). No exploitation, no credential attacks, no changes.

Checks:
  1  plain-HTTP admin UI without redirect
  2  Wi-Fi security mode offers WPA3
  3  legacy TLS versions rejected (all TLS devices)
  7  weak TLS cipher suites rejected (CBC/3DES)
  8  HTTPS certificate sane (not self-signed, not expiring)
  9  HTTP security headers present on admin UI
  10 no risky HTTP methods (TRACE/PUT/DELETE)
  11 management API descriptions not readable unauthenticated
  12 no unknown LAN devices vs allowlist
  4  optional services closed
  5  unauthenticated status pages
  6  firmware currency via SSDP
  B  baselines: expected services reachable, legacy ports closed

Exit code: 0 if no FAIL, 1 otherwise.
"""

import argparse
import concurrent.futures
import ipaddress
import json
import re
import socket
import ssl
import subprocess
import sys
import tempfile
import urllib.error
import urllib.request
import warnings
from pathlib import Path

warnings.filterwarnings("ignore", category=DeprecationWarning)

DEFAULT_CONFIG = Path("pentest-check.json")

# Edit targets inline here, or keep them in a JSON config (-c overrides).
# Run with --init to generate a config file from this dict.
CONFIG = {
    "wifi_interface": "en0",       # macOS Wi-Fi interface, "" to skip Wi-Fi checks
    "router": {
        "host": "192.168.1.1",
        "expected_firmware": "",   # e.g. "7.57"; empty string skips the check
        "optional_services": [],   # ports that should stay closed, e.g. [49000]
        "expected_open": [443],
    },
    "repeater": {
        "host": "",                # empty host = role skipped
        "expected_open": [443],
    },
    "printer": {
        "host": "",
        "expected_open": [9100],
        "status_path": "/general/status.html",
    },
    "legacy_ports": [21, 23, 69, 445, 3389, 5900],
    # LAN discovery: sweep cidr and WARN on devices not in known_hosts.
    # Empty cidr skips the check. "me" in known_hosts = the scanning host.
    "lan_cidr": "",
    "known_hosts": [],
}

MGMT_API_PATHS = [
    "/tr64desc.xml", "/igddesc.xml", "/fboxdesc.xml", "/MediaServerDevDesc.xml",
]
MGMT_PORTS = [(80, False), (49000, False), (49443, True)]  # (port, https)
HTTP_TIMEOUT = 8
TCP_TIMEOUT = 5
SSDP_TIMEOUT = 4
SWEEP_PORTS = [22, 80, 443, 445, 9100]
SWEEP_TIMEOUT = 0.4
SWEEP_WORKERS = 64

results = []


def record(check_id, finding, status, detail=""):
    results.append((check_id, finding, status, detail))


# --- probes -------------------------------------------------------------------


def http_req(url, method="GET"):
    req = urllib.request.Request(url, method=method,
                                 headers={"User-Agent": "pentest-check/1.0"})
    ctx = ssl._create_unverified_context()
    try:
        with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT, context=ctx) as r:
            return r.status, r.geturl(), dict(r.headers)
    except urllib.error.HTTPError as e:
        return e.code, url, dict(e.headers)
    except Exception as e:
        return None, str(e), {}


def tcp_open(host, port, timeout=TCP_TIMEOUT):
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


def tls_probe(host, version=None, ciphers=None):
    """Handshake with optional pinned version/ciphers. True=success, False=rejected, None=untestable."""
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    try:
        if version:
            ctx.minimum_version = version
            ctx.maximum_version = version
        if ciphers:
            ctx.set_ciphers(ciphers)
    except (ValueError, ssl.SSLError):
        return None
    try:
        with socket.create_connection((host, 443), timeout=TCP_TIMEOUT) as sock:
            with ctx.wrap_socket(sock, server_hostname=host):
                return True
    except ssl.SSLError:
        return False
    except OSError:
        return None


def get_certificate(host):
    pem = ssl.get_server_certificate((host, 443))
    with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as f:
        f.write(pem)
        path = f.name
    return ssl._ssl._test_decode_cert(path)


def ssdp_server_header(host, attempts=2):
    msg = (
        "M-SEARCH * HTTP/1.1\r\nHOST: 239.255.255.250:1900\r\n"
        'MAN: "ssdp:discover"\r\nMX: 2\r\nST: ssdp:all\r\n\r\n'
    ).encode()
    for _ in range(attempts):
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.settimeout(SSDP_TIMEOUT)
        try:
            s.sendto(msg, ("239.255.255.250", 1900))
            while True:
                data, addr = s.recvfrom(4096)
                if addr[0] == host:
                    m = re.search(r"Server:\s*(.+)", data.decode(errors="replace"))
                    return m.group(1).strip() if m else None
        except socket.timeout:
            continue
        finally:
            s.close()
    return None


def local_ip(router_host):
    """LAN IP of this machine via a connected UDP socket (no packet sent)."""
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        s.connect((router_host, 9))
        return s.getsockname()[0]
    except OSError:
        return None
    finally:
        s.close()


# --- checks -------------------------------------------------------------------


def check_admin_http(role, cfg):
    host = cfg["host"]
    code, url, _ = http_req(f"http://{host}/")
    if code is None:
        record("1", f"[{role}] plain-HTTP admin UI", "WARN", f"unreachable: {url}")
    elif code == 200:
        record("1", f"[{role}] plain-HTTP admin UI", "FAIL", "port 80 serves content with no redirect")
    else:
        record("1", f"[{role}] plain-HTTP admin UI", "PASS", f"HTTP {code} -> {url}")


def check_wifi(cfg):
    iface = cfg.get("wifi_interface", "")
    if not iface:
        return
    try:
        out = subprocess.run(
            ["ipconfig", "getsummary", iface], capture_output=True, text=True, timeout=10
        ).stdout
        sec = re.findall(r"Security\s*:\s*(\S+)", out)
        if not sec:
            record("2", "Wi-Fi security", "WARN", f"no security info on {iface} (Wi-Fi off?)")
        elif any("WPA3" in s for s in sec):
            record("2", "Wi-Fi security", "PASS", f"security={sec[0]}")
        elif any(s in ("WEP", "OPEN") for s in sec):
            record("2", "Wi-Fi security", "FAIL", f"insecure mode: {sec[0]}")
        else:
            record("2", "Wi-Fi security", "FAIL", f"security={sec[0]}, no WPA3")
    except Exception as e:
        record("2", "Wi-Fi security", "WARN", f"could not query: {e}")


def check_tls(role, cfg):
    """Legacy TLS versions and weak ciphers on any device exposing 443."""
    host = cfg["host"]
    if not tcp_open(host, 443):
        record("3", f"[{role}] TLS surface", "WARN", "443 closed, TLS checks skipped")
        return
    failures = []
    for name, ver in (("TLSv1", ssl.TLSVersion.TLSv1), ("TLSv1.1", ssl.TLSVersion.TLSv1_1)):
        sup = tls_probe(host, version=ver)
        if sup is None:
            record("3", f"[{role}] {name} disabled", "WARN", "could not test")
        elif sup:
            failures.append(name)
            record("3", f"[{role}] {name} disabled", "FAIL", f"{name} handshake succeeded")
        else:
            record("3", f"[{role}] {name} disabled", "PASS", f"{name} rejected")
    weak = tls_probe(host, version=ssl.TLSVersion.TLSv1_2,
                     ciphers="AES128-SHA:AES256-SHA:DES-CBC3-SHA")
    if weak is None:
        record("7", f"[{role}] weak ciphers rejected", "WARN", "could not test")
    elif weak:
        record("7", f"[{role}] weak ciphers rejected", "FAIL",
               "TLS 1.2 CBC/3DES-only handshake succeeded")
    else:
        record("7", f"[{role}] weak ciphers rejected", "PASS", "TLS 1.2 CBC/3DES-only handshake rejected")


def check_certificate(role, cfg):
    host = cfg["host"]
    try:
        cert = get_certificate(host)
    except Exception as e:
        record("8", f"[{role}] HTTPS certificate", "WARN", f"could not read cert: {e}")
        return
    issuer = dict(x[0] for x in cert.get("issuer", ())).get("organizationName", "?")
    subject = dict(x[0] for x in cert.get("subject", ())).get("organizationName", "?")
    if issuer == subject:
        record("8", f"[{role}] HTTPS certificate", "WARN", f"self-signed (issuer={issuer}); common for LAN appliances")
    else:
        record("8", f"[{role}] HTTPS certificate", "PASS", f"CA-issued: {issuer}")
    not_after = cert.get("notAfter", "")
    m = re.match(r"(\w+ +\d+ \d+:\d+:\d+ \w+ (\w+ \d+ \d+))", not_after)
    if m:
        import datetime
        try:
            exp = datetime.datetime.strptime(m.group(2), "%b %d %Y")
            days = (exp - datetime.datetime.now()).days
            if days < 30:
                record("8", f"[{role}] HTTPS certificate", "FAIL", f"expires in {days} days ({not_after})")
        except ValueError:
            pass


def check_headers_and_methods(role, cfg):
    host = cfg["host"]
    code, _, headers = http_req(f"https://{host}/")
    if code is None:
        code, _, headers = http_req(f"http://{host}/")
    if code is None:
        record("9", f"[{role}] HTTP security headers", "WARN", "admin UI unreachable")
        record("10", f"[{role}] HTTP methods", "WARN", "admin UI unreachable")
        return
    lower = {k.lower(): v for k, v in headers.items()}
    missing = [h for h in ("content-security-policy", "x-frame-options",
                           "x-content-type-options") if h not in lower]
    if missing:
        record("9", f"[{role}] HTTP security headers", "WARN", f"missing: {', '.join(missing)}")
    else:
        record("9", f"[{role}] HTTP security headers", "PASS", "CSP, X-Frame-Options, X-Content-Type-Options present")
    allow = lower.get("allow", "")
    risky = [m for m in ("TRACE", "PUT", "DELETE") if m in allow.upper()]
    if risky:
        record("10", f"[{role}] HTTP methods", "WARN", f"advertised risky methods: {', '.join(risky)}")
    else:
        record("10", f"[{role}] HTTP methods", "PASS", f"no risky methods advertised (Allow: {allow or 'n/a'})")


def check_mgmt_api(role, cfg):
    host = cfg["host"]
    open_paths = []
    for port, use_tls in MGMT_PORTS:
        if not tcp_open(host, port):
            continue
        scheme = "https" if use_tls else "http"
        for path in MGMT_API_PATHS:
            code, _, _ = http_req(f"{scheme}://{host}:{port}{path}")
            if code == 200:
                open_paths.append(f":{port}{path}")
    if open_paths:
        record("11", f"[{role}] mgmt API descriptions", "WARN",
               f"readable unauthenticated: {', '.join(open_paths)} (by design on many routers)")
    else:
        record("11", f"[{role}] mgmt API descriptions", "PASS", "not readable unauthenticated")


def check_optional_services(role, cfg):
    for port in cfg.get("optional_services", []):
        closed = not tcp_open(cfg["host"], port)
        record("4", f"[{role}] optional service {port}", "PASS" if closed else "WARN",
               "closed" if closed else "open (close if unused)")


def check_unauth_pages(role, cfg):
    path = cfg.get("status_path")
    if not path:
        return
    code, _, _ = http_req(f"http://{cfg['host']}{path}")
    if code == 200:
        record("5", f"[{role}] unauthenticated status page", "WARN", f"{path} readable without login")
    elif code is None:
        record("5", f"[{role}] unauthenticated status page", "WARN", "unreachable")
    else:
        record("5", f"[{role}] unauthenticated status page", "PASS", f"HTTP {code}, auth required")


def check_firmware(role, cfg):
    expected = cfg.get("expected_firmware", "")
    if not expected:
        return
    hdr = ssdp_server_header(cfg["host"])
    if hdr is None:
        record("6", f"[{role}] firmware currency", "WARN", "no SSDP reply")
    elif expected in hdr:
        record("6", f"[{role}] firmware currency", "PASS", hdr)
    else:
        record("6", f"[{role}] firmware currency", "FAIL", f"'{hdr}' lacks expected {expected}")


def check_lan_devices(cfg):
    cidr = cfg.get("lan_cidr", "")
    if not cidr:
        return
    known = set(cfg.get("known_hosts", []))
    router_host = next((d["host"] for d in cfg.values()
                        if isinstance(d, dict) and d.get("host")), None)
    me = local_ip(router_host) if router_host else None
    if me:
        known.add(me)
    net = ipaddress.ip_network(cidr, strict=False)
    hosts = [str(h) for h in net.hosts()]

    def probe(ip):
        for port in SWEEP_PORTS:
            if tcp_open(ip, port, timeout=SWEEP_TIMEOUT):
                return ip
        return None

    with concurrent.futures.ThreadPoolExecutor(max_workers=SWEEP_WORKERS) as ex:
        found = {ip for ip in ex.map(probe, hosts) if ip}
    unknown = sorted(found - known)
    if unknown:
        record("12", "LAN unknown devices", "WARN", f"not in allowlist: {', '.join(unknown)}")
    else:
        record("12", "LAN unknown devices", "PASS", f"{len(found)} devices found, all allowlisted")


def check_baselines(cfg):
    for role, dev in cfg.items():
        if not isinstance(dev, dict) or not dev.get("host"):
            continue
        for port in dev.get("expected_open", []):
            label = f"[{role}] service {dev['host']}:{port}"
            record("B", label, "PASS" if tcp_open(dev["host"], port) else "FAIL", "expected reachable")
        for port in cfg.get("legacy_ports", []):
            label = f"[{role}] legacy port {dev['host']}:{port} closed"
            record("B", label, "PASS" if not tcp_open(dev["host"], port) else "FAIL", "must stay closed")


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("-c", "--config", type=Path, help="external JSON config (default: built-in CONFIG)")
    ap.add_argument("--init", action="store_true", help="write the built-in CONFIG to pentest-check.json and exit")
    ap.add_argument("--report", action="store_true", help="regenerate the HTML report after checking")
    args = ap.parse_args()

    if args.init:
        DEFAULT_CONFIG.write_text(json.dumps(CONFIG, indent=2) + "\n")
        print(f"template written to {DEFAULT_CONFIG}")
        return

    cfg = json.loads(args.config.read_text()) if args.config else CONFIG

    device_roles = ("router", "repeater", "gateway", "ap", "nas", "host")
    devices = {r: cfg[r] for r in device_roles
               if isinstance(cfg.get(r), dict) and cfg[r].get("host")}

    for role, dev in devices.items():
        check_admin_http(role, dev)
        check_tls(role, dev)
        check_certificate(role, dev)
        check_headers_and_methods(role, dev)
        check_mgmt_api(role, dev)
        check_optional_services(role, dev)
        check_firmware(role, dev)
    printer = cfg.get("printer")
    if isinstance(printer, dict) and printer.get("host"):
        check_unauth_pages("printer", printer)
    check_wifi(cfg)
    check_lan_devices(cfg)
    check_baselines(cfg)

    width = max(len(f) for _, f, _, _ in results)
    n_fail = sum(1 for r in results if r[2] == "FAIL")
    n_warn = sum(1 for r in results if r[2] == "WARN")
    n_pass = len(results) - n_fail - n_warn
    for cid, finding, status, detail in results:
        print(f"[{status}] #{cid:<2} {finding:<{width}}  {detail}")
    print(f"\n{len(results)} checks: {n_fail} FAIL, {n_warn} WARN, {n_pass} PASS")

    if args.report:
        report = Path(__file__).with_name("pentest-report-2026-09-16.md")
        html = report.with_suffix(".html")
        subprocess.run(
            ["pandoc", str(report), "-o", str(html), "--standalone", "--css", "print.css",
             "--embed-resources", "--variable", "pagetitle=Local Network Pentest Report"],
            cwd=str(report.parent), check=True,
        )
        subprocess.run(["open", str(html)], check=True)
        print(f"report regenerated: {html}")

    sys.exit(1 if n_fail else 0)


if __name__ == "__main__":
    main()