85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
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
|