client_portscan.py

#!/usr/bin/env python3
from __future__ import annotations

import ipaddress
import json
import os
import subprocess
import time
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from pathlib import Path


SUMMARY_FILE = Path("/var/lib/openvpn-lab/passive-summary.json")
OUTPUT_FILE = Path("/var/lib/openvpn-lab/portscan.json")
VPN_NET = ipaddress.ip_network("10.8.0.0/24")
PORTS = ",".join(str(port) for port in (
    21, 22, 23, 25, 53, 80, 110, 135, 139, 143, 389, 443, 445,
    3389, 5357, 5900, 5901, 5985, 5986, 8000, 8080, 8443,
    1080, 1081, 3128, 8118, 8888, 8889,
    5938, 6568, 7070, 21115, 21116, 21117, 21118, 21119,
    9001, 9030, 9040, 9050, 9051, 9150, 9151,
))


def now_iso() -> str:
    return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


def active_vpn_ips() -> list[str]:
    try:
        data = json.loads(SUMMARY_FILE.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return []
    ips: list[str] = []
    for row in (data.get("clients") or {}).values():
        ip = str(row.get("vpn_ip") or "")
        if not row.get("active"):
            continue
        try:
            if ipaddress.ip_address(ip) in VPN_NET:
                ips.append(ip)
        except ValueError:
            continue
    return sorted(set(ips), key=lambda value: tuple(int(part) for part in value.split(".")))


def scan_ip(ip: str) -> dict:
    base = {
        "last_scan": now_iso(),
        "state": "unknown",
        "note": "",
        "open_ports": [],
    }
    cmd = [
        "nmap",
        "-Pn",
        "-n",
        "--max-retries",
        "1",
        "--host-timeout",
        "25s",
        "-T3",
        "-p",
        PORTS,
        "-oX",
        "-",
        ip,
    ]
    try:
        proc = subprocess.run(cmd, check=False, capture_output=True, text=True, timeout=35)
    except Exception as exc:
        base["state"] = "error"
        base["note"] = str(exc)[:240]
        return base
    if proc.returncode not in {0, 1}:
        base["state"] = "error"
        base["note"] = (proc.stderr or proc.stdout)[-240:]
        return base
    try:
        root = ET.fromstring(proc.stdout)
    except ET.ParseError as exc:
        base["state"] = "parse_error"
        base["note"] = str(exc)[:240]
        return base
    host = root.find("host")
    if host is None:
        base["state"] = "no_host"
        return base
    status = host.find("status")
    base["state"] = status.get("state", "unknown") if status is not None else "unknown"
    for port in host.findall("./ports/port"):
        state = port.find("state")
        if state is None or state.get("state") != "open":
            continue
        service = port.find("service")
        base["open_ports"].append({
            "port": int(port.get("portid", "0")),
            "protocol": port.get("protocol", "tcp"),
            "service": service.get("name", "") if service is not None else "",
            "product": service.get("product", "") if service is not None else "",
        })
    if not base["open_ports"] and base["state"] == "up":
        base["note"] = "host reachable, selected common ports closed/filtered"
    return base


def main() -> int:
    ips = active_vpn_ips()
    result = {
        "generated_at": now_iso(),
        "mode": "active-scan-vpn-clients-only",
        "ports": PORTS,
        "clients": {},
    }
    for ip in ips:
        result["clients"][ip] = scan_ip(ip)
    OUTPUT_FILE.parent.mkdir(parents=True, exist_ok=True)
    tmp = OUTPUT_FILE.with_suffix(".tmp")
    tmp.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    os.chmod(tmp, 0o644)
    os.replace(tmp, OUTPUT_FILE)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())