← zurück zur Toolbox
port_usage_report_v2.1.py · Quelltext
22607 Bytes · SHA-256: 6509204b19f03b7010a374a82b0679dd04ac71007462ef952800270075d4a5dc
#!/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("&", "&")
.replace("<", "<")
.replace(">", ">")
.replace('"', """)
.replace("'", "'"))
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())