From 614c882bcb731159ea921d37d59a070b0762fd97 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rodrigo=20P=C3=A9rez?= Date: Mon, 8 Jun 2026 23:28:19 -0400 Subject: [PATCH] feat(proxy): integrated UDP fan-in with inject-only MASTER (V2-P3-002) Hotspots connect on PROXY.LISTEN_PORT with in-process inject into SYSTEM, session-preserving SIGHUP reload, and composition-root wiring in main. --- adn-server.example.yaml | 22 +- src/adn_server/application/ports.py | 43 +++- src/adn_server/application/proxy/__init__.py | 4 + .../application/proxy/deployment.py | 43 ++++ .../application/proxy/session_teardown.py | 17 ++ src/adn_server/application/proxy/use_cases.py | 104 +++++--- src/adn_server/domain/proxy.py | 48 +--- .../infrastructure/config_loader.py | 3 + .../infrastructure/config_normalizer.py | 4 + .../infrastructure/config_reload.py | 48 +++- .../infrastructure/config_validator.py | 66 +++++ .../infrastructure/proxy/__init__.py | 20 ++ src/adn_server/infrastructure/proxy/config.py | 41 ++++ .../infrastructure/proxy/hbp_adapters.py | 47 ++++ .../infrastructure/proxy/ip_blacklist.py | 21 ++ .../infrastructure/proxy/reply_transport.py | 32 +++ .../infrastructure/proxy/runtime.py | 225 ++++++++++++++++++ .../infrastructure/proxy/session_executor.py | 27 +++ .../infrastructure/proxy/slot_store.py | 28 +-- .../infrastructure/proxy/udp_fanin.py | 95 ++++++++ src/adn_server/main.py | 67 +++++- tests/application/test_proxy_use_cases.py | 71 ++++-- tests/conftest.py | 31 +++ .../infrastructure/test_config_regression.py | 19 ++ tests/infrastructure/test_mqtt_config.py | 10 +- tests/infrastructure/test_proxy_config.py | 118 +++++++++ tests/infrastructure/test_proxy_reload.py | 87 +++++++ .../test_proxy_smoke_live_udp.py | 91 +++++++ tests/infrastructure/test_session_teardown.py | 35 +++ tests/infrastructure/test_udp_fanin.py | 109 +++++++++ 30 files changed, 1435 insertions(+), 141 deletions(-) create mode 100644 src/adn_server/application/proxy/deployment.py create mode 100644 src/adn_server/application/proxy/session_teardown.py create mode 100644 src/adn_server/infrastructure/proxy/config.py create mode 100644 src/adn_server/infrastructure/proxy/hbp_adapters.py create mode 100644 src/adn_server/infrastructure/proxy/ip_blacklist.py create mode 100644 src/adn_server/infrastructure/proxy/reply_transport.py create mode 100644 src/adn_server/infrastructure/proxy/runtime.py create mode 100644 src/adn_server/infrastructure/proxy/session_executor.py create mode 100644 src/adn_server/infrastructure/proxy/udp_fanin.py create mode 100644 tests/conftest.py create mode 100644 tests/infrastructure/test_config_regression.py create mode 100644 tests/infrastructure/test_proxy_config.py create mode 100644 tests/infrastructure/test_proxy_reload.py create mode 100644 tests/infrastructure/test_proxy_smoke_live_udp.py create mode 100644 tests/infrastructure/test_session_teardown.py create mode 100644 tests/infrastructure/test_udp_fanin.py diff --git a/adn-server.example.yaml b/adn-server.example.yaml index d0b4558..8ec6bde 100644 --- a/adn-server.example.yaml +++ b/adn-server.example.yaml @@ -67,20 +67,25 @@ ALIASES: CHECKSUM_FILE: file_checksums.json KEYS_FILE: keys.json -# Systems: MASTER, PEER, OPENBRIDGE. Names match legacy [SYSTEM], [D-APRS], [ECHO], [OBP-*]. -# -# SYSTEM + GENERATOR>1 expands to SYSTEM-0..SYSTEM-(GENERATOR-1), each with UDP PORT+n -# (e.g. PORT 56400 + GENERATOR 102 → listeners 56400–56501). Hotspot proxy must use the same -# PROXY.PORT and PROXY.GENERATOR in adn-monitor/proxy/adn-proxy.example.yaml. +# Integrated hotspot proxy (always enabled). Hotspots use PROXY.LISTEN_PORT only. +PROXY: + LISTEN_PORT: 62031 + LISTEN_IP: "" + TARGET_SYSTEM: SYSTEM + TIMEOUT: 30 + DEBUG: false + CLIENT_INFO: true + BLACK_LIST: [] + IP_BLACK_LIST: {} + +# SYSTEM is inject-only (no PORT / GENERATOR on proxy target — use MAX_PEERS). SYSTEMS: SYSTEM: MODE: MASTER ENABLED: true REPEAT: true - MAX_PEERS: 2 + MAX_PEERS: 102 EXPORT_AMBE: false - IP: 127.0.0.1 - PORT: 56400 PASSPHRASE: "" GROUP_HANGTIME: 5 USE_ACL: true @@ -95,7 +100,6 @@ SYSTEMS: TS2_STATIC: "" DEFAULT_REFLECTOR: 0 ANNOUNCEMENT_LANGUAGE: es_ES - GENERATOR: 102 ALLOW_UNREG_ID: false PROXY_CONTROL: false OVERRIDE_IDENT_TG: "" diff --git a/src/adn_server/application/ports.py b/src/adn_server/application/ports.py index 1390149..003c978 100644 --- a/src/adn_server/application/ports.py +++ b/src/adn_server/application/ports.py @@ -306,7 +306,7 @@ class SubscriptionStore(ABC): class ProxySlotStore(ABC): - """Hotspot session registry: peer_id ↔ upstream port (Phase 3).""" + """Hotspot session registry keyed by peer_id (Phase 3).""" @abstractmethod def bind(self, slot: "ClientSlot") -> None: @@ -325,27 +325,52 @@ class ProxySlotStore(ABC): ... @abstractmethod - def get_by_upstream(self, upstream_port: int) -> "ClientSlot | None": + def list_slots(self) -> tuple["ClientSlot", ...]: ... + +class PendingRptoQueue(ABC): + """Pending RPTO payloads for self-service / login options push.""" + @abstractmethod - def free_upstream_ports(self) -> tuple[int, ...]: + def enqueue(self, peer_id: bytes, payload: bytes) -> None: ... @abstractmethod - def list_slots(self) -> tuple["ClientSlot", ...]: + def dequeue(self) -> tuple[bytes, bytes] | None: ... -class PendingRptoQueue(ABC): - """Pending RPTO payloads for self-service / login options push.""" +class ProxyIpBlacklist(ABC): + """Temporary IP blocks (legacy proxy ``ip_black_list`` / PRBL).""" @abstractmethod - def enqueue(self, peer_id: bytes, payload: bytes) -> None: + def block_until(self, host: str, expire_at: float) -> None: ... @abstractmethod - def dequeue(self) -> tuple[bytes, bytes] | None: + def is_blocked(self, host: str, now: float) -> bool: + ... + + +class ProxyMasterSink(Protocol): + """Inject hotspot datagrams into the target MASTER (in-process).""" + + def inject(self, data: bytes, client_addr: tuple[str, int]) -> None: + ... + + +class ProxyClientSender(Protocol): + """Send datagrams to hotspot clients via LISTEN_PORT.""" + + def send_to_client(self, data: bytes, client: "ClientEndpoint") -> None: + ... + + +class MasterPeerRegistry(Protocol): + """Drop MASTER peer state when proxy session ends.""" + + def remove_peer(self, peer_id: bytes) -> None: ... @@ -365,4 +390,4 @@ class PeerTransport(Protocol): if TYPE_CHECKING: from adn_server.domain.mesh_routing import MeshEgress, MeshIngress, PeerMeshConfig - from adn_server.domain.proxy import ClientSlot + from adn_server.domain.proxy import ClientEndpoint, ClientSlot diff --git a/src/adn_server/application/proxy/__init__.py b/src/adn_server/application/proxy/__init__.py index 41d6fff..f0da312 100644 --- a/src/adn_server/application/proxy/__init__.py +++ b/src/adn_server/application/proxy/__init__.py @@ -1,10 +1,14 @@ """Proxy application layer (Phase 3).""" +from .deployment import is_proxy_inject_only, normalize_proxy_target, proxy_target_system from .packet_helpers import peer_id_from_packet from .use_cases import ProxySlotError, ProxyUseCases __all__ = [ "ProxySlotError", "ProxyUseCases", + "is_proxy_inject_only", + "normalize_proxy_target", "peer_id_from_packet", + "proxy_target_system", ] diff --git a/src/adn_server/application/proxy/deployment.py b/src/adn_server/application/proxy/deployment.py new file mode 100644 index 0000000..952381c --- /dev/null +++ b/src/adn_server/application/proxy/deployment.py @@ -0,0 +1,43 @@ +"""Proxy deployment policy from config dict (no I/O; used at startup/reload).""" + +from __future__ import annotations + +from typing import Any + + +def proxy_target_system(config: dict[str, Any]) -> str | None: + proxy = config.get("PROXY", {}) + target = proxy.get("TARGET_SYSTEM") + return str(target) if target else None + + +def config_has_enabled_master(config: dict[str, Any]) -> bool: + """True when config defines at least one enabled MASTER (adn-server, not parrot-only).""" + systems = config.get("SYSTEMS", {}) + if not isinstance(systems, dict): + return False + return any( + isinstance(cfg, dict) and cfg.get("ENABLED", True) and cfg.get("MODE") == "MASTER" + for cfg in systems.values() + ) + + +def is_proxy_inject_only(config: dict[str, Any], system_name: str) -> bool: + target = proxy_target_system(config) + return target is not None and target == system_name + + +def normalize_proxy_target(config: dict[str, Any]) -> None: + """Strip direct UDP bind fields from inject-only proxy target (D-23).""" + target = proxy_target_system(config) + if not target: + return + sys_cfg = config.get("SYSTEMS", {}).get(target) + if not isinstance(sys_cfg, dict): + return + port = sys_cfg.pop("PORT", None) + sys_cfg.pop("IP", None) + if port is not None: + sys_cfg["_REPORT_BASE_PORT"] = int(port) + else: + sys_cfg.setdefault("_REPORT_BASE_PORT", 56400) diff --git a/src/adn_server/application/proxy/session_teardown.py b/src/adn_server/application/proxy/session_teardown.py new file mode 100644 index 0000000..d73a927 --- /dev/null +++ b/src/adn_server/application/proxy/session_teardown.py @@ -0,0 +1,17 @@ +"""Wire packets for proxy session teardown (legacy reaper parity).""" + +from __future__ import annotations + +# Homebrew command prefixes (wire vocabulary; no I/O) +_MSTCL = b"MSTCL" +_RPTCL = b"RPTCL" + +CLIENT_TEARDOWN_REPEAT = 3 + + +def master_teardown_packet(peer_id: bytes) -> bytes: + return _RPTCL + peer_id + + +def client_teardown_packet() -> bytes: + return _MSTCL diff --git a/src/adn_server/application/proxy/use_cases.py b/src/adn_server/application/proxy/use_cases.py index 185b8f7..5995db4 100644 --- a/src/adn_server/application/proxy/use_cases.py +++ b/src/adn_server/application/proxy/use_cases.py @@ -1,13 +1,13 @@ -"""ProxyService use cases: client registration, upstream routing, pending RPTO (Phase 3).""" +"""ProxyService use cases: client registration, pending RPTO (Phase 3).""" from __future__ import annotations -import random +import time from collections.abc import Sequence -from adn_server.application.ports import PendingRptoQueue, ProxySlotStore +from adn_server.application.ports import PendingRptoQueue, ProxyIpBlacklist, ProxySlotStore from adn_server.domain.errors import DomainError -from adn_server.domain.proxy import ClientEndpoint, ClientSlot, PendingRpto, UpstreamPortRange +from adn_server.domain.proxy import ClientEndpoint, ClientSlot, PendingRpto, SessionTeardown from adn_server.domain.result import Fail, Result, Success from adn_server.domain.value_objects import int_id @@ -17,22 +17,34 @@ class ProxySlotError(DomainError): class ProxyUseCases: - """Register hotspot clients, map them to upstream ports, queue RPTO for the master.""" + """Register hotspot clients and queue RPTO for the master (in-process inject).""" def __init__( self, slot_store: ProxySlotStore, - port_range: UpstreamPortRange, rpto_queue: PendingRptoQueue, *, + max_peers: int = 1, black_list: Sequence[int] = (), - rng: random.Random | None = None, + ip_blacklist: ProxyIpBlacklist | None = None, ) -> None: self._slots = slot_store - self._port_range = port_range self._rpto_queue = rpto_queue + self._max_peers = max_peers self._black_list = frozenset(black_list) - self._rng = rng or random.Random() + self._ip_blacklist = ip_blacklist + + def _allocate_report_slot(self) -> int | None: + """Lowest free upstream slot index (legacy adn-proxy ``connTrack`` port pool).""" + used = { + slot.report_slot + for slot in self._slots.list_slots() + if slot.report_slot is not None + } + for index in range(self._max_peers): + if index not in used: + return index + return None def attach_client( self, @@ -43,6 +55,8 @@ class ProxyUseCases: """Bind or refresh a hotspot session (legacy ``peer_track`` on client packet).""" if len(peer_id) != 4: return Fail(ProxySlotError("peer_id must be 4 bytes")) + if self.is_ip_blocked(host): + return Fail(ProxySlotError("client IP is blacklisted")) existing = self._slots.get_by_peer(peer_id) if existing is not None: updated = existing.with_client(host, port) @@ -50,29 +64,59 @@ class ProxyUseCases: return Success(updated) if int_id(peer_id) in self._black_list: return Fail(ProxySlotError("peer is blacklisted")) - upstream_port = self._pick_upstream_port() - if upstream_port is None: - return Fail(ProxySlotError("no upstream ports available")) + if len(self._slots.list_slots()) >= self._max_peers: + return Fail(ProxySlotError("maximum peers exceeded")) + report_slot = self._allocate_report_slot() slot = ClientSlot( peer_id=peer_id, client=ClientEndpoint(host=host, port=port), - upstream_port=upstream_port, + report_slot=report_slot, ) self._slots.bind(slot) return Success(slot) def detach_client(self, peer_id: bytes) -> ClientSlot | None: - """Release session and upstream port (legacy ``reaper`` without I/O side effects).""" + """Release session (legacy ``reaper`` slot drop without I/O).""" return self._slots.unbind(peer_id) - def resolve_upstream(self, peer_id: bytes) -> int | None: - """Upstream port for forwarding client → master.""" - slot = self._slots.get_by_peer(peer_id) - return slot.upstream_port if slot else None + def expire_session(self, peer_id: bytes) -> SessionTeardown | None: + """End session and return teardown plan for infrastructure I/O (legacy ``reaper``).""" + slot = self.detach_client(peer_id) + if slot is None: + return None + return SessionTeardown(peer_id=slot.peer_id, client=slot.client) + + def is_ip_blocked(self, host: str, now: float | None = None) -> bool: + if self._ip_blacklist is None: + return False + return self._ip_blacklist.is_blocked(host, now if now is not None else time.time()) + + def block_ip_until(self, host: str, expire_at: float) -> None: + if self._ip_blacklist is not None: + self._ip_blacklist.block_until(host, expire_at) - def resolve_client(self, upstream_port: int) -> ClientEndpoint | None: - """Client endpoint for forwarding master → hotspot.""" - slot = self._slots.get_by_upstream(upstream_port) + def block_ip_from_prbl( + self, + data: bytes, + host: str, + *, + default_ttl: float = 300, + now: float | None = None, + ) -> float: + """Parse PRBL expiry and block client IP (legacy ``proxy`` PRBL handler).""" + ts = now if now is not None else time.time() + expire = ts + default_ttl + if len(data) > 8: + try: + expire = float(data[8:].decode("utf-8")) + except (ValueError, UnicodeDecodeError): + pass + self.block_ip_until(host, expire) + return expire + + def resolve_client(self, peer_id: bytes) -> ClientEndpoint | None: + """Client endpoint for a connected peer.""" + slot = self._slots.get_by_peer(peer_id) return slot.client if slot else None def schedule_rpto(self, peer_id: bytes, payload: bytes) -> bool: @@ -84,7 +128,7 @@ class ProxyUseCases: return True def next_pending_rpto(self) -> PendingRpto | None: - """Dequeue one pending RPTO with its upstream port (for master send loop).""" + """Dequeue one pending RPTO with its client endpoint (for master inject loop).""" item = self._rpto_queue.dequeue() if item is None: return None @@ -92,13 +136,17 @@ class ProxyUseCases: slot = self._slots.get_by_peer(peer_id) if slot is None: return None - return PendingRpto(peer_id=peer_id, payload=payload, upstream_port=slot.upstream_port) + return PendingRpto(peer_id=peer_id, payload=payload, client=slot.client) def list_slots(self) -> tuple[ClientSlot, ...]: return self._slots.list_slots() - def _pick_upstream_port(self) -> int | None: - free = self._slots.free_upstream_ports() - if not free: - return None - return self._rng.choice(free) + def apply_runtime_settings( + self, + *, + max_peers: int, + black_list: Sequence[int], + ) -> None: + """Hot-reload proxy limits without dropping active sessions.""" + self._max_peers = max_peers + self._black_list = frozenset(black_list) diff --git a/src/adn_server/domain/proxy.py b/src/adn_server/domain/proxy.py index 49c9458..3d12b40 100644 --- a/src/adn_server/domain/proxy.py +++ b/src/adn_server/domain/proxy.py @@ -1,4 +1,4 @@ -"""Hotspot proxy domain: client sessions and upstream port bindings (Phase 3).""" +"""Hotspot proxy domain: client sessions (Phase 3).""" from __future__ import annotations @@ -13,56 +13,34 @@ class ClientEndpoint: port: int -@dataclass(frozen=True, slots=True) -class UpstreamPortRange: - """Fan-in UDP ports on the peer server (``PORT`` .. ``PORT+GENERATOR-1``).""" - - port_start: int - port_count: int - - def __post_init__(self) -> None: - if self.port_count < 1: - raise ValueError("port_count must be >= 1") - if self.port_start < 1: - raise ValueError("port_start must be >= 1") - - @property - def port_end(self) -> int: - return self.port_start + self.port_count - 1 - - def ports(self) -> tuple[int, ...]: - return tuple(range(self.port_start, self.port_end + 1)) - - -@dataclass(frozen=True, slots=True) -class UpstreamBinding: - """Maps one fan-in listen port to the internal master HBP endpoint.""" - - listen_port: int - master_host: str - system_name: str | None = None - - @dataclass(slots=True) class ClientSlot: """Active hotspot session (legacy ``peer_track`` entry).""" peer_id: bytes client: ClientEndpoint - upstream_port: int + report_slot: int | None = None def with_client(self, host: str, port: int) -> ClientSlot: return ClientSlot( peer_id=self.peer_id, client=ClientEndpoint(host=host, port=port), - upstream_port=self.upstream_port, + report_slot=self.report_slot, ) @dataclass(frozen=True, slots=True) class PendingRpto: - """Options payload queued for delivery to the master on a peer upstream port.""" + """Options payload queued for delivery to the master on a peer session.""" peer_id: bytes payload: bytes - upstream_port: int + client: ClientEndpoint + + +@dataclass(frozen=True, slots=True) +class SessionTeardown: + """Hotspot session removed (timeout / reaper); I/O applied by infrastructure.""" + + peer_id: bytes + client: ClientEndpoint diff --git a/src/adn_server/infrastructure/config_loader.py b/src/adn_server/infrastructure/config_loader.py index fbf393c..e91959b 100644 --- a/src/adn_server/infrastructure/config_loader.py +++ b/src/adn_server/infrastructure/config_loader.py @@ -35,6 +35,7 @@ import yaml from ..domain import ID_MAX, ID_MIN, PEER_MAX from ..domain.errors import ConfigError from .config_validator import validate_config +from .proxy.config import apply_proxy_env_overrides def acl_build(acl_str: str | None, max_id: int) -> tuple[bool, list[tuple[int, int]]]: @@ -105,7 +106,9 @@ class YamlConfigLoader: "LOGGER": data.get("LOGGER", {}), "ALIASES": data.get("ALIASES", {}), "SYSTEMS": data.get("SYSTEMS", {}), + "PROXY": data.get("PROXY", {}), } + apply_proxy_env_overrides(config) # Ensure REPORT_CLIENTS is list if "REPORT_CLIENTS" in config["REPORTS"] and isinstance(config["REPORTS"]["REPORT_CLIENTS"], str): config["REPORTS"]["REPORT_CLIENTS"] = [ diff --git a/src/adn_server/infrastructure/config_normalizer.py b/src/adn_server/infrastructure/config_normalizer.py index 65388cb..52dad17 100644 --- a/src/adn_server/infrastructure/config_normalizer.py +++ b/src/adn_server/infrastructure/config_normalizer.py @@ -34,6 +34,8 @@ import time def expand_generator(config: dict, logger: logging.Logger) -> None: """Replace MASTER systems with GENERATOR > 1 by SYSTEM-0, SYSTEM-1, ... (legacy generator).""" + from adn_server.application.proxy.deployment import is_proxy_inject_only + systems = config.get("SYSTEMS", {}) to_remove: list[str] = [] new_systems: dict = {} @@ -42,6 +44,8 @@ def expand_generator(config: dict, logger: logging.Logger) -> None: continue if sys_cfg.get("MODE") != "MASTER": continue + if is_proxy_inject_only(config, system_name): + continue generator = int(sys_cfg.get("GENERATOR", 1)) if generator <= 1: continue diff --git a/src/adn_server/infrastructure/config_reload.py b/src/adn_server/infrastructure/config_reload.py index 8302b36..be280d2 100644 --- a/src/adn_server/infrastructure/config_reload.py +++ b/src/adn_server/infrastructure/config_reload.py @@ -19,6 +19,7 @@ from .config_normalizer import ( normalize_obp_config, normalize_peer_config, ) +from adn_server.application.proxy.deployment import normalize_proxy_target from .logging_config import reapply_log_level logger = logging.getLogger(__name__) @@ -63,6 +64,7 @@ def prepare_incoming_config( incoming = loader.load(config_path) apply_talker_alias_defaults(incoming) expand_generator(incoming, log) + normalize_proxy_target(incoming) ensure_system_runtime_config(incoming) normalize_peer_config(incoming) normalize_obp_config(incoming) @@ -95,7 +97,7 @@ def merge_system_config(old_cfg: dict[str, Any], new_cfg: dict[str, Any]) -> dic def merge_top_level_config(config: dict[str, Any], incoming: dict[str, Any]) -> None: """Update GLOBAL / REPORTS / ALIASES / LOGGER in the live config dict.""" kill_flag = config.get("GLOBAL", {}).get("_KILL_SERVER") - for key in ("GLOBAL", "REPORTS", "ALIASES", "LOGGER"): + for key in ("GLOBAL", "REPORTS", "ALIASES", "LOGGER", "PROXY"): if key not in incoming: continue config[key] = copy.deepcopy(incoming[key]) @@ -128,6 +130,7 @@ def reload_server_config( stop_listener: Callable[[Any], None], on_systems_changed: Callable[[], None] | None = None, on_system_removed: Callable[[str, Any], None] | None = None, + should_bind_udp: Callable[[str, dict[str, Any]], bool] | None = None, log: logging.Logger | None = None, ) -> ReloadResult: """ @@ -177,15 +180,27 @@ def reload_server_config( removed.append(name) log.info("(CONFIG-RELOAD) removed system %s", name) + def _start_listener(name: str, sys_cfg: dict[str, Any], proto: Any) -> None: + if should_bind_udp is not None and not should_bind_udp(name, sys_cfg): + protocols[name] = proto + transports.pop(name, None) + log.info("(CONFIG-RELOAD) %s inject-only (no UDP bind)", name) + return + bind = bind_spec(sys_cfg) + transports[name] = listen_udp(name, bind, proto) + protocols[name] = proto + for name in sorted(new_enabled - old_enabled): sys_cfg = copy.deepcopy(new_systems[name]) config.setdefault("SYSTEMS", {})[name] = sys_cfg proto = create_protocol(name) - bind = bind_spec(sys_cfg) - transports[name] = listen_udp(name, bind, proto) - protocols[name] = proto + _start_listener(name, sys_cfg, proto) added.append(name) - log.info("(CONFIG-RELOAD) added system %s on %s:%s", name, bind.ip, bind.port) + if should_bind_udp is None or should_bind_udp(name, sys_cfg): + bind = bind_spec(sys_cfg) + log.info("(CONFIG-RELOAD) added system %s on %s:%s", name, bind.ip, bind.port) + else: + log.info("(CONFIG-RELOAD) added inject-only system %s", name) for name in sorted(old_enabled & new_enabled): old_cfg = old_systems[name] @@ -195,15 +210,32 @@ def reload_server_config( merged = merge_system_config(old_cfg, new_cfg) config["SYSTEMS"][name] = merged proto = protocols.get(name) + inject_only = should_bind_udp is not None and not should_bind_udp(name, merged) + was_inject_only = should_bind_udp is not None and not should_bind_udp(name, old_cfg) if proto is None: proto = create_protocol(name) - transports[name] = listen_udp(name, new_bind, proto) - protocols[name] = proto + _start_listener(name, merged, proto) added.append(name) - log.info("(CONFIG-RELOAD) started missing listener %s on %s:%s", name, new_bind.ip, new_bind.port) + if inject_only: + log.info("(CONFIG-RELOAD) started missing inject-only listener %s", name) + else: + log.info("(CONFIG-RELOAD) started missing listener %s on %s:%s", name, new_bind.ip, new_bind.port) continue if hasattr(proto, "apply_system_config"): proto.apply_system_config(config) + if inject_only: + port = transports.pop(name, None) + if port is not None: + stop_listener(port) + protocols[name] = proto + updated.append(name) + log.debug("(CONFIG-RELOAD) %s inject-only (bind skipped)", name) + continue + if was_inject_only and not inject_only: + _start_listener(name, merged, proto) + rebound.append(name) + log.info("(CONFIG-RELOAD) started UDP bind for %s on %s:%s", name, new_bind.ip, new_bind.port) + continue if old_bind != new_bind: port = transports.get(name) if port is not None: diff --git a/src/adn_server/infrastructure/config_validator.py b/src/adn_server/infrastructure/config_validator.py index 9f1aa03..b7ef39d 100644 --- a/src/adn_server/infrastructure/config_validator.py +++ b/src/adn_server/infrastructure/config_validator.py @@ -199,6 +199,69 @@ def _validate_logger(logger_cfg: dict[str, Any], errors: list[str]) -> None: _section_string_keys("LOGGER", logger_cfg, LOGGER_STRING_KEYS, errors) +def _validate_proxy(proxy_cfg: dict[str, Any] | None, systems: dict[str, Any], errors: list[str]) -> None: + from adn_server.application.proxy.deployment import config_has_enabled_master + + if not isinstance(systems, dict) or not config_has_enabled_master({"SYSTEMS": systems}): + return + if not proxy_cfg or not isinstance(proxy_cfg, dict): + errors.append("PROXY: required when config has enabled MASTER systems (adn-server).") + return + for key in ("DEBUG", "CLIENT_INFO", "STATS"): + if key in proxy_cfg: + _expect_bool(f"PROXY.{key}", proxy_cfg[key], errors) + if "LISTEN_PORT" in proxy_cfg: + _expect_int("PROXY.LISTEN_PORT", proxy_cfg["LISTEN_PORT"], errors) + if "TIMEOUT" in proxy_cfg: + _expect_number("PROXY.TIMEOUT", proxy_cfg["TIMEOUT"], errors) + if "TARGET_SYSTEM" in proxy_cfg and not _is_empty(proxy_cfg["TARGET_SYSTEM"]): + _expect_str("PROXY.TARGET_SYSTEM", proxy_cfg["TARGET_SYSTEM"], errors) + if "LISTEN_IP" in proxy_cfg: + _expect_str("PROXY.LISTEN_IP", proxy_cfg["LISTEN_IP"], errors) + if "BLACK_LIST" in proxy_cfg and not isinstance(proxy_cfg["BLACK_LIST"], list): + errors.append( + f"PROXY.BLACK_LIST: expected list, got {type(proxy_cfg['BLACK_LIST']).__name__}." + ) + if "IP_BLACK_LIST" in proxy_cfg and not isinstance(proxy_cfg["IP_BLACK_LIST"], dict): + errors.append( + f"PROXY.IP_BLACK_LIST: expected mapping, got {type(proxy_cfg['IP_BLACK_LIST']).__name__}." + ) + for key in ("DISPATCH", "ENABLED", "PORT", "GENERATOR", "MASTER", "MAX_PROXY_SESSIONS", "udp_pool"): + if key in proxy_cfg: + errors.append(f"PROXY.{key}: removed in v2; integrated proxy is always enabled.") + + listen_port = proxy_cfg.get("LISTEN_PORT", 62031) + if isinstance(listen_port, bool) or not isinstance(listen_port, int) or listen_port < 1: + errors.append("PROXY.LISTEN_PORT: required >= 1.") + + target = proxy_cfg.get("TARGET_SYSTEM") + if _is_empty(target): + errors.append("PROXY.TARGET_SYSTEM: required.") + return + if not isinstance(systems, dict) or target not in systems: + errors.append(f"PROXY.TARGET_SYSTEM: unknown system {target!r}.") + return + target_cfg = systems[target] + if not isinstance(target_cfg, dict): + errors.append(f"SYSTEMS.{target}: expected mapping.") + return + if not target_cfg.get("ENABLED", True): + errors.append(f"PROXY.TARGET_SYSTEM: SYSTEMS.{target} must be ENABLED.") + if target_cfg.get("MODE") != "MASTER": + errors.append(f"PROXY.TARGET_SYSTEM: SYSTEMS.{target} must be MODE MASTER.") + + port = target_cfg.get("PORT", 0) + if not _is_empty(port) and int(port) > 0: + errors.append( + f"SYSTEMS.{target}.PORT: must be omitted or 0 for inject-only proxy target (D-23)." + ) + generator = int(target_cfg.get("GENERATOR", 1) or 1) + if generator > 1: + errors.append( + f"SYSTEMS.{target}.GENERATOR: must be 0 or 1 on proxy target (use MAX_PEERS, not GENERATOR)." + ) + + def _validate_system(name: str, sys_cfg: dict[str, Any], errors: list[str]) -> None: prefix = f"SYSTEMS.{name}" _section_string_keys(prefix, sys_cfg, SYSTEM_STRING_KEYS, errors) @@ -254,6 +317,9 @@ def validate_config(config: dict[str, Any], *, config_path: str | None = None) - continue _validate_system(name, sys_cfg, errors) + proxy_cfg = config.get("PROXY") + _validate_proxy(proxy_cfg if isinstance(proxy_cfg, dict) else None, systems if isinstance(systems, dict) else {}, errors) + if errors: header = f"Configuration error in {config_path}:" if config_path else "Configuration error:" raise ConfigError("\n".join([header, *[f" - {err}" for err in errors]])) diff --git a/src/adn_server/infrastructure/proxy/__init__.py b/src/adn_server/infrastructure/proxy/__init__.py index 8ee21ed..d6ef51f 100644 --- a/src/adn_server/infrastructure/proxy/__init__.py +++ b/src/adn_server/infrastructure/proxy/__init__.py @@ -1,9 +1,29 @@ """Proxy infrastructure adapters (Phase 3).""" +from .config import apply_proxy_env_overrides, proxy_settings +from .hbp_adapters import FanInClientSender, HbpMasterPeerRegistry, InProcessHbpSink +from .ip_blacklist import InMemoryProxyIpBlacklist +from .reply_transport import ProxyReplyTransport +from .runtime import ProxyServiceState, apply_proxy_config_reload, start_proxy_service from .rpto_queue import InMemoryPendingRptoQueue +from .session_executor import apply_session_teardown from .slot_store import InMemoryProxySlotStore +from .udp_fanin import ProxyFanInProtocol, listen_proxy_fanin __all__ = [ + "FanInClientSender", + "HbpMasterPeerRegistry", "InMemoryPendingRptoQueue", + "InMemoryProxyIpBlacklist", "InMemoryProxySlotStore", + "InProcessHbpSink", + "ProxyFanInProtocol", + "ProxyReplyTransport", + "ProxyServiceState", + "apply_proxy_config_reload", + "apply_proxy_env_overrides", + "apply_session_teardown", + "listen_proxy_fanin", + "proxy_settings", + "start_proxy_service", ] diff --git a/src/adn_server/infrastructure/proxy/config.py b/src/adn_server/infrastructure/proxy/config.py new file mode 100644 index 0000000..1f2621a --- /dev/null +++ b/src/adn_server/infrastructure/proxy/config.py @@ -0,0 +1,41 @@ +"""PROXY runtime settings from config dict (infrastructure; no business rules).""" + +from __future__ import annotations + +from typing import Any + + +def apply_proxy_env_overrides(config: dict[str, Any]) -> None: + """Apply ADN_PROXY_* environment overrides (design §5.6).""" + import os + + proxy = config.setdefault("PROXY", {}) + if os.environ.get("ADN_PROXY_DEBUG", "").strip() in ("1", "true", "TRUE", "yes", "YES"): + proxy["DEBUG"] = True + listen_port = os.environ.get("ADN_PROXY_LISTENPORT", "").strip() + if listen_port: + proxy["LISTEN_PORT"] = int(listen_port) + if os.environ.get("ADN_PROXY_IPV6", "").strip() in ("1", "true", "TRUE", "yes", "YES"): + proxy["LISTEN_IP"] = "::" + + +def proxy_settings(config: dict[str, Any]) -> dict[str, Any]: + """Resolved PROXY runtime settings with defaults.""" + proxy = config.get("PROXY", {}) + black_list = proxy.get("BLACK_LIST") or [] + if not isinstance(black_list, list): + black_list = [] + ip_black_list = proxy.get("IP_BLACK_LIST") or {} + if not isinstance(ip_black_list, dict): + ip_black_list = {} + return { + "listen_port": int(proxy.get("LISTEN_PORT", 62031)), + "listen_ip": str(proxy.get("LISTEN_IP") or ""), + "target_system": str(proxy.get("TARGET_SYSTEM") or ""), + "timeout": float(proxy.get("TIMEOUT", 30)), + "debug": bool(proxy.get("DEBUG")), + "client_info": bool(proxy.get("CLIENT_INFO", True)), + "black_list": tuple(int(x) for x in black_list), + "ip_black_list": {str(k): float(v) for k, v in ip_black_list.items()}, + "stats": bool(proxy.get("STATS")), + } diff --git a/src/adn_server/infrastructure/proxy/hbp_adapters.py b/src/adn_server/infrastructure/proxy/hbp_adapters.py new file mode 100644 index 0000000..d74ee89 --- /dev/null +++ b/src/adn_server/infrastructure/proxy/hbp_adapters.py @@ -0,0 +1,47 @@ +"""HBP adapters for proxy application ports.""" + +from __future__ import annotations + +from typing import Any, Protocol + +from adn_server.application.ports import MasterPeerRegistry, ProxyClientSender, ProxyMasterSink +from adn_server.domain.proxy import ClientEndpoint + + +class _MasterHbpReceiver(Protocol): + def _master_datagram_received(self, data: bytes, sockaddr: tuple[str, int]) -> None: + ... + + +class InProcessHbpSink(ProxyMasterSink): + """Deliver client datagrams to the target MASTER without a UDP hop.""" + + def __init__(self, hbp: _MasterHbpReceiver) -> None: + self._hbp = hbp + + def inject(self, data: bytes, client_addr: tuple[str, int]) -> None: + self._hbp._master_datagram_received(data, client_addr) + + +class FanInClientSender(ProxyClientSender): + """Send to hotspots through the fan-in UDP transport.""" + + def __init__(self, transport: Any) -> None: + self._transport = transport + + def send_to_client(self, data: bytes, client: ClientEndpoint) -> None: + if self._transport is None: + return + self._transport.write(data, (client.host, client.port)) + + +class HbpMasterPeerRegistry(MasterPeerRegistry): + """Remove timed-out peers from MASTER ``_peers``.""" + + def __init__(self, hbp: Any) -> None: + self._hbp = hbp + + def remove_peer(self, peer_id: bytes) -> None: + peers = getattr(self._hbp, "_peers", None) + if isinstance(peers, dict): + peers.pop(peer_id, None) diff --git a/src/adn_server/infrastructure/proxy/ip_blacklist.py b/src/adn_server/infrastructure/proxy/ip_blacklist.py new file mode 100644 index 0000000..80f0c24 --- /dev/null +++ b/src/adn_server/infrastructure/proxy/ip_blacklist.py @@ -0,0 +1,21 @@ +"""In-memory IP blacklist for proxy fan-in.""" + +from __future__ import annotations + +from adn_server.application.ports import ProxyIpBlacklist + + +class InMemoryProxyIpBlacklist(ProxyIpBlacklist): + def __init__(self, initial: dict[str, float] | None = None) -> None: + self._entries: dict[str, float] = dict(initial or {}) + + def block_until(self, host: str, expire_at: float) -> None: + self._entries[host] = expire_at + + def is_blocked(self, host: str, now: float) -> bool: + expire = self._entries.get(host) + return expire is not None and now < expire + + def merge_static_entries(self, entries: dict[str, float]) -> None: + """Apply config IP_BLACK_LIST entries (runtime PRBL blocks are kept).""" + self._entries.update(entries) diff --git a/src/adn_server/infrastructure/proxy/reply_transport.py b/src/adn_server/infrastructure/proxy/reply_transport.py new file mode 100644 index 0000000..def7ad9 --- /dev/null +++ b/src/adn_server/infrastructure/proxy/reply_transport.py @@ -0,0 +1,32 @@ +"""Route MASTER HBP replies through the proxy fan-in UDP socket.""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Protocol + +from adn_server.infrastructure.hbp_constants import PRBL + + +class _DatagramWriter(Protocol): + def write(self, data: bytes, addr: tuple[str, int]) -> None: + ... + + +class ProxyReplyTransport: + """Wrap the fan-in transport so MASTER replies leave via LISTEN_PORT.""" + + def __init__( + self, + fanin_transport: _DatagramWriter, + *, + prbl_handler: Callable[[bytes, tuple[str, int]], None] | None = None, + ) -> None: + self._fanin = fanin_transport + self._prbl_handler = prbl_handler + + def write(self, data: bytes, addr: tuple[str, int]) -> None: + if len(data) >= 4 and data[:4] == PRBL and self._prbl_handler is not None: + self._prbl_handler(data, addr) + return + self._fanin.write(data, addr) diff --git a/src/adn_server/infrastructure/proxy/runtime.py b/src/adn_server/infrastructure/proxy/runtime.py new file mode 100644 index 0000000..240e694 --- /dev/null +++ b/src/adn_server/infrastructure/proxy/runtime.py @@ -0,0 +1,225 @@ +"""Wire integrated hotspot proxy at startup (composition root wiring).""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from typing import Any + +from twisted.internet import reactor +from twisted.internet.interfaces import IDelayedCall + +from adn_server.application.proxy import ProxyUseCases +from adn_server.application.ports import ProxyClientSender, ProxyMasterSink +from adn_server.domain.value_objects import int_id +from adn_server.infrastructure.proxy.config import proxy_settings +from adn_server.infrastructure.proxy.hbp_adapters import ( + FanInClientSender, + HbpMasterPeerRegistry, + InProcessHbpSink, +) +from adn_server.infrastructure.proxy.ip_blacklist import InMemoryProxyIpBlacklist +from adn_server.infrastructure.proxy.rpto_queue import InMemoryPendingRptoQueue +from adn_server.infrastructure.proxy.session_executor import apply_session_teardown +from adn_server.infrastructure.proxy.slot_store import InMemoryProxySlotStore +from adn_server.infrastructure.proxy.udp_fanin import ProxyFanInProtocol, listen_proxy_fanin +from adn_server.infrastructure.proxy.reply_transport import ProxyReplyTransport + + +def _proxy_runtime_snapshot(config: dict[str, Any]) -> dict[str, Any]: + settings = proxy_settings(config) + target = settings["target_system"] + max_peers = int(config.get("SYSTEMS", {}).get(target, {}).get("MAX_PEERS", 1)) + return {**settings, "max_peers": max_peers} + + +@dataclass +class ProxyServiceState: + """Live proxy handles (for shutdown / reload).""" + + target_system: str + use_cases: ProxyUseCases + master_sink: ProxyMasterSink + client_sender: ProxyClientSender + fanin: ProxyFanInProtocol + udp_port: Any + listen_port: int = 62031 + listen_ip: str = "" + _timers: dict[bytes, IDelayedCall] = field(default_factory=dict) + _runtime: dict[str, Any] = field(default_factory=dict) + + def stop(self) -> Any: + """Stop timers and UDP listener. Returns Twisted Deferred when a port was bound.""" + for call in self._timers.values(): + if call.active(): + call.cancel() + self._timers.clear() + if self.udp_port is None: + return None + port = self.udp_port + self.udp_port = None + return port.stopListening() + + +def apply_proxy_config_reload( + state: ProxyServiceState, + config: dict[str, Any], + *, + logger: logging.Logger, +) -> None: + """Hot-apply PROXY settings on SIGHUP without closing LISTEN_PORT or dropping sessions.""" + incoming = _proxy_runtime_snapshot(config) + bind_changed = ( + state.listen_port != incoming["listen_port"] + or state.listen_ip != incoming["listen_ip"] + ) + target_changed = state.target_system != incoming["target_system"] + state._runtime.update(incoming) + state.use_cases.apply_runtime_settings( + max_peers=incoming["max_peers"], + black_list=incoming["black_list"], + ) + ip_bl = state.use_cases._ip_blacklist # noqa: SLF001 + if isinstance(ip_bl, InMemoryProxyIpBlacklist): + ip_bl.merge_static_entries(incoming["ip_black_list"]) + state.fanin.debug = bool(incoming["debug"]) + if bind_changed: + logger.warning( + "(CONFIG-RELOAD) PROXY bind change ignored at runtime " + "(still listening on %s:%s); restart adn-server to apply %s:%s", + state.listen_ip or "*", + state.listen_port, + incoming["listen_ip"] or "*", + incoming["listen_port"], + ) + if target_changed: + logger.warning( + "(CONFIG-RELOAD) PROXY TARGET_SYSTEM change ignored at runtime " + "(still injecting into %s); restart adn-server to apply %s", + state.target_system, + incoming["target_system"], + ) + logger.debug( + "(CONFIG-RELOAD) proxy settings applied (%s active session(s), port kept open)", + len(state.use_cases.list_slots()), + ) + + +def start_proxy_service( + config: dict[str, Any], + protocols: dict[str, Any], + *, + logger: logging.Logger, +) -> ProxyServiceState: + """Start LISTEN_PORT fan-in and inject into ``PROXY.TARGET_SYSTEM`` MASTER.""" + runtime = _proxy_runtime_snapshot(config) + target = runtime["target_system"] + target_proto = protocols.get(target) + if target_proto is None: + raise RuntimeError(f"(PROXY) TARGET_SYSTEM {target!r} has no HBP protocol instance") + + ip_blacklist = InMemoryProxyIpBlacklist(runtime["ip_black_list"]) + use_cases = ProxyUseCases( + InMemoryProxySlotStore(), + InMemoryPendingRptoQueue(), + max_peers=runtime["max_peers"], + black_list=runtime["black_list"], + ip_blacklist=ip_blacklist, + ) + master_sink = InProcessHbpSink(target_proto) + peer_registry = HbpMasterPeerRegistry(target_proto) + fanin = ProxyFanInProtocol( + use_cases, + master_sink, + debug=runtime["debug"], + logger=logger, + ) + state = ProxyServiceState( + target_system=target, + use_cases=use_cases, + master_sink=master_sink, + client_sender=FanInClientSender(None), + fanin=fanin, + udp_port=None, + listen_port=runtime["listen_port"], + listen_ip=runtime["listen_ip"], + _runtime=dict(runtime), + ) + + def _on_client_attached(peer_id: bytes, host: str, port: int, new_session: bool) -> None: + rt = state._runtime + if ( + new_session + and rt["client_info"] + and peer_id != b"\xff\xff\xff\xff" + ): + logger.info( + "(PROXY) New client: ID:%s IP:%s Port:%s", + str(int_id(peer_id)).rjust(9), + host.rjust(15), + port, + ) + existing = state._timers.get(peer_id) + if existing is not None and existing.active(): + existing.reset(rt["timeout"]) + return + if existing is not None: + existing.cancel() + state._timers[peer_id] = reactor.callLater(rt["timeout"], _reap_session, peer_id) + + def _reap_session(peer_id: bytes) -> None: + rt = state._runtime + state._timers.pop(peer_id, None) + teardown = use_cases.expire_session(peer_id) + if teardown is None: + return + if rt["debug"]: + logger.debug( + "(PROXY) session timeout peer=%s client=%s:%s", + int_id(peer_id), + teardown.client.host, + teardown.client.port, + ) + if rt["client_info"] and peer_id != b"\xff\xff\xff\xff": + logger.info( + "(PROXY) Client: ID:%s IP:%s Port:%s Removed.", + str(int_id(peer_id)).rjust(9), + teardown.client.host.rjust(15), + teardown.client.port, + ) + apply_session_teardown( + teardown, + master_sink=master_sink, + client_sender=state.client_sender, + peer_registry=peer_registry, + ) + + def _handle_prbl(data: bytes, addr: tuple[str, int]) -> None: + expire = use_cases.block_ip_from_prbl(data, addr[0]) + if state._runtime["client_info"]: + logger.info("(PROXY) Add to blacklist: host %s expire %s", addr[0], expire) + + fanin._on_attached = _on_client_attached # noqa: SLF001 + fanin_proto, udp_port = listen_proxy_fanin( + reactor, + runtime["listen_ip"], + runtime["listen_port"], + use_cases, + master_sink, + debug=runtime["debug"], + logger=logger, + protocol=fanin, + ) + state.udp_port = udp_port + state.client_sender = FanInClientSender(fanin_proto.transport) + target_proto.transport = ProxyReplyTransport(fanin_proto.transport, prbl_handler=_handle_prbl) + + logger.info( + "(PROXY) Hotspot fan-in on %s:%s → inject %s (MAX_PEERS=%s, TIMEOUT=%ss)", + runtime["listen_ip"] or "*", + runtime["listen_port"], + target, + runtime["max_peers"], + runtime["timeout"], + ) + return state diff --git a/src/adn_server/infrastructure/proxy/session_executor.py b/src/adn_server/infrastructure/proxy/session_executor.py new file mode 100644 index 0000000..95ba0af --- /dev/null +++ b/src/adn_server/infrastructure/proxy/session_executor.py @@ -0,0 +1,27 @@ +"""Execute proxy session teardown via application ports (infrastructure).""" + +from __future__ import annotations + +from adn_server.application.ports import MasterPeerRegistry, ProxyClientSender, ProxyMasterSink +from adn_server.application.proxy.session_teardown import ( + CLIENT_TEARDOWN_REPEAT, + client_teardown_packet, + master_teardown_packet, +) +from adn_server.domain.proxy import SessionTeardown + + +def apply_session_teardown( + teardown: SessionTeardown, + *, + master_sink: ProxyMasterSink, + client_sender: ProxyClientSender, + peer_registry: MasterPeerRegistry, +) -> None: + """Legacy reaper: RPTCL inject, MSTCL×3 to client, drop MASTER peer.""" + client_addr = (teardown.client.host, teardown.client.port) + master_sink.inject(master_teardown_packet(teardown.peer_id), client_addr) + pkt = client_teardown_packet() + for _ in range(CLIENT_TEARDOWN_REPEAT): + client_sender.send_to_client(pkt, teardown.client) + peer_registry.remove_peer(teardown.peer_id) diff --git a/src/adn_server/infrastructure/proxy/slot_store.py b/src/adn_server/infrastructure/proxy/slot_store.py index e33ff61..91da28a 100644 --- a/src/adn_server/infrastructure/proxy/slot_store.py +++ b/src/adn_server/infrastructure/proxy/slot_store.py @@ -1,26 +1,21 @@ -"""In-memory proxy slot store (Phase 3; not wired to UDP yet).""" +"""In-memory proxy slot store (Phase 3).""" from __future__ import annotations from adn_server.application.ports import ProxySlotStore -from adn_server.domain.proxy import ClientSlot, UpstreamPortRange +from adn_server.domain.proxy import ClientSlot class InMemoryProxySlotStore(ProxySlotStore): - """Track ``ClientSlot`` by peer_id and upstream port occupancy.""" + """Track ``ClientSlot`` by peer_id.""" - def __init__(self, port_range: UpstreamPortRange) -> None: - self._port_range = port_range + def __init__(self) -> None: self._by_peer: dict[bytes, ClientSlot] = {} - self._by_upstream: dict[int, bytes] = {} def bind(self, slot: ClientSlot) -> None: - if slot.upstream_port in self._by_upstream: - raise ValueError(f"upstream port {slot.upstream_port} already bound") if slot.peer_id in self._by_peer: raise ValueError("peer_id already bound") self._by_peer[slot.peer_id] = slot - self._by_upstream[slot.upstream_port] = slot.peer_id def update_client(self, peer_id: bytes, host: str, port: int) -> None: slot = self._by_peer.get(peer_id) @@ -29,23 +24,10 @@ class InMemoryProxySlotStore(ProxySlotStore): self._by_peer[peer_id] = slot.with_client(host, port) def unbind(self, peer_id: bytes) -> ClientSlot | None: - slot = self._by_peer.pop(peer_id, None) - if slot is None: - return None - self._by_upstream.pop(slot.upstream_port, None) - return slot + return self._by_peer.pop(peer_id, None) def get_by_peer(self, peer_id: bytes) -> ClientSlot | None: return self._by_peer.get(peer_id) - def get_by_upstream(self, upstream_port: int) -> ClientSlot | None: - peer_id = self._by_upstream.get(upstream_port) - if peer_id is None: - return None - return self._by_peer.get(peer_id) - - def free_upstream_ports(self) -> tuple[int, ...]: - return tuple(p for p in self._port_range.ports() if p not in self._by_upstream) - def list_slots(self) -> tuple[ClientSlot, ...]: return tuple(self._by_peer.values()) diff --git a/src/adn_server/infrastructure/proxy/udp_fanin.py b/src/adn_server/infrastructure/proxy/udp_fanin.py new file mode 100644 index 0000000..4d6a47a --- /dev/null +++ b/src/adn_server/infrastructure/proxy/udp_fanin.py @@ -0,0 +1,95 @@ +"""UDP fan-in: hotspot LISTEN_PORT with in-process inject (Phase 3).""" + +from __future__ import annotations + +import logging +from collections.abc import Callable +from typing import Any, Protocol + +from twisted.internet.protocol import DatagramProtocol + +from adn_server.application.ports import ProxyMasterSink +from adn_server.application.proxy import ProxyUseCases, peer_id_from_packet +from adn_server.domain.result import is_fail + +_logger = logging.getLogger(__name__) + + +class _DatagramWriter(Protocol): + def write(self, data: bytes, addr: tuple[str, int]) -> None: + ... + + +class ProxyFanInProtocol(DatagramProtocol): + """UDP listener for hotspots; attaches sessions and injects into the target MASTER.""" + + def __init__( + self, + proxy: ProxyUseCases, + master_sink: ProxyMasterSink, + *, + debug: bool = False, + logger: logging.Logger | None = None, + on_attached: Callable[[bytes, str, int, bool], None] | None = None, + ) -> None: + self._proxy = proxy + self._master_sink = master_sink + self.debug = debug + self._log = logger or _logger + self._on_attached = on_attached + + def datagramReceived(self, data: bytes, addr: tuple[str, int]) -> None: + host, port = addr + if self._proxy.is_ip_blocked(host): + if self.debug: + self._log.debug("(PROXY) dropped packet from blacklisted IP %s:%s", host, port) + return + command = data[:4] if len(data) >= 4 else b"" + if self.debug: + self._log.debug( + "(PROXY) RX from %s:%s len=%d cmd=%r", + host, + port, + len(data), + command, + ) + peer_id = peer_id_from_packet(data, from_master=False) + if peer_id is None: + if self.debug: + self._log.debug("(PROXY) ignored packet with no peer_id from %s:%s", host, port) + return + new_session = self._proxy.resolve_client(peer_id) is None + result = self._proxy.attach_client(peer_id, host, port) + if is_fail(result): + if self.debug: + self._log.debug( + "(PROXY) attach rejected peer=%s from %s:%s: %s", + peer_id.hex(), + host, + port, + result.error, + ) + return + if self._on_attached is not None: + self._on_attached(peer_id, host, port, new_session) + self._master_sink.inject(data, addr) + + +def listen_proxy_fanin( + reactor: Any, + listen_ip: str, + listen_port: int, + proxy: ProxyUseCases, + master_sink: ProxyMasterSink, + *, + debug: bool = False, + logger: logging.Logger | None = None, + protocol: ProxyFanInProtocol | None = None, +) -> tuple[ProxyFanInProtocol, Any]: + """Bind LISTEN_PORT and return ``(protocol, udp_port)``.""" + fanin = protocol or ProxyFanInProtocol(proxy, master_sink, debug=debug, logger=logger) + udp_port = reactor.listenUDP(listen_port, fanin, interface=listen_ip or "0.0.0.0") + return fanin, udp_port + + +__all__ = ["ProxyFanInProtocol", "listen_proxy_fanin"] diff --git a/src/adn_server/main.py b/src/adn_server/main.py index 55545a2..9faa3dc 100644 --- a/src/adn_server/main.py +++ b/src/adn_server/main.py @@ -81,6 +81,8 @@ from .infrastructure.config_normalizer import ( normalize_peer_config as _normalize_peer_config, ) from .infrastructure.config_reload import BindSpec, reload_server_config +from .application.proxy.deployment import is_proxy_inject_only, normalize_proxy_target, proxy_target_system +from .infrastructure.proxy import apply_proxy_config_reload, start_proxy_service from .domain.dmr.bptc import encode_emblc from .infrastructure.persistence import PickleSubMapStore from .infrastructure.persistence.alias_loader import DefaultAliasLoader @@ -124,6 +126,28 @@ class ReportSenderAdapter(ReportSender): def send_bridge_event(self, event: str) -> None: self._factory.send_bridge_event(event) + def set_peer_slot_map(self, provider) -> None: + self._factory.set_peer_slot_map(provider) + + +def _wire_proxy_report_slots( + report_factory: ReportServerFactory, + proxy_state: Any, +) -> None: + """Bind proxy upstream slot indices into monitor topology expansion.""" + if proxy_state is None: + report_factory.set_peer_slot_map(None) + return + + def _slot_map() -> dict[bytes, int]: + return { + slot.peer_id: slot.report_slot + for slot in proxy_state.use_cases.list_slots() + if slot.report_slot is not None + } + + report_factory.set_peer_slot_map(_slot_map) + def _make_echo_bridges(config: dict) -> dict: """Initial BRIDGES for ECHO system (legacy make_bridges 9990 + MASTER expansion). @@ -243,6 +267,7 @@ def main() -> None: # Generator: expand MASTER systems with GENERATOR > 1 into SYSTEM-0, SYSTEM-1, ... (legacy) _expand_generator(config, logger) + normalize_proxy_target(config) _ensure_system_runtime_config(config) _normalize_peer_config(config) _normalize_obp_config(config) @@ -524,7 +549,8 @@ def main() -> None: on_options_received=bridge_use_cases.options_config_for_system, on_deactivate_dynamic_bridges=bridge_use_cases.deactivate_all_dynamic_bridges, on_obp_bcsq_received=bridge_use_cases.on_obp_bcsq_received, - on_talker_alias_local_repeat=bridge_use_cases.send_talker_alias_local_repeat, + on_talker_alias_repeat_prepare=bridge_use_cases.prepare_talker_alias_local_repeat, + on_talker_alias_repeat_burst=bridge_use_cases.rewrite_repeat_voice_burst, on_talker_alias_stream_end=bridge_use_cases.clear_talker_alias_stream, on_dmra_fragment_stored=bridge_use_cases.on_dmra_fragment_stored, ) @@ -538,13 +564,19 @@ def main() -> None: if port is not None: port.stopListening() + # UDP / proxy listeners (proxy_state declared before reload handler uses nonlocal) + proxy_state = None + + def _should_bind_udp(system_name: str, sys_cfg: dict[str, Any]) -> bool: + return not is_proxy_inject_only(config, system_name) + def _on_config_systems_changed() -> None: reporting_use_cases.send_config(config.get("SYSTEMS", {})) reporting_use_cases.send_bridge(bridge_router.get_bridges()) user_passwords_loader.load(config) def _do_config_reload() -> None: - nonlocal report_mqtt + nonlocal report_mqtt, proxy_state mqtt_before = mqtt_settings_from_config(config) new_config = prepare_reload_config(runtime_holder) try: @@ -559,9 +591,11 @@ def main() -> None: stop_listener=_stop_udp_port, on_systems_changed=None, on_system_removed=bridge_use_cases.flush_monitor_events_for_system, + should_bind_udp=_should_bind_udp, log=logger, ) swap_runtime_config(runtime_holder, new_config, config_path=config_path) + normalize_proxy_target(config) report_factory.set_config(config) mqtt_after = mqtt_settings_from_config(config) report_mqtt = reconcile_mqtt_publisher( @@ -571,6 +605,14 @@ def main() -> None: mqtt_after, report_enabled=config.get("REPORTS", {}).get("REPORT", True), ) + if proxy_state is not None: + apply_proxy_config_reload(proxy_state, config, logger=logger) + _wire_proxy_report_slots(report_factory, proxy_state) + elif proxy_target_system(config): + proxy_state = start_proxy_service(config, protocols, logger=logger) + _wire_proxy_report_slots(report_factory, proxy_state) + else: + _wire_proxy_report_slots(report_factory, None) if result.added or result.removed or result.updated or result.rebound: _on_config_systems_changed() except Exception as e: @@ -621,10 +663,13 @@ def main() -> None: for system_name, sys_cfg in systems_cfg.items(): if not sys_cfg.get("ENABLED", True): continue - bind = BindSpec(ip=str(sys_cfg.get("IP") or "0.0.0.0"), port=int(sys_cfg.get("PORT", 56400))) protocol = _create_hbp_protocol(system_name) - udp_ports[system_name] = _listen_system(system_name, bind, protocol) protocols[system_name] = protocol + if not _should_bind_udp(system_name, sys_cfg): + logger.info("(PROXY) %s inject-only (no UDP bind)", system_name) + continue + bind = BindSpec(ip=str(sys_cfg.get("IP") or "0.0.0.0"), port=int(sys_cfg.get("PORT", 56400))) + udp_ports[system_name] = _listen_system(system_name, bind, protocol) logger.debug( "(GLOBAL) %s instance created: %s, %s", sys_cfg.get("MODE", "?"), @@ -632,6 +677,20 @@ def main() -> None: protocol, ) + if proxy_target_system(config): + try: + proxy_state = start_proxy_service(config, protocols, logger=logger) + except Exception as exc: + logger.error("(PROXY) failed to start integrated proxy: %s", exc) + raise + + def _stop_proxy(_: Any = None) -> None: + if proxy_state is not None: + proxy_state.stop() + + reactor.addSystemEventTrigger("before", "shutdown", _stop_proxy) + _wire_proxy_report_slots(report_factory, proxy_state) + logger.info("(GLOBAL) ADN DMR Peer Server started. Use adn-dmr-server as reference.") reactor.suggestThreadPoolSize(100) reactor.run() diff --git a/tests/application/test_proxy_use_cases.py b/tests/application/test_proxy_use_cases.py index 8f75a0f..8c97685 100644 --- a/tests/application/test_proxy_use_cases.py +++ b/tests/application/test_proxy_use_cases.py @@ -2,43 +2,41 @@ from __future__ import annotations -import random +import time import pytest from adn_server.application.proxy import ProxyUseCases, peer_id_from_packet -from adn_server.domain.proxy import ClientEndpoint, UpstreamPortRange +from adn_server.domain.proxy import ClientEndpoint from adn_server.domain.result import is_fail, is_ok from adn_server.domain.value_objects import bytes_4 from adn_server.infrastructure.proxy import InMemoryPendingRptoQueue, InMemoryProxySlotStore -_PORT_RANGE = UpstreamPortRange(port_start=56400, port_count=4) _PEER_A = bytes_4(1234567) _PEER_B = bytes_4(7654321) +_MAX_PEERS = 4 -def _service(*, black_list: tuple[int, ...] = (), seed: int = 1) -> ProxyUseCases: +def _service(*, max_peers: int = _MAX_PEERS, black_list: tuple[int, ...] = ()) -> ProxyUseCases: return ProxyUseCases( - InMemoryProxySlotStore(_PORT_RANGE), - _PORT_RANGE, + InMemoryProxySlotStore(), InMemoryPendingRptoQueue(), + max_peers=max_peers, black_list=black_list, - rng=random.Random(seed), ) -def test_attach_allocates_upstream_port_and_refreshes_client() -> None: +def test_attach_creates_session_and_refreshes_client() -> None: svc = _service() first = svc.attach_client(_PEER_A, "10.0.0.1", 62031) assert is_ok(first) slot = first.value - assert slot.upstream_port in _PORT_RANGE.ports() - assert slot.client.host == "10.0.0.1" + assert slot.client == ClientEndpoint(host="10.0.0.1", port=62031) again = svc.attach_client(_PEER_A, "10.0.0.2", 62032) assert is_ok(again) - assert again.value.upstream_port == slot.upstream_port assert again.value.client == ClientEndpoint(host="10.0.0.2", port=62032) + assert len(svc.list_slots()) == 1 def test_attach_rejects_blacklisted_peer() -> None: @@ -47,32 +45,65 @@ def test_attach_rejects_blacklisted_peer() -> None: assert is_fail(result) -def test_attach_fails_when_ports_exhausted() -> None: - svc = _service() +def test_attach_fails_when_max_peers_exceeded() -> None: + svc = _service(max_peers=4) for n in range(4): peer = bytes_4(1000 + n) assert is_ok(svc.attach_client(peer, "10.0.0.1", 62031 + n)) assert is_fail(svc.attach_client(bytes_4(9999), "10.0.0.9", 62099)) -def test_detach_frees_upstream_port() -> None: - svc = _service() +def test_detach_removes_session() -> None: + svc = _service(max_peers=4) slot = svc.attach_client(_PEER_A, "10.0.0.1", 62031).value removed = svc.detach_client(_PEER_A) assert removed == slot - assert svc.resolve_upstream(_PEER_A) is None + assert svc.resolve_client(_PEER_A) is None assert is_ok(svc.attach_client(_PEER_B, "10.0.0.5", 62035)) -def test_resolve_client_by_upstream_port() -> None: +def test_resolve_client_by_peer_id() -> None: svc = _service() - slot = svc.attach_client(_PEER_A, "192.168.1.10", 12345).value - client = svc.resolve_client(slot.upstream_port) + svc.attach_client(_PEER_A, "192.168.1.10", 12345) + client = svc.resolve_client(_PEER_A) assert client is not None assert client.host == "192.168.1.10" assert client.port == 12345 +def test_expire_session_returns_teardown_plan() -> None: + svc = _service() + svc.attach_client(_PEER_A, "10.0.0.1", 62031) + teardown = svc.expire_session(_PEER_A) + assert teardown is not None + assert teardown.peer_id == _PEER_A + assert teardown.client.host == "10.0.0.1" + assert svc.resolve_client(_PEER_A) is None + + +def test_ip_blacklist_blocks_attach() -> None: + from adn_server.infrastructure.proxy import InMemoryProxyIpBlacklist + + bl = InMemoryProxyIpBlacklist() + bl.block_until("10.0.0.9", time.time() + 60) + svc = ProxyUseCases( + InMemoryProxySlotStore(), + InMemoryPendingRptoQueue(), + ip_blacklist=bl, + ) + assert is_fail(svc.attach_client(_PEER_A, "10.0.0.9", 62031)) + + +def test_attach_client_assigns_report_slot() -> None: + svc = _service(max_peers=4) + first = svc.attach_client(_PEER_A, "10.0.0.1", 62031) + second = svc.attach_client(_PEER_B, "10.0.0.2", 62032) + assert is_ok(first) + assert is_ok(second) + assert first.value.report_slot == 0 + assert second.value.report_slot == 1 + + def test_schedule_and_dequeue_rpto() -> None: svc = _service() svc.attach_client(_PEER_A, "10.0.0.1", 62031) @@ -83,7 +114,7 @@ def test_schedule_and_dequeue_rpto() -> None: assert pending is not None assert pending.peer_id == _PEER_A assert pending.payload == payload - assert pending.upstream_port == svc.resolve_upstream(_PEER_A) + assert pending.client == ClientEndpoint(host="10.0.0.1", port=62031) @pytest.mark.parametrize( diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..482fa49 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,31 @@ +"""Shared test fixtures.""" + +from __future__ import annotations + +from typing import Any + + +def minimal_valid_config(**overrides: Any) -> dict[str, Any]: + """Minimal config dict that passes validate_config (integrated proxy required).""" + config: dict[str, Any] = { + "GLOBAL": {"SERVER_ID": 1}, + "REPORTS": {"REPORT": False}, + "SYSTEMS": { + "HOTSPOT": { + "MODE": "MASTER", + "ENABLED": True, + "MAX_PEERS": 8, + } + }, + "PROXY": { + "LISTEN_PORT": 62031, + "TARGET_SYSTEM": "HOTSPOT", + "TIMEOUT": 30, + }, + } + for key, value in overrides.items(): + if isinstance(value, dict) and isinstance(config.get(key), dict): + config[key] = {**config[key], **value} + else: + config[key] = value + return config diff --git a/tests/infrastructure/test_config_regression.py b/tests/infrastructure/test_config_regression.py new file mode 100644 index 0000000..24312b4 --- /dev/null +++ b/tests/infrastructure/test_config_regression.py @@ -0,0 +1,19 @@ +"""Regression: configs that must keep loading after proxy integration.""" + +from __future__ import annotations + +from adn_server.infrastructure import YamlConfigLoader + + +def test_adn_parrot_yaml_loads_without_proxy_block() -> None: + loader = YamlConfigLoader("/opt/new-adn-server") + config = loader.load("/opt/new-adn-server/adn-parrot.yaml") + assert "PROXY" not in config or config.get("PROXY") == {} + assert config["SYSTEMS"]["PARROT"]["MODE"] == "PEER" + + +def test_adn_server_yaml_loads_with_proxy() -> None: + loader = YamlConfigLoader("/opt/new-adn-server") + config = loader.load("/opt/new-adn-server/adn-server.yaml") + assert config["PROXY"]["TARGET_SYSTEM"] == "SYSTEM" + assert config["PROXY"]["LISTEN_PORT"] == 62031 diff --git a/tests/infrastructure/test_mqtt_config.py b/tests/infrastructure/test_mqtt_config.py index 585f672..2dbcb23 100644 --- a/tests/infrastructure/test_mqtt_config.py +++ b/tests/infrastructure/test_mqtt_config.py @@ -8,6 +8,8 @@ from adn_server.infrastructure.twisted_adapters.report.mqtt_config import ( parse_mqtt_broker, ) +from tests.conftest import minimal_valid_config + def test_mqtt_disabled_by_default(): config = {"REPORTS": {"REPORT": True}, "GLOBAL": {"SERVER_ID": 73010}} @@ -52,11 +54,9 @@ def test_mqtt_default_topic_prefix_from_server_id(): def test_validate_mqtt_enabled_requires_url(): - config = { - "GLOBAL": {"SERVER_ID": 1}, - "REPORTS": {"MQTT": {"ENABLED": True}}, - "SYSTEMS": {}, - } + config = minimal_valid_config( + REPORTS={"MQTT": {"ENABLED": True}}, + ) try: validate_config(config) raised = False diff --git a/tests/infrastructure/test_proxy_config.py b/tests/infrastructure/test_proxy_config.py new file mode 100644 index 0000000..6f6ae03 --- /dev/null +++ b/tests/infrastructure/test_proxy_config.py @@ -0,0 +1,118 @@ +"""PROXY configuration validation (Phase 3).""" + +from __future__ import annotations + +import logging + +import pytest + +from adn_server.application.proxy.deployment import ( + is_proxy_inject_only, + normalize_proxy_target, +) +from adn_server.domain.errors import ConfigError +from adn_server.infrastructure.config_normalizer import expand_generator +from adn_server.infrastructure.config_validator import validate_config +from adn_server.infrastructure.proxy.config import apply_proxy_env_overrides + +from tests.conftest import minimal_valid_config + + +def test_proxy_section_required_when_master_present() -> None: + with pytest.raises(ConfigError) as exc: + validate_config({ + "GLOBAL": {"SERVER_ID": 1}, + "SYSTEMS": {"HOTSPOT": {"MODE": "MASTER", "ENABLED": True, "MAX_PEERS": 1}}, + }) + assert "PROXY" in str(exc.value) + + +def test_parrot_config_without_proxy_is_valid() -> None: + validate_config({ + "GLOBAL": {"SERVER_ID": 9990}, + "SYSTEMS": {"PARROT": {"MODE": "PEER", "ENABLED": True, "PORT": 54915}}, + }) + + +def test_proxy_enabled_key_rejected() -> None: + config = minimal_valid_config() + config["PROXY"]["ENABLED"] = False + with pytest.raises(ConfigError) as exc: + validate_config(config) + assert "PROXY.ENABLED" in str(exc.value) + + +def test_proxy_enabled_requires_listen_port_and_target() -> None: + config = minimal_valid_config() + del config["PROXY"]["TARGET_SYSTEM"] + with pytest.raises(ConfigError) as exc: + validate_config(config) + assert "TARGET_SYSTEM" in str(exc.value) + + +def test_proxy_target_rejects_port_and_generator() -> None: + config = minimal_valid_config() + config["SYSTEMS"]["HOTSPOT"]["PORT"] = 56400 + with pytest.raises(ConfigError) as exc: + validate_config(config) + assert "PORT" in str(exc.value) + + config = minimal_valid_config() + config["SYSTEMS"]["HOTSPOT"]["GENERATOR"] = 102 + with pytest.raises(ConfigError) as exc: + validate_config(config) + assert "GENERATOR" in str(exc.value) + + +def test_proxy_rejects_deprecated_keys() -> None: + config = minimal_valid_config() + config["PROXY"]["DISPATCH"] = True + with pytest.raises(ConfigError) as exc: + validate_config(config) + assert "DISPATCH" in str(exc.value) + + +def test_valid_minimal_proxy_config() -> None: + validate_config(minimal_valid_config()) + + +def test_normalize_proxy_target_strips_bind_fields() -> None: + config = minimal_valid_config() + config["SYSTEMS"]["HOTSPOT"]["PORT"] = 56400 + config["SYSTEMS"]["HOTSPOT"]["IP"] = "127.0.0.1" + config["SYSTEMS"]["HOTSPOT"]["GENERATOR"] = 102 + normalize_proxy_target(config) + target = config["SYSTEMS"]["HOTSPOT"] + assert "PORT" not in target + assert "IP" not in target + assert target["_REPORT_BASE_PORT"] == 56400 + assert is_proxy_inject_only(config, "HOTSPOT") + + +def test_apply_proxy_env_overrides(monkeypatch: pytest.MonkeyPatch) -> None: + config = minimal_valid_config() + monkeypatch.setenv("ADN_PROXY_DEBUG", "1") + monkeypatch.setenv("ADN_PROXY_LISTENPORT", "63000") + monkeypatch.setenv("ADN_PROXY_IPV6", "1") + apply_proxy_env_overrides(config) + assert config["PROXY"]["DEBUG"] is True + assert config["PROXY"]["LISTEN_PORT"] == 63000 + assert config["PROXY"]["LISTEN_IP"] == "::" + + +def test_expand_generator_skips_proxy_target() -> None: + config = minimal_valid_config( + PROXY={"LISTEN_PORT": 62031, "TARGET_SYSTEM": "HOTSPOT"}, + SYSTEMS={ + "HOTSPOT": { + "MODE": "MASTER", + "ENABLED": True, + "PORT": 56400, + "GENERATOR": 4, + "MAX_PEERS": 8, + } + }, + ) + expand_generator(config, logging.getLogger("test")) + assert "HOTSPOT-0" not in config["SYSTEMS"] + assert "HOTSPOT" in config["SYSTEMS"] diff --git a/tests/infrastructure/test_proxy_reload.py b/tests/infrastructure/test_proxy_reload.py new file mode 100644 index 0000000..a6fa014 --- /dev/null +++ b/tests/infrastructure/test_proxy_reload.py @@ -0,0 +1,87 @@ +"""Proxy hot-reload keeps LISTEN_PORT and active sessions.""" + +from __future__ import annotations + +import logging + +from adn_server.application.proxy import ProxyUseCases +from adn_server.domain.proxy import ClientEndpoint, ClientSlot +from adn_server.infrastructure.proxy.ip_blacklist import InMemoryProxyIpBlacklist +from adn_server.infrastructure.proxy.rpto_queue import InMemoryPendingRptoQueue +from adn_server.infrastructure.proxy.runtime import ProxyServiceState, apply_proxy_config_reload +from adn_server.infrastructure.proxy.slot_store import InMemoryProxySlotStore +from adn_server.infrastructure.proxy.udp_fanin import ProxyFanInProtocol + + +def _minimal_proxy_state() -> ProxyServiceState: + use_cases = ProxyUseCases( + InMemoryProxySlotStore(), + InMemoryPendingRptoQueue(), + max_peers=10, + black_list=(), + ip_blacklist=InMemoryProxyIpBlacklist(), + ) + peer = b"\x00\x07\x06\xf5" # 730039101 + use_cases._slots.bind( # noqa: SLF001 + ClientSlot( + peer_id=peer, + client=ClientEndpoint(host="10.0.0.1", port=62031), + report_slot=0, + ) + ) + fanin = ProxyFanInProtocol(use_cases, master_sink=None) # type: ignore[arg-type] + return ProxyServiceState( + target_system="SYSTEM", + use_cases=use_cases, + master_sink=None, # type: ignore[arg-type] + client_sender=None, # type: ignore[arg-type] + fanin=fanin, + udp_port=object(), + listen_port=62031, + listen_ip="", + _runtime={ + "listen_port": 62031, + "listen_ip": "", + "target_system": "SYSTEM", + "timeout": 30.0, + "debug": False, + "client_info": True, + "black_list": (), + "ip_black_list": {}, + "max_peers": 10, + }, + ) + + +def test_apply_proxy_config_reload_keeps_sessions_and_updates_timeout() -> None: + state = _minimal_proxy_state() + config = { + "PROXY": { + "LISTEN_PORT": 62031, + "TARGET_SYSTEM": "SYSTEM", + "TIMEOUT": 45, + "DEBUG": True, + }, + "SYSTEMS": {"SYSTEM": {"MAX_PEERS": 20}}, + } + log = logging.getLogger("test.proxy.reload") + + apply_proxy_config_reload(state, config, logger=log) + + assert len(state.use_cases.list_slots()) == 1 + assert state.udp_port is not None + assert state._runtime["timeout"] == 45.0 + assert state.fanin.debug is True + assert state.use_cases._max_peers == 20 # noqa: SLF001 + + +def test_proxy_use_cases_apply_runtime_settings() -> None: + uc = ProxyUseCases( + InMemoryProxySlotStore(), + InMemoryPendingRptoQueue(), + max_peers=5, + black_list=(1001,), + ) + uc.apply_runtime_settings(max_peers=12, black_list=(2002, 3003)) + assert uc._max_peers == 12 # noqa: SLF001 + assert uc._black_list == frozenset({2002, 3003}) # noqa: SLF001 diff --git a/tests/infrastructure/test_proxy_smoke_live_udp.py b/tests/infrastructure/test_proxy_smoke_live_udp.py new file mode 100644 index 0000000..6a66772 --- /dev/null +++ b/tests/infrastructure/test_proxy_smoke_live_udp.py @@ -0,0 +1,91 @@ +"""Live UDP smoke test for integrated proxy (isolated port; no production restart).""" + +from __future__ import annotations + +import socket +import threading + +import pytest +from twisted.internet import reactor + +from adn_server.application.proxy import ProxyUseCases +from adn_server.domain.value_objects import bytes_4 +from adn_server.infrastructure.config_normalizer import ensure_system_runtime_config +from adn_server.infrastructure.hbp_constants import RPTACK, RPTL +from adn_server.infrastructure.proxy import ( + InMemoryPendingRptoQueue, + InMemoryProxySlotStore, + InProcessHbpSink, + ProxyFanInProtocol, + ProxyReplyTransport, +) +from adn_server.infrastructure.twisted_adapters.udp_hbp import HBPProtocol + +_SMOKE_PORT = 62032 +_PEER = bytes_4(1234567) + + +class _AclRouter: + def acl_check(self, peer_id: bytes, acl: object) -> bool: + return True + + +def _build_fanin() -> ProxyFanInProtocol: + config = { + "GLOBAL": {"PING_TIME": 10, "MAX_MISSED": 3, "USE_ACL": False}, + "SYSTEMS": { + "HOTSPOT": { + "MODE": "MASTER", + "ENABLED": True, + "MAX_PEERS": 8, + "OPTIONS": "TS2=9990;", + } + }, + } + ensure_system_runtime_config(config) + hbp = HBPProtocol("HOTSPOT", config) + hbp._router = _AclRouter() # type: ignore[assignment] + proxy = ProxyUseCases( + InMemoryProxySlotStore(), + InMemoryPendingRptoQueue(), + max_peers=8, + ) + sink = InProcessHbpSink(hbp) + fanin = ProxyFanInProtocol(proxy, sink) + return fanin, hbp, sink + + +@pytest.mark.smoke +def test_live_udp_rptl_rptack_on_isolated_port() -> None: + """RPTL in → inject → RPTACK out on 127.0.0.1:62032 (does not use production 62031).""" + fanin, hbp, _ = _build_fanin() + result: dict[str, bytes | None] = {"reply": None, "error": None} + + def _run_client() -> None: + try: + client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + client.settimeout(2.0) + client.bind(("127.0.0.1", 0)) + client.sendto(RPTL + _PEER, ("127.0.0.1", _SMOKE_PORT)) + data, _ = client.recvfrom(4096) + result["reply"] = data + client.close() + except OSError as exc: + result["error"] = str(exc).encode() + finally: + reactor.callFromThread(reactor.stop) + + listener = reactor.listenUDP(_SMOKE_PORT, fanin, interface="127.0.0.1") + hbp.transport = ProxyReplyTransport(fanin.transport) + reactor.callWhenRunning( + lambda: threading.Thread(target=_run_client, daemon=True).start() + ) + reactor.callLater(5.0, reactor.stop) + reactor.run() + + listener.stopListening() + if result["error"]: + pytest.fail(result["error"].decode()) + reply = result["reply"] + assert reply is not None, "no UDP reply (timeout?)" + assert reply.startswith(RPTACK), f"expected RPTACK, got {reply[:8]!r}" diff --git a/tests/infrastructure/test_session_teardown.py b/tests/infrastructure/test_session_teardown.py new file mode 100644 index 0000000..f6119d9 --- /dev/null +++ b/tests/infrastructure/test_session_teardown.py @@ -0,0 +1,35 @@ +"""Session teardown executor (legacy reaper wire parity).""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +from adn_server.application.proxy.session_teardown import CLIENT_TEARDOWN_REPEAT, client_teardown_packet, master_teardown_packet +from adn_server.domain.proxy import ClientEndpoint, SessionTeardown +from adn_server.infrastructure.proxy.session_executor import apply_session_teardown +from adn_server.domain.value_objects import bytes_4 + +_PEER = bytes_4(1234567) +_CLIENT = ClientEndpoint(host="10.0.0.8", port=5000) + + +def test_apply_session_teardown_sends_rptcl_and_mstcl() -> None: + master = MagicMock() + client = MagicMock() + registry = MagicMock() + teardown = SessionTeardown(peer_id=_PEER, client=_CLIENT) + apply_session_teardown( + teardown, + master_sink=master, + client_sender=client, + peer_registry=registry, + ) + master.inject.assert_called_once_with( + master_teardown_packet(_PEER), + ("10.0.0.8", 5000), + ) + assert client.send_to_client.call_count == CLIENT_TEARDOWN_REPEAT + for call in client.send_to_client.call_args_list: + assert call.args[0] == client_teardown_packet() + assert call.args[1] == _CLIENT + registry.remove_peer.assert_called_once_with(_PEER) diff --git a/tests/infrastructure/test_udp_fanin.py b/tests/infrastructure/test_udp_fanin.py new file mode 100644 index 0000000..9c520c1 --- /dev/null +++ b/tests/infrastructure/test_udp_fanin.py @@ -0,0 +1,109 @@ +"""UDP fan-in protocol tests (no live reactor).""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +from adn_server.application.proxy import ProxyUseCases +from adn_server.domain.value_objects import bytes_4 +from adn_server.infrastructure.config_normalizer import ensure_system_runtime_config +from adn_server.infrastructure.hbp_constants import RPTACK, RPTL +from adn_server.infrastructure.proxy import ( + InMemoryPendingRptoQueue, + InMemoryProxySlotStore, + InProcessHbpSink, + ProxyFanInProtocol, + ProxyReplyTransport, +) +from adn_server.infrastructure.twisted_adapters.udp_hbp import HBPProtocol + +_PEER = bytes_4(1234567) +_CLIENT_ADDR = ("192.168.1.50", 62031) + + +class _RecordingTransport: + def __init__(self) -> None: + self.sent: list[tuple[bytes, tuple[str, int]]] = [] + + def write(self, data: bytes, addr: tuple[str, int]) -> None: + self.sent.append((data, addr)) + + +class _AclRouter: + def acl_check(self, peer_id: bytes, acl: object) -> bool: + return True + + +def _hotspot_master_config() -> dict: + config = { + "GLOBAL": {"PING_TIME": 10, "MAX_MISSED": 3, "USE_ACL": False}, + "SYSTEMS": { + "HOTSPOT": { + "MODE": "MASTER", + "ENABLED": True, + "MAX_PEERS": 8, + "OPTIONS": "TS2=9990;", + } + }, + } + ensure_system_runtime_config(config) + return config + + +def _fanin_stack(*, max_peers: int = 8) -> tuple[ProxyFanInProtocol, ProxyUseCases, InProcessHbpSink, _RecordingTransport, MagicMock]: + transport = _RecordingTransport() + config = _hotspot_master_config() + hbp = HBPProtocol("HOTSPOT", config) + hbp._router = _AclRouter() # type: ignore[assignment] + hbp.transport = ProxyReplyTransport(transport) + sink = InProcessHbpSink(hbp) + inject_spy = MagicMock(wraps=sink.inject) + sink.inject = inject_spy # type: ignore[method-assign] + proxy = ProxyUseCases( + InMemoryProxySlotStore(), + InMemoryPendingRptoQueue(), + max_peers=max_peers, + ) + protocol = ProxyFanInProtocol(proxy, sink) + protocol.transport = transport # type: ignore[assignment] + return protocol, proxy, sink, transport, inject_spy + + +def test_client_packet_attaches_session_and_injects() -> None: + protocol, proxy, _, _, inject_spy = _fanin_stack() + packet = RPTL + _PEER + protocol.datagramReceived(packet, _CLIENT_ADDR) + inject_spy.assert_called_once_with(packet, _CLIENT_ADDR) + slot = proxy.resolve_client(_PEER) + assert slot is not None + assert slot.host == _CLIENT_ADDR[0] + assert slot.port == _CLIENT_ADDR[1] + + +def test_existing_client_refreshes_endpoint_before_inject() -> None: + protocol, proxy, _, _, inject_spy = _fanin_stack() + packet = RPTL + _PEER + protocol.datagramReceived(packet, _CLIENT_ADDR) + inject_spy.reset_mock() + new_addr = ("192.168.1.51", 62032) + protocol.datagramReceived(packet, new_addr) + inject_spy.assert_called_once_with(packet, new_addr) + client = proxy.resolve_client(_PEER) + assert client is not None + assert client.host == new_addr[0] + assert client.port == new_addr[1] + + +def test_master_rptl_reply_sent_via_proxy_listen_socket() -> None: + protocol, _, _, transport, _ = _fanin_stack() + protocol.datagramReceived(RPTL + _PEER, _CLIENT_ADDR) + assert len(transport.sent) == 1 + data, addr = transport.sent[0] + assert data.startswith(RPTACK) + assert addr == _CLIENT_ADDR + + +def test_attach_rejection_skips_inject() -> None: + protocol, _, _, _, inject_spy = _fanin_stack(max_peers=0) + protocol.datagramReceived(RPTL + _PEER, _CLIENT_ADDR) + inject_spy.assert_not_called()