feat: add persistence and audit trails

app/policy.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from typing import Iterable, Sequence
+
+DEFAULT_BLOCKED_PORTS = frozenset({22, 23, 2375, 2376, 2377, 6443, 2377, 9000})
+
+
+class TunnelPolicy:
+    def __init__(self, whitelist: frozenset[int] | None, blocklist: frozenset[int]) -> None:
+        self.whitelist = whitelist
+        self.blocklist = blocklist
+
+    @classmethod
+    def load(cls) -> "TunnelPolicy":
+        whitelist_inputs: list[int] = []
+        blocklist_inputs: list[int] = []
+
+        env_whitelist = os.environ.get("TUNNEL_WHITELIST")
+        if env_whitelist:
+            whitelist_inputs.extend(_parse_port_sequence(env_whitelist))
+
+        env_blocklist = os.environ.get("TUNNEL_BLOCKLIST")
+        if env_blocklist:
+            blocklist_inputs.extend(_parse_port_sequence(env_blocklist))
+
+        whitelist_file = os.environ.get("TUNNEL_WHITELIST_FILE")
+        if whitelist_file:
+            whitelist_inputs.extend(_load_ports_from_file(whitelist_file, "whitelist"))
+
+        blocklist_file = os.environ.get("TUNNEL_BLOCKLIST_FILE")
+        if blocklist_file:
+            blocklist_inputs.extend(_load_ports_from_file(blocklist_file, "blocklist"))
+
+        whitelist = frozenset(whitelist_inputs) if whitelist_inputs else None
+        blocklist = frozenset(blocklist_inputs) or DEFAULT_BLOCKED_PORTS
+        return cls(whitelist=whitelist, blocklist=blocklist)
+
+    def is_port_allowed(self, port: int) -> tuple[bool, str | None]:
+        if port in self.blocklist:
+            return False, f"Porta {port} in lista sensibile/protetta"
+        if self.whitelist is not None and port not in self.whitelist:
+            return False, f"Porta {port} non presente nella whitelist"
+        return True, None
+
+
+def _parse_port_sequence(value: str) -> Sequence[int]:
+    cleaned = [item.strip() for item in value.replace(";", ",").split(",") if item.strip()]
+    ports: list[int] = []
+    for bit in cleaned:
+        try:
+            ports.append(int(bit, 10))
+        except ValueError:
+            continue
+    return ports
+
+
+def _load_ports_from_file(path: str, key: str) -> Sequence[int]:
+    try:
+        raw = Path(path).read_text(encoding="utf-8")
+    except OSError:
+        return []
+    try:
+        parsed = json.loads(raw)
+    except json.JSONDecodeError:
+        return []
+    if isinstance(parsed, dict):
+        target = parsed.get(key)
+        return _normalize_numeric_sequence(target)
+    return _normalize_numeric_sequence(parsed)
+
+
+def _normalize_numeric_sequence(value: Iterable | None) -> Sequence[int]:
+    if not value:
+        return []
+    ports: list[int] = []
+    for entry in value:
+        try:
+            ports.append(int(entry))
+        except (TypeError, ValueError):
+            continue
+    return ports