#!/usr/bin/env python3
# jozapf.de toolbox · rev jzt-7c3f9a2e
# TOOLBOX-TILE: {"title": "Port Usage Report", "desc": "CLI-Tool: lauschende TCP/UDP-Ports mit Prozess- und Docker-Kontext plus Firewall-Status.", "icon": "🌐", "type": "download", "os": ["linux", "windows"], "order": 30}
"""
Port & Firewall Report (HTML default)
====================================

This script generates a port usage report with process attribution and optional
Docker container correlation, plus a UFW firewall status block at the top.

Outputs
-------
By default, an HTML report is written to:
  <script_dir>/reports/YYYYMMDD_<host>_port_fw_report.html

Optional JSON output:
  --json  -> additionally writes
            <script_dir>/reports/YYYYMMDD_<host>_port_fw_report.json

Examples
--------
  ./port_usage_report.py
  ./port_usage_report.py --json
  ./port_usage_report.py --check-port 8123
  ./port_usage_report.py --check-port 8123 --host 127.0.0.1 --json
  ./port_usage_report.py --out-dir /tmp/reports --with-time

Port check exit codes
---------------------
If you use --check-port:
  0 -> port is free (connect() failed / no listener)
  1 -> port is in use (connect() succeeded)

Notes
-----
- The HTML uses a shared dark-mode theme and a full-width layout.
- The script never prompts for sudo; it may try `sudo -n` for UFW (non-interactive).

"""

from __future__ import annotations

import argparse
import datetime as dt
import json
import os
import shutil
import socket
import subprocess
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

try:
    import psutil
except Exception as exc:  # pragma: no cover
    raise SystemExit("Missing dependency: psutil. Install with: pip install psutil") from exc

import socket as pysocket  # for check_port_free


__version__ = "2.0.0"
SCHEMA = "jozapf.report.port_fw"
SCHEMA_VERSION = "1.0.0"


# ---- Shared HTML Theme (Darkmode + Fullscreen) --------------------------------

HTML_CSS = r"""
:root {
  --bg-primary: #0f172a;
  --bg-secondary: #1e293b;
  --bg-tertiary: #334155;
  --text-primary: #f1f5f9;
  --text-secondary: #94a3b8;
  --accent: #3b82f6;
  --accent-secondary: #8b5cf6;
  --border: #334155;
  --code-bg: #0b1220;
  --success: #10b981;
  --error: #ef4444;
}

* { box-sizing: border-box; }

body {
  font-family: system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Arial, sans-serif;
  background: var(--bg-primary);
  color: var(--text-primary);
  line-height: 1.6;
  min-height: 100vh;
  padding: 1.25rem;
  margin: 0;
}

.container {
  width: 100%;
  max-width: none;   /* ✅ fullscreen */
  margin: 0;
}

header { margin-bottom: 1.25rem; }

h1 {
  font-size: 2rem;
  font-weight: 800;
  margin: 0 0 0.25rem 0;
  background: linear-gradient(135deg, var(--accent) 0%, var(--accent-secondary) 100%);
  -webkit-background-clip: text;
  -webkit-text-fill-color: transparent;
}

.subtitle {
  color: var(--text-secondary);
  font-size: 0.95rem;
}

.badges {
  display: flex;
  flex-wrap: wrap;
  gap: 0.5rem;
  margin-top: 0.75rem;
}

.badge {
  display: inline-flex;
  align-items: center;
  gap: 0.35rem;
  background: rgba(59, 130, 246, 0.12);
  border: 1px solid rgba(59, 130, 246, 0.22);
  color: var(--text-primary);
  padding: 0.25rem 0.55rem;
  border-radius: 999px;
  font-size: 0.8rem;
}

.grid {
  display: grid;
  grid-template-columns: repeat(12, 1fr);
  gap: 1rem;
}

.card {
  grid-column: span 12;
  background: var(--bg-secondary);
  border: 1px solid var(--border);
  border-radius: 16px;
  padding: 1rem;
  box-shadow: 0 10px 30px rgba(0,0,0,0.15);
}

.card h2 {
  font-size: 1.1rem;
  margin: 0 0 0.75rem 0;
  padding-bottom: 0.5rem;
  border-bottom: 1px solid var(--border);
}

.kv {
  width: 100%;
  border-collapse: collapse;
}

.kv td {
  padding: 0.45rem 0.5rem;
  border-bottom: 1px solid rgba(148, 163, 184, 0.12);
  vertical-align: top;
}

.kv td.key {
  width: 260px;
  color: var(--text-secondary);
}

.table-wrap {
  overflow-x: auto;
  border-radius: 12px;
  border: 1px solid rgba(148, 163, 184, 0.18);
}

table.data {
  width: 100%;
  border-collapse: collapse;
  min-width: 900px;
}

table.data th, table.data td {
  padding: 0.55rem 0.65rem;
  border-bottom: 1px solid rgba(148, 163, 184, 0.12);
  text-align: left;
  vertical-align: top;
}

table.data th {
  position: sticky;
  top: 0;
  background: var(--bg-tertiary);
  z-index: 1;
}

code, pre {
  font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace;
}

pre {
  background: var(--code-bg);
  border: 1px solid rgba(148, 163, 184, 0.18);
  border-radius: 12px;
  padding: 0.9rem;
  overflow-x: auto;
  margin: 0;
}

.muted { color: var(--text-secondary); }

.status-ok { color: var(--success); font-weight: 700; }
.status-bad { color: var(--error); font-weight: 700; }
"""

HTML_TEMPLATE = """<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8" />
  <meta name="viewport" content="width=device-width, initial-scale=1" />
  <title>{title}</title>
  <style>{css}</style>
</head>
<body>
  <div class="container">
    <header>
      <h1>{headline}</h1>
      <div class="subtitle">{subtitle}</div>
      <div class="badges">
        <span class="badge">🔌 Ports</span>
        <span class="badge">🧱 Firewall</span>
        <span class="badge">🌙 Dark mode</span>
        <span class="badge">📄 HTML default</span>
        <span class="badge">🧩 v{version}</span>
      </div>
    </header>

    <div class="grid">
      {cards}
    </div>
  </div>
</body>
</html>
"""


# ---- Helpers -----------------------------------------------------------------

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


def yyyymmdd_local() -> str:
    return dt.datetime.now().strftime("%Y%m%d")


def which(cmd: str) -> Optional[str]:
    return shutil.which(cmd)


def run_cmd(cmd: List[str], timeout_s: float = 10.0) -> Tuple[int, str, str]:
    try:
        cp = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout_s, check=False)
        return cp.returncode, cp.stdout.strip(), cp.stderr.strip()
    except Exception as exc:
        return 999, "", f"{type(exc).__name__}: {exc}"


def sudo_n_available() -> bool:
    """Return True if passwordless (non-interactive) sudo is available."""
    if not which('sudo'):
        return False
    rc, _, _ = run_cmd(['sudo', '-n', 'true'], timeout_s=2.0)
    return rc == 0


def ensure_out_dir(path: Path) -> None:
    path.mkdir(parents=True, exist_ok=True)


def escape_html(s: str) -> str:
    return (s.replace("&", "&amp;")
            .replace("<", "&lt;")
            .replace(">", "&gt;")
            .replace('"', "&quot;")
            .replace("'", "&#39;"))


def card(title: str, inner_html: str) -> str:
    return f'<section class="card"><h2>{escape_html(title)}</h2>{inner_html}</section>'


def kv_table(rows: List[Tuple[str, str]]) -> str:
    tds = []
    for k, v in rows:
        tds.append(f"<tr><td class='key'>{escape_html(k)}</td><td>{v}</td></tr>")
    return "<table class='kv'>" + "".join(tds) + "</table>"


def data_table(headers: List[str], rows: List[List[str]]) -> str:
    thead = "<tr>" + "".join(f"<th>{escape_html(h)}</th>" for h in headers) + "</tr>"
    body_rows = []
    for r in rows:
        body_rows.append("<tr>" + "".join(f"<td>{c}</td>" for c in r) + "</tr>")
    tbody = "".join(body_rows) if body_rows else "<tr><td colspan='%d' class='muted'>No data</td></tr>" % len(headers)
    return f"<div class='table-wrap'><table class='data'><thead>{thead}</thead><tbody>{tbody}</tbody></table></div>"


# ---- UFW ---------------------------------------------------------------------

def collect_ufw_status() -> Dict[str, Any]:
    if not which("ufw"):
        return {"installed": False, "status_text": "", "error": "ufw not installed"}

    # Try without sudo first (preferred; avoids password prompts entirely)
    rc, out, err = run_cmd(["ufw", "status", "verbose"], timeout_s=8)
    if rc == 0 and out:
        return {"installed": True, "status_text": out, "error": None}

    # If non-interactive sudo is available, try it. Never prompt.
    if sudo_n_available():
        rc2, out2, err2 = run_cmd(["sudo", "-n", "ufw", "status", "verbose"], timeout_s=8)
        if rc2 == 0 and out2:
            return {"installed": True, "status_text": out2, "error": None}
        msg = err2 or err or "ufw status failed"
        return {"installed": True, "status_text": out2 or out or "", "error": msg}

    # No passwordless sudo: provide a clean, actionable message (no sudo error noise)
    combined = (err + "\n" + out).strip()
    if combined:
        msg = combined.splitlines()[0]
    else:
        msg = "insufficient permissions to read UFW status (run as root or allow passwordless sudo for ufw status)"

    return {"installed": True, "status_text": out or "", "error": msg}


# ---- Docker port mapping ------------------------------------------------------

def collect_docker_port_mappings() -> Tuple[Dict[Tuple[str, str, int], Dict[str, Any]], Dict[str, Any]]:
    """
    Build a mapping for published host ports to container metadata.

    Key: (proto, host_ip, host_port) where host_ip can be '0.0.0.0', '::', or a specific IP.
    """
    docker_meta: Dict[str, Any] = {
        "available": False,
        "error": None,
        "containers_total": 0,
        "containers_with_published_ports": 0,
        "command": None,
    }
    mapping: Dict[Tuple[str, str, int], Dict[str, Any]] = {}

    if not which("docker"):
        docker_meta["error"] = "docker not installed"
        return mapping, docker_meta

    docker_meta["available"] = True
    docker_meta["command"] = "docker ps --format ..."

    # Use a predictable delimiter and include image for nicer reporting.
    rc, out, err = run_cmd(
        ["docker", "ps", "--format", "{{.Names}}||{{.ID}}||{{.Image}}||{{.Ports}}"],
        timeout_s=10,
    )
    if rc != 0:
        docker_meta["error"] = err or out or "docker ps failed"
        return mapping, docker_meta

    lines = [l for l in out.splitlines() if l.strip()]
    docker_meta["containers_total"] = len(lines)

    for line in lines:
        parts = line.split("||")
        if len(parts) < 4:
            continue
        name, cid, image, ports_raw = parts[0], parts[1], parts[2], parts[3]
        if "->" not in ports_raw:
            continue

        docker_meta["containers_with_published_ports"] += 1

        # ports_raw example: "0.0.0.0:8123->8123/tcp, :::8123->8123/tcp"
        for seg in ports_raw.split(","):
            seg = seg.strip()
            if "->" not in seg:
                continue
            left, right = seg.split("->", 1)
            # right looks like "8123/tcp"
            try:
                cport_s, proto = right.split("/", 1)
                cport = int(cport_s)
                proto = proto.strip().lower()
            except Exception:
                continue

            # left looks like "0.0.0.0:8123" or "[::]:8123" or "127.0.0.1:1234"
            host_ip = "0.0.0.0"
            host_port = None
            try:
                if left.startswith("[") and "]" in left:
                    # IPv6 form: [::]:8123
                    ip_part, port_part = left.rsplit("]:", 1)
                    host_ip = ip_part[1:]  # strip '['
                    host_port = int(port_part)
                else:
                    if ":" in left:
                        ip_part, port_part = left.rsplit(":", 1)
                        host_ip = ip_part.strip()
                        host_port = int(port_part)
            except Exception:
                continue

            if host_port is None:
                continue

            mapping[(proto, host_ip, host_port)] = {
                "docker_container_name": name,
                "docker_container_id": cid,
                "docker_image": image,
                "docker_port_spec": seg,
                "docker_container_port": cport,
            }

    return mapping, docker_meta


# ---- Port usage ---------------------------------------------------------------

def get_ip_local_port_range() -> Optional[Dict[str, int]]:
    path = Path("/proc/sys/net/ipv4/ip_local_port_range")
    try:
        txt = path.read_text(encoding="utf-8").strip()
        low_s, high_s = txt.split()
        return {"low": int(low_s), "high": int(high_s)}
    except Exception:
        return None


def collect_port_usage(
    docker_map: Optional[Dict[Tuple[str, str, int], Dict[str, Any]]] = None
) -> List[Dict[str, Any]]:
    """
    Collect all LISTEN sockets (TCP) and UDP sockets with local address.
    Returns a sorted list of records with process metadata where possible.
    """
    records: List[Dict[str, Any]] = []
    docker_map = docker_map or {}

    # psutil.net_connections(kind="inet") includes tcp + udp
    for conn in psutil.net_connections(kind="inet"):
        if not conn.laddr:
            continue

        # TCP: only LISTEN matters for conflicts
        if conn.type == pysocket.SOCK_STREAM:
            if conn.status != psutil.CONN_LISTEN:
                continue
            proto = "tcp"
        elif conn.type == pysocket.SOCK_DGRAM:
            proto = "udp"
        else:
            proto = "other"

        ip = getattr(conn.laddr, "ip", None) or conn.laddr[0]
        port = getattr(conn.laddr, "port", None) or conn.laddr[1]

        pid = conn.pid
        process_name = None
        cmdline = None
        username = None

        if pid:
            try:
                p = psutil.Process(pid)
                process_name = p.name()
                cmdline = " ".join(p.cmdline())
                username = p.username()
            except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
                pass

        record: Dict[str, Any] = {
            "proto": proto,
            "ip": ip,
            "port": int(port),
            "status": conn.status if proto != "udp" else "UDP",
            "pid": pid,
            "user": username,
            "process": process_name,
            "cmdline": cmdline,
        }

        # Attach docker metadata - try exact ip match first, then wildcard matches
        docker_meta = (
            docker_map.get((proto, ip, int(port)))
            or docker_map.get((proto, "0.0.0.0", int(port)))
            or docker_map.get((proto, "::", int(port)))
        )
        if docker_meta:
            record.update(docker_meta)

        records.append(record)

    records.sort(key=lambda r: (r.get("proto", ""), r.get("ip", ""), int(r.get("port", 0))))
    return records


# ---- Port check ---------------------------------------------------------------

def check_port_free(host: str, port: int, timeout_s: float = 0.5) -> bool:
    """
    True = free (no listener), False = in use (connect succeeded).
    """
    s = pysocket.socket(pysocket.AF_INET, pysocket.SOCK_STREAM)
    s.settimeout(timeout_s)
    try:
        s.connect((host, port))
        return False
    except Exception:
        return True
    finally:
        try:
            s.close()
        except Exception:
            pass


# ---- Report build + rendering -------------------------------------------------

def build_report(check: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    host = socket.gethostname()

    docker_map, docker_meta = collect_docker_port_mappings()
    ports = collect_port_usage(docker_map=docker_map)

    fw = collect_ufw_status()
    ip_range = get_ip_local_port_range()

    return {
        "schema": SCHEMA,
        "schema_version": SCHEMA_VERSION,
        "script_version": __version__,
        "psutil_version": getattr(psutil, "__version__", "unknown"),
        "host": host,
        "generated_at": iso_now_utc(),
        "firewall": {"ufw": fw},
        "ip_local_port_range": ip_range,
        "docker": docker_meta,
        "ports": ports,
        "check_port": check,
    }


def report_paths(out_dir: Path, host: str, with_time: bool) -> Tuple[Path, Path]:
    prefix = yyyymmdd_local()
    if with_time:
        prefix = prefix + "_" + dt.datetime.now().strftime("%H%M%S")
    base = f"{prefix}_{host}_port_fw_report"
    return out_dir / f"{base}.html", out_dir / f"{base}.json"


def render_html(report: Dict[str, Any]) -> str:
    host = report.get("host", "unknown")
    subtitle = f"Host: <b>{escape_html(host)}</b> • Generated: <b>{escape_html(report.get('generated_at',''))}</b> (UTC) • Schema: <b>{escape_html(report.get('schema_version',''))}</b>"

    cards: List[str] = []

    # Firewall first
    fw = report.get("firewall", {}).get("ufw", {})
    fw_inst = fw.get("installed", False)
    fw_err = fw.get("error")
    status_html = "<pre>" + escape_html(fw.get("status_text","") or "") + "</pre>"
    if not fw_inst:
        inner = f"<div class='muted'>UFW not installed: {escape_html(str(fw_err or ''))}</div>"
    else:
        head = "<div class='status-ok'>UFW status captured</div>" if not fw_err else f"<div class='status-bad'>UFW status partial/error: {escape_html(str(fw_err))}</div>"
        inner = head + status_html
    cards.append(card("Firewall (UFW)", inner))

    # Summary card
    docker = report.get("docker", {})
    ipr = report.get("ip_local_port_range") or {}
    ports = report.get("ports", []) or []
    summary_rows = [
        ("Listening sockets found", escape_html(str(len(ports)))),
        ("Ephemeral port range", escape_html(f"{ipr.get('low','?')}–{ipr.get('high','?')}" if ipr else "n/a")),
        ("Docker available", escape_html(str(bool(docker.get('available'))))),
        ("Docker containers total", escape_html(str(docker.get("containers_total", 0)))),
        ("Containers with published ports", escape_html(str(docker.get("containers_with_published_ports", 0)))),
        ("Docker error", escape_html(str(docker.get("error") or "")) if docker.get("error") else "<span class='muted'>none</span>"),
    ]
    cards.append(card("Summary", kv_table(summary_rows)))

    # Ports table
    rows: List[List[str]] = []
    for r in ports:
        docker_name = r.get("docker_container_name")
        docker_badge = escape_html(docker_name) if docker_name else "<span class='muted'>—</span>"
        cmd = r.get("cmdline") or ""
        # keep cmdline readable
        cmd_short = (cmd[:180] + "…") if len(cmd) > 181 else cmd
        rows.append([
            escape_html(str(r.get("proto",""))),
            escape_html(str(r.get("ip",""))),
            escape_html(str(r.get("port",""))),
            escape_html(str(r.get("status",""))),
            escape_html(str(r.get("pid") or "")),
            escape_html(str(r.get("user") or "")),
            escape_html(str(r.get("process") or "")),
            docker_badge,
            escape_html(str(r.get("docker_port_spec") or "")),
            f"<code>{escape_html(cmd_short)}</code>",
        ])
    cards.append(card("Listening Ports", data_table(
        ["Proto", "IP", "Port", "Status", "PID", "User", "Process", "Docker", "Docker port spec", "Cmdline"],
        rows
    )))

    # Port check result (optional)
    cp = report.get("check_port")
    if cp and cp.get("enabled"):
        is_free = bool(cp.get("is_free"))
        state = "<span class='status-ok'>FREE</span>" if is_free else "<span class='status-bad'>IN USE</span>"
        cp_rows = [
            ("Target", f"<code>{escape_html(cp.get('host',''))}:{escape_html(str(cp.get('port','')))}</code>"),
            ("Result", state),
            ("Timeout", escape_html(str(cp.get("timeout_s","")))),
        ]
        cards.append(card("Port Check", kv_table(cp_rows)))

    return HTML_TEMPLATE.format(
        title=f"Port & Firewall Report - {host}",
        headline="Port & Firewall Report",
        subtitle=subtitle,
        version=__version__,
        css=HTML_CSS,
        cards="\n".join(cards),
    )


# ---- CLI ---------------------------------------------------------------------

def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="Generate a full-width dark-mode Port & Firewall Report (HTML default)."
    )

    # New behaviour:
    p.add_argument(
        "--json",
        action="store_true",
        help="Additionally write a JSON report next to the HTML output.",
    )

    # Keep legacy flag (no longer required)
    p.add_argument(
        "--html",
        action="store_true",
        help="(Deprecated) Kept for backward compatibility. HTML is always generated by default.",
    )

    p.add_argument(
        "--out-dir",
        type=str,
        default="",
        help="Output directory. Default: <script_dir>/reports",
    )

    p.add_argument(
        "--with-time",
        action="store_true",
        help="Include HHMMSS in the filename to avoid overwriting same-day reports.",
    )

    p.add_argument(
        "--check-port",
        type=int,
        default=None,
        help="Check if a TCP port is free on the given host. Exit code: 0 free, 1 in use.",
    )
    p.add_argument(
        "--host",
        type=str,
        default="127.0.0.1",
        help="Host for --check-port (default: 127.0.0.1).",
    )
    p.add_argument(
        "--timeout",
        type=float,
        default=0.5,
        help="Timeout (seconds) for --check-port (default: 0.5).",
    )

    return p.parse_args()


def main() -> int:
    args = parse_args()

    # Port check (optional)
    check_block: Optional[Dict[str, Any]] = None
    exit_code = 0

    if args.check_port is not None:
        is_free = check_port_free(args.host, int(args.check_port), float(args.timeout))
        check_block = {
            "enabled": True,
            "host": args.host,
            "port": int(args.check_port),
            "timeout_s": float(args.timeout),
            "is_free": bool(is_free),
        }
        exit_code = 0 if is_free else 1

    report = build_report(check=check_block)

    script_dir = Path(__file__).resolve().parent
    out_dir = Path(args.out_dir).expanduser().resolve() if args.out_dir else (script_dir / "reports")
    ensure_out_dir(out_dir)

    host = report.get("host", socket.gethostname())
    html_path, json_path = report_paths(out_dir, host, with_time=bool(args.with_time))

    html_path.write_text(render_html(report), encoding="utf-8")
    print(f"[OK] HTML report written: {html_path}")

    if args.json:
        json_path.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8")
        print(f"[OK] JSON report written: {json_path}")

    return exit_code


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