diff --git a/adn-server.example.yaml b/adn-server.example.yaml index 8ec6bde..4c6960f 100644 --- a/adn-server.example.yaml +++ b/adn-server.example.yaml @@ -78,6 +78,17 @@ PROXY: BLACK_LIST: [] IP_BLACK_LIST: {} +# Self-service: same MySQL + PBKDF2 as monitor/adn-monitor.yaml (Clients table). +SELF_SERVICE: + USE_SELFSERVICE: true + DB_SERVER: localhost + DB_USERNAME: hbmon + DB_PASSWORD: "" + DB_NAME: hbmon + DB_PORT: 3306 + PBKDF2_SALT: "ADN" + PBKDF2_ITERATIONS: 2000 + # SYSTEM is inject-only (no PORT / GENERATOR on proxy target — use MAX_PEERS). SYSTEMS: SYSTEM: diff --git a/pyproject.toml b/pyproject.toml index 77697f5..b84af12 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,7 @@ dependencies = [ [project.optional-dependencies] mqtt = ["paho-mqtt>=2.0"] +selfservice = ["mysqlclient>=2.0"] dev = ["pytest>=7.0", "jsonschema>=4.0", "paho-mqtt>=2.0"] docs = ["mkdocs>=1.6", "mkdocs-material>=9.5", "pymdown-extensions>=10.3"] diff --git a/src/adn_server/application/ports.py b/src/adn_server/application/ports.py index 003c978..ca29d16 100644 --- a/src/adn_server/application/ports.py +++ b/src/adn_server/application/ports.py @@ -341,6 +341,51 @@ class PendingRptoQueue(ABC): ... +class ProxySelfServiceStore(ABC): + """Self-service ``Clients`` table (legacy adn-proxy / hotspot_proxy_self_service).""" + + @abstractmethod + def test_db(self) -> Any: + """Verify DB connectivity. Returns Twisted Deferred.""" + + @abstractmethod + def ins_conf( + self, + int_id: int, + peer_id_bytes: bytes, + callsign: str, + host: str, + mode: str, + ) -> None: + ... + + @abstractmethod + def updt_tbl( + self, + action: str, + peer_id_bytes: bytes, + *, + psswd: str | None = None, + ) -> None: + ... + + @abstractmethod + def slct_opt(self, peer_id_bytes: bytes) -> Any: + """Returns Deferred firing with row list, e.g. ``((options_str,),)``.""" + + @abstractmethod + def slct_db(self) -> Any: + """Returns Deferred firing with ``(dmr_id, options)`` rows for ``modified=1``.""" + + @abstractmethod + def updt_lstseen(self, dmrid_list: list[tuple[bytes, ...]]) -> None: + ... + + @abstractmethod + def clean_tbl(self) -> Any: + """Returns Deferred.""" + + class ProxyIpBlacklist(ABC): """Temporary IP blocks (legacy proxy ``ip_black_list`` / PRBL).""" diff --git a/src/adn_server/infrastructure/config_loader.py b/src/adn_server/infrastructure/config_loader.py index e91959b..38d6c1e 100644 --- a/src/adn_server/infrastructure/config_loader.py +++ b/src/adn_server/infrastructure/config_loader.py @@ -107,6 +107,7 @@ class YamlConfigLoader: "ALIASES": data.get("ALIASES", {}), "SYSTEMS": data.get("SYSTEMS", {}), "PROXY": data.get("PROXY", {}), + "SELF_SERVICE": data.get("SELF_SERVICE", {}), } apply_proxy_env_overrides(config) # Ensure REPORT_CLIENTS is list diff --git a/src/adn_server/infrastructure/proxy/__init__.py b/src/adn_server/infrastructure/proxy/__init__.py index d6ef51f..aaeb61f 100644 --- a/src/adn_server/infrastructure/proxy/__init__.py +++ b/src/adn_server/infrastructure/proxy/__init__.py @@ -1,6 +1,7 @@ """Proxy infrastructure adapters (Phase 3).""" from .config import apply_proxy_env_overrides, proxy_settings +from .self_service_config import self_service_settings from .hbp_adapters import FanInClientSender, HbpMasterPeerRegistry, InProcessHbpSink from .ip_blacklist import InMemoryProxyIpBlacklist from .reply_transport import ProxyReplyTransport @@ -25,5 +26,6 @@ __all__ = [ "apply_session_teardown", "listen_proxy_fanin", "proxy_settings", + "self_service_settings", "start_proxy_service", ] diff --git a/src/adn_server/infrastructure/proxy/null_self_service.py b/src/adn_server/infrastructure/proxy/null_self_service.py new file mode 100644 index 0000000..9c6043b --- /dev/null +++ b/src/adn_server/infrastructure/proxy/null_self_service.py @@ -0,0 +1,53 @@ +"""No-op self-service store when ``SELF_SERVICE.USE_SELFSERVICE`` is false.""" + +from __future__ import annotations + +from typing import Any + +from adn_server.application.ports import ProxySelfServiceStore + + +class NullProxySelfServiceStore(ProxySelfServiceStore): + """Disabled self-service: all methods are no-ops.""" + + def test_db(self) -> Any: + from twisted.internet.defer import succeed + + return succeed(True) + + def ins_conf( + self, + int_id: int, + peer_id_bytes: bytes, + callsign: str, + host: str, + mode: str, + ) -> None: + pass + + def updt_tbl( + self, + action: str, + peer_id_bytes: bytes, + *, + psswd: str | None = None, + ) -> None: + pass + + def slct_opt(self, peer_id_bytes: bytes) -> Any: + from twisted.internet.defer import succeed + + return succeed([]) + + def slct_db(self) -> Any: + from twisted.internet.defer import succeed + + return succeed([]) + + def updt_lstseen(self, dmrid_list: list[tuple[bytes, ...]]) -> None: + pass + + def clean_tbl(self) -> Any: + from twisted.internet.defer import succeed + + return succeed(None) diff --git a/src/adn_server/infrastructure/proxy/persistence/__init__.py b/src/adn_server/infrastructure/proxy/persistence/__init__.py new file mode 100644 index 0000000..b1388af --- /dev/null +++ b/src/adn_server/infrastructure/proxy/persistence/__init__.py @@ -0,0 +1,6 @@ +"""Proxy self-service persistence (MySQL Clients table).""" + +from .db_pool import create_pool, test_db +from .proxy_repository import ProxySelfServiceRepository + +__all__ = ["ProxySelfServiceRepository", "create_pool", "test_db"] diff --git a/src/adn_server/infrastructure/proxy/persistence/db_pool.py b/src/adn_server/infrastructure/proxy/persistence/db_pool.py new file mode 100644 index 0000000..10ca948 --- /dev/null +++ b/src/adn_server/infrastructure/proxy/persistence/db_pool.py @@ -0,0 +1,42 @@ +"""Twisted adbapi MySQL pool for proxy self-service (legacy adn-proxy parity).""" + +from __future__ import annotations + +import logging + +from twisted.enterprise import adbapi +from twisted.internet.defer import inlineCallbacks, returnValue + +logger = logging.getLogger(__name__) + + +def create_pool( + host: str, + user: str, + password: str, + db_name: str, + port: int, +) -> adbapi.ConnectionPool: + """Create Twisted adbapi pool using ``MySQLdb`` (``mysqlclient`` package).""" + return adbapi.ConnectionPool( + "MySQLdb", + host=host, + user=user, + passwd=password, + db=db_name, + port=port, + charset="utf8mb4", + ) + + +@inlineCallbacks +def test_db(pool: adbapi.ConnectionPool) -> bool: + """Verify DB connectivity. Returns True on success.""" + try: + res = yield pool.runQuery("SELECT 1") + if res: + logger.info("(SELF_SERVICE) Database connection test: OK") + returnValue(True) + except Exception as err: + logger.error("(SELF_SERVICE) Database connection error: %s", err) + returnValue(False) diff --git a/src/adn_server/infrastructure/proxy/persistence/proxy_repository.py b/src/adn_server/infrastructure/proxy/persistence/proxy_repository.py new file mode 100644 index 0000000..5675eed --- /dev/null +++ b/src/adn_server/infrastructure/proxy/persistence/proxy_repository.py @@ -0,0 +1,121 @@ +"""MySQL ``Clients`` table adapter (legacy ``ProxyDbRepository`` parity).""" + +from __future__ import annotations + +import logging +from typing import Any + +from twisted.enterprise import adbapi +from twisted.internet.defer import inlineCallbacks, returnValue + +from adn_server.application.ports import ProxySelfServiceStore + +from .db_pool import test_db + +logger = logging.getLogger(__name__) + + +class ProxySelfServiceRepository(ProxySelfServiceStore): + """``ProxySelfServiceStore`` via Twisted adbapi (``Clients`` table).""" + + def __init__(self, pool: adbapi.ConnectionPool) -> None: + self._pool = pool + + def test_db(self) -> Any: + return test_db(self._pool) + + def ins_conf( + self, + int_id: int, + peer_id_bytes: bytes, + callsign: str, + host: str, + mode: str, + ) -> None: + self._pool.runOperation( + """INSERT INTO Clients (int_id, dmr_id, callsign, host, mode, logged_in, last_seen) + VALUES (%s, %s, %s, %s, %s, 1, UNIX_TIMESTAMP()) + ON DUPLICATE KEY UPDATE callsign=%s, host=%s, mode=%s, logged_in=1, last_seen=UNIX_TIMESTAMP()""", + (int_id, peer_id_bytes, callsign, host, mode, callsign, host, mode), + ).addErrback(lambda f: logger.error("(SELF_SERVICE) ins_conf: %s", f.getTraceback())) + + def updt_tbl( + self, + action: str, + peer_id_bytes: bytes, + *, + psswd: str | None = None, + ) -> None: + if action == "log_out": + self._pool.runOperation( + "UPDATE Clients SET logged_in=0, last_seen=UNIX_TIMESTAMP() WHERE dmr_id=%s", + (peer_id_bytes,), + ).addErrback( + lambda f: logger.error("(SELF_SERVICE) log_out: %s", f.getTraceback()) + ) + elif action == "psswd" and psswd is not None: + blob = psswd.encode("utf-8") if isinstance(psswd, str) else psswd + self._pool.runOperation( + "UPDATE Clients SET psswd=%s WHERE dmr_id=%s", + (blob, peer_id_bytes), + ).addErrback( + lambda f: logger.error("(SELF_SERVICE) psswd: %s", f.getTraceback()) + ) + elif action == "opt_rcvd": + self._pool.runOperation( + "UPDATE Clients SET opt_rcvd=1 WHERE dmr_id=%s", + (peer_id_bytes,), + ).addErrback( + lambda f: logger.error("(SELF_SERVICE) opt_rcvd: %s", f.getTraceback()) + ) + elif action == "rst_mod": + self._pool.runOperation( + "UPDATE Clients SET modified=0 WHERE dmr_id=%s", + (peer_id_bytes,), + ).addErrback( + lambda f: logger.error("(SELF_SERVICE) rst_mod: %s", f.getTraceback()) + ) + + @inlineCallbacks + def slct_opt(self, peer_id_bytes: bytes) -> Any: + try: + rows = yield self._pool.runQuery( + "SELECT options FROM Clients WHERE dmr_id=%s AND options IS NOT NULL AND options != ''", + (peer_id_bytes,), + ) + returnValue(rows) + except Exception as err: + logger.error("(SELF_SERVICE) slct_opt: %s", err) + returnValue([]) + + @inlineCallbacks + def slct_db(self) -> Any: + try: + rows = yield self._pool.runQuery( + "SELECT dmr_id, options FROM Clients WHERE modified=1 AND logged_in=1 " + "AND options IS NOT NULL AND options != ''" + ) + returnValue(rows or []) + except Exception as err: + logger.error("(SELF_SERVICE) slct_db: %s", err) + returnValue([]) + + def updt_lstseen(self, dmrid_list: list[tuple[bytes, ...]]) -> None: + if not dmrid_list: + return + for item in dmrid_list: + peer_id = item[0] + self._pool.runOperation( + "UPDATE Clients SET last_seen=UNIX_TIMESTAMP() WHERE dmr_id=%s", + (peer_id,), + ).addErrback( + lambda f, pid=peer_id: logger.error( + "(SELF_SERVICE) updt_lstseen: %s for %s", f.getTraceback(), pid + ) + ) + + @inlineCallbacks + def clean_tbl(self) -> Any: + yield self._pool.runOperation( + "UPDATE Clients SET logged_in=0 WHERE logged_in=1 AND last_seen < UNIX_TIMESTAMP() - 86400" + ).addErrback(lambda f: logger.error("(SELF_SERVICE) clean_tbl: %s", f.getTraceback())) diff --git a/src/adn_server/infrastructure/proxy/runtime.py b/src/adn_server/infrastructure/proxy/runtime.py index 240e694..31448f2 100644 --- a/src/adn_server/infrastructure/proxy/runtime.py +++ b/src/adn_server/infrastructure/proxy/runtime.py @@ -22,6 +22,8 @@ from adn_server.infrastructure.proxy.ip_blacklist import InMemoryProxyIpBlacklis 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.self_service_bridge import ProxySelfServiceBridge +from adn_server.infrastructure.proxy.self_service_config import self_service_settings from adn_server.infrastructure.proxy.udp_fanin import ProxyFanInProtocol, listen_proxy_fanin from adn_server.infrastructure.proxy.reply_transport import ProxyReplyTransport @@ -47,9 +49,13 @@ class ProxyServiceState: listen_ip: str = "" _timers: dict[bytes, IDelayedCall] = field(default_factory=dict) _runtime: dict[str, Any] = field(default_factory=dict) + self_service: ProxySelfServiceBridge | None = None def stop(self) -> Any: """Stop timers and UDP listener. Returns Twisted Deferred when a port was bound.""" + if self.self_service is not None: + self.self_service.stop_loops() + self.self_service = None for call in self._timers.values(): if call.active(): call.cancel() @@ -105,6 +111,47 @@ def apply_proxy_config_reload( ) +def _build_self_service( + config: dict[str, Any], + use_cases: ProxyUseCases, + master_sink: InProcessHbpSink, + client_sender: FanInClientSender, + *, + logger: logging.Logger, +) -> ProxySelfServiceBridge | None: + ss = self_service_settings(config) + if not ss["enabled"]: + return None + try: + from adn_server.infrastructure.proxy.persistence import ( + ProxySelfServiceRepository, + create_pool, + ) + except ImportError as err: + raise RuntimeError( + "(SELF_SERVICE) USE_SELFSERVICE requires mysqlclient " + "(pip install mysqlclient)" + ) from err + pool = create_pool( + ss["db_server"], + ss["db_username"], + ss["db_password"], + ss["db_name"], + ss["db_port"], + ) + store = ProxySelfServiceRepository(pool) + bridge = ProxySelfServiceBridge( + store, + use_cases, + master_sink, + client_sender, + pbkdf2_salt=ss["pbkdf2_salt"], + pbkdf2_iterations=ss["pbkdf2_iterations"], + logger=logger, + ) + return bridge + + def start_proxy_service( config: dict[str, Any], protocols: dict[str, Any], @@ -134,6 +181,7 @@ def start_proxy_service( debug=runtime["debug"], logger=logger, ) + self_service_bridge: ProxySelfServiceBridge | None = None state = ProxyServiceState( target_system=target, use_cases=use_cases, @@ -187,6 +235,8 @@ def start_proxy_service( teardown.client.host.rjust(15), teardown.client.port, ) + if state.self_service is not None: + state.self_service.on_session_expired(peer_id) apply_session_teardown( teardown, master_sink=master_sink, @@ -214,6 +264,18 @@ def start_proxy_service( state.client_sender = FanInClientSender(fanin_proto.transport) target_proto.transport = ProxyReplyTransport(fanin_proto.transport, prbl_handler=_handle_prbl) + ss_settings = self_service_settings(config) + if ss_settings["enabled"]: + self_service_bridge = _build_self_service( + config, + use_cases, + master_sink, + state.client_sender, + logger=logger, + ) + if self_service_bridge is not None: + fanin._self_service = self_service_bridge # noqa: SLF001 + logger.info( "(PROXY) Hotspot fan-in on %s:%s → inject %s (MAX_PEERS=%s, TIMEOUT=%ss)", runtime["listen_ip"] or "*", @@ -222,4 +284,16 @@ def start_proxy_service( runtime["max_peers"], runtime["timeout"], ) + if self_service_bridge is not None: + store = self_service_bridge._store # noqa: SLF001 + + def _on_db_ok(ok: bool) -> None: + if not ok: + logger.error("(SELF_SERVICE) Database connection failed — self-service disabled") + return + self_service_bridge.start_loops() + state.self_service = self_service_bridge + logger.info("(SELF_SERVICE) Enabled (shared Clients table with adn-monitor)") + + store.test_db().addCallback(_on_db_ok) return state diff --git a/src/adn_server/infrastructure/proxy/self_service_bridge.py b/src/adn_server/infrastructure/proxy/self_service_bridge.py new file mode 100644 index 0000000..fdd0c55 --- /dev/null +++ b/src/adn_server/infrastructure/proxy/self_service_bridge.py @@ -0,0 +1,203 @@ +"""Self-service hooks on integrated proxy (legacy adn-proxy DB + RPTO parity).""" + +from __future__ import annotations + +import logging +import struct +from hashlib import pbkdf2_hmac +from typing import Any + +from twisted.internet import reactor +from twisted.internet.defer import inlineCallbacks +from twisted.internet.interfaces import IDelayedCall +from twisted.internet.task import LoopingCall + +from adn_server.application.ports import ( + ProxyClientSender, + ProxyMasterSink, + ProxySelfServiceStore, +) +from adn_server.application.proxy import ProxyUseCases +from adn_server.domain.proxy import ClientEndpoint +from adn_server.domain.value_objects import int_id +from adn_server.infrastructure.hbp_constants import RPTACK, RPTC, RPTCL, RPTO + + +def _peer_id_from_db(value: Any) -> bytes | None: + if isinstance(value, bytes) and len(value) == 4: + return value + if isinstance(value, int) and 0 <= value <= 0xFFFFFFFF: + return struct.pack(">I", value) + return None + + +class ProxySelfServiceBridge: + """Mirror legacy proxy ``Clients`` writes and RPTO push loops on core fan-in.""" + + def __init__( + self, + store: ProxySelfServiceStore, + use_cases: ProxyUseCases, + master_sink: ProxyMasterSink, + client_sender: ProxyClientSender, + *, + pbkdf2_salt: str = "ADN", + pbkdf2_iterations: int = 2000, + logger: logging.Logger | None = None, + ) -> None: + self._store = store + self._use_cases = use_cases + self._master_sink = master_sink + self._client_sender = client_sender + self._pbkdf2_salt = pbkdf2_salt + self._pbkdf2_iterations = pbkdf2_iterations + self._log = logger or logging.getLogger(__name__) + self._opt_timers: dict[bytes, IDelayedCall] = {} + self._loop_calls: list[LoopingCall] = [] + + def start_loops(self) -> None: + """Legacy timers: send_opts 10s, lst_seen 120s, clean_tbl 3600s.""" + for interval, fn in ( + (10.0, self.send_opts), + (120.0, self.lst_seen), + (3600.0, self._clean_tbl), + ): + call = LoopingCall(fn) + call.start(interval, now=False) + self._loop_calls.append(call) + self._log.info( + "(SELF_SERVICE) DB options at login, send_opts every 10s, " + "clean_tbl every 1h, lst_seen every 2min" + ) + + def stop_loops(self) -> None: + for call in self._loop_calls: + if call.running: + call.stop() + self._loop_calls.clear() + for timer in self._opt_timers.values(): + if timer.active(): + timer.cancel() + self._opt_timers.clear() + + def before_inject( + self, + data: bytes, + addr: tuple[str, int], + peer_id: bytes, + ) -> bool: + """Handle RPTC/RPTO DB side effects. Return True to skip master inject.""" + if len(data) < 4: + return False + command = data[:4] + host, port = addr + if command == RPTO: + return self._handle_rpto(data, peer_id, host, port) + if command == RPTC and len(data) >= 5 and data[:5] != RPTCL: + self._handle_rptc(data, peer_id, host) + return False + + def on_session_expired(self, peer_id: bytes) -> None: + self._cancel_opt_timer(peer_id) + self._store.updt_tbl("log_out", peer_id) + + def _handle_rptc(self, data: bytes, peer_id: bytes, host: str) -> None: + if self._use_cases.resolve_client(peer_id) is None: + return + mode = data[97:98].decode("utf-8", errors="replace") if len(data) >= 98 else "4" + callsign = data[8:16].rstrip().decode("utf-8", errors="replace") + self._store.ins_conf(int_id(peer_id), peer_id, callsign, host, mode) + self._cancel_opt_timer(peer_id) + self._opt_timers[peer_id] = reactor.callLater(10, self._login_opt, peer_id) + + def _handle_rpto( + self, + data: bytes, + peer_id: bytes, + host: str, + port: int, + ) -> bool: + if self._use_cases.resolve_client(peer_id) is None: + return False + if data[8:].upper().startswith(b"PASS=") and len(data) >= 13: + psswd_raw = data[13:] + if len(psswd_raw) >= 6: + dk = pbkdf2_hmac( + "sha256", + psswd_raw, + self._pbkdf2_salt.encode("utf-8"), + self._pbkdf2_iterations, + ).hex() + self._store.updt_tbl("psswd", peer_id, psswd=dk) + self._client_sender.send_to_client( + RPTACK + peer_id, + ClientEndpoint(host=host, port=port), + ) + self._log.info("(SELF_SERVICE) Password stored for: %s", int_id(peer_id)) + return True + self._store.updt_tbl("opt_rcvd", peer_id) + self._cancel_opt_timer(peer_id) + self._log.info("(SELF_SERVICE) Options received from: %s", int_id(peer_id)) + return False + + def _cancel_opt_timer(self, peer_id: bytes) -> None: + timer = self._opt_timers.pop(peer_id, None) + if timer is not None and timer.active(): + timer.cancel() + + def _inject_rpto(self, peer_id: bytes, options: str | bytes) -> None: + client = self._use_cases.resolve_client(peer_id) + if client is None: + return + body = options.encode("utf-8") if isinstance(options, str) else options + packet = RPTO + peer_id + body + self._master_sink.inject(packet, (client.host, client.port)) + self._log.info( + "(SELF_SERVICE) Options sent for: %s, opt: %s", + int_id(peer_id), + options if isinstance(options, str) else options.decode("utf-8", errors="replace"), + ) + + @inlineCallbacks + def _login_opt(self, peer_id: bytes) -> None: + self._opt_timers.pop(peer_id, None) + if self._use_cases.resolve_client(peer_id) is None: + return + try: + rows = yield self._store.slct_opt(peer_id) + if not rows or not rows[0]: + return + options = rows[0][0] + if not options: + return + self._inject_rpto(peer_id, options) + self._log.info("(SELF_SERVICE) Options sent at login for: %s", int_id(peer_id)) + except Exception as err: + self._log.warning("(SELF_SERVICE) login_opt error: %s", err) + + @inlineCallbacks + def send_opts(self) -> None: + try: + results = yield self._store.slct_db() + for row in results: + if len(row) < 2: + continue + pid = _peer_id_from_db(row[0]) + options = row[1] + if pid is None or not options: + continue + if self._use_cases.resolve_client(pid) is None: + continue + self._store.updt_tbl("rst_mod", pid) + self._inject_rpto(pid, options) + self._log.info("(SELF_SERVICE) Options update sent for: %s", int_id(pid)) + except Exception as err: + self._log.warning("(SELF_SERVICE) send_opts error: %s", err) + + def lst_seen(self) -> None: + dmrid_list = [(slot.peer_id,) for slot in self._use_cases.list_slots()] + if dmrid_list: + self._store.updt_lstseen(dmrid_list) + + def _clean_tbl(self) -> None: + self._store.clean_tbl() diff --git a/src/adn_server/infrastructure/proxy/self_service_config.py b/src/adn_server/infrastructure/proxy/self_service_config.py new file mode 100644 index 0000000..608293f --- /dev/null +++ b/src/adn_server/infrastructure/proxy/self_service_config.py @@ -0,0 +1,23 @@ +"""SELF_SERVICE block from config (same keys as adn-monitor / adn-proxy).""" + +from __future__ import annotations + +from typing import Any + + +def self_service_settings(config: dict[str, Any]) -> dict[str, Any]: + """Resolved self-service settings; disabled when block missing or USE_SELFSERVICE false.""" + block = config.get("SELF_SERVICE") or {} + if not isinstance(block, dict): + block = {} + enabled = bool(block.get("USE_SELFSERVICE", block.get("ENABLED", False))) + return { + "enabled": enabled, + "db_server": str(block.get("DB_SERVER", "localhost")), + "db_username": str(block.get("DB_USERNAME", "")), + "db_password": str(block.get("DB_PASSWORD", "")), + "db_name": str(block.get("DB_NAME", "")), + "db_port": int(block.get("DB_PORT", 3306)), + "pbkdf2_salt": str(block.get("PBKDF2_SALT", "ADN")), + "pbkdf2_iterations": int(block.get("PBKDF2_ITERATIONS", 2000)), + } diff --git a/src/adn_server/infrastructure/proxy/udp_fanin.py b/src/adn_server/infrastructure/proxy/udp_fanin.py index 4d6a47a..f409f8d 100644 --- a/src/adn_server/infrastructure/proxy/udp_fanin.py +++ b/src/adn_server/infrastructure/proxy/udp_fanin.py @@ -4,7 +4,7 @@ from __future__ import annotations import logging from collections.abc import Callable -from typing import Any, Protocol +from typing import TYPE_CHECKING, Any, Protocol from twisted.internet.protocol import DatagramProtocol @@ -12,6 +12,9 @@ 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 +if TYPE_CHECKING: + from .self_service_bridge import ProxySelfServiceBridge + _logger = logging.getLogger(__name__) @@ -31,12 +34,14 @@ class ProxyFanInProtocol(DatagramProtocol): debug: bool = False, logger: logging.Logger | None = None, on_attached: Callable[[bytes, str, int, bool], None] | None = None, + self_service: ProxySelfServiceBridge | None = None, ) -> None: self._proxy = proxy self._master_sink = master_sink self.debug = debug self._log = logger or _logger self._on_attached = on_attached + self._self_service = self_service def datagramReceived(self, data: bytes, addr: tuple[str, int]) -> None: host, port = addr @@ -72,6 +77,10 @@ class ProxyFanInProtocol(DatagramProtocol): return if self._on_attached is not None: self._on_attached(peer_id, host, port, new_session) + if self._self_service is not None and self._self_service.before_inject( + data, addr, peer_id + ): + return self._master_sink.inject(data, addr) diff --git a/tests/infrastructure/test_proxy_self_service.py b/tests/infrastructure/test_proxy_self_service.py new file mode 100644 index 0000000..c07c157 --- /dev/null +++ b/tests/infrastructure/test_proxy_self_service.py @@ -0,0 +1,178 @@ +"""Integrated proxy self-service DB hooks (legacy adn-proxy parity).""" + +from __future__ import annotations + +from twisted.internet.defer import Deferred + +from adn_server.application.proxy import ProxyUseCases +from adn_server.domain import bytes_4 +from adn_server.domain.proxy import ClientEndpoint +from adn_server.infrastructure.hbp_constants import RPTC, RPTO +from adn_server.infrastructure.proxy.null_self_service import NullProxySelfServiceStore +from adn_server.infrastructure.proxy.rpto_queue import InMemoryPendingRptoQueue +from adn_server.infrastructure.proxy.self_service_bridge import ProxySelfServiceBridge +from adn_server.infrastructure.proxy.self_service_config import self_service_settings +from adn_server.infrastructure.proxy.slot_store import InMemoryProxySlotStore + + +class _RecordingSink: + def __init__(self) -> None: + self.injected: list[tuple[bytes, tuple[str, int]]] = [] + + def inject(self, data: bytes, client_addr: tuple[str, int]) -> None: + self.injected.append((data, client_addr)) + + +class _RecordingSender: + def __init__(self) -> None: + self.sent: list[tuple[bytes, ClientEndpoint]] = [] + + def send_to_client(self, data: bytes, client: ClientEndpoint) -> None: + self.sent.append((data, client)) + + +class _FakeStore(NullProxySelfServiceStore): + def __init__(self) -> None: + self.actions: list[tuple[str, bytes]] = [] + self.options_by_peer: dict[bytes, str] = {} + self.pending_modified: list[tuple[bytes, str]] = [] + + def ins_conf( + self, + int_id: int, + peer_id_bytes: bytes, + callsign: str, + host: str, + mode: str, + ) -> None: + self.actions.append(("ins_conf", peer_id_bytes)) + + def updt_tbl( + self, + action: str, + peer_id_bytes: bytes, + *, + psswd: str | None = None, + ) -> None: + self.actions.append((action, peer_id_bytes)) + + def slct_opt(self, peer_id_bytes: bytes): + from twisted.internet.defer import succeed + + opt = self.options_by_peer.get(peer_id_bytes) + if not opt: + return succeed([]) + return succeed([(opt,)]) + + def slct_db(self): + from twisted.internet.defer import succeed + + return succeed([(pid, opt) for pid, opt in self.pending_modified]) + + +def _bridge() -> tuple[ProxySelfServiceBridge, _RecordingSink, _FakeStore, _RecordingSender]: + store = _FakeStore() + sink = _RecordingSink() + sender = _RecordingSender() + use_cases = ProxyUseCases(InMemoryProxySlotStore(), InMemoryPendingRptoQueue(), max_peers=4) + peer = bytes_4(7300444) + use_cases.attach_client(peer, "192.168.1.10", 62031) + bridge = ProxySelfServiceBridge( + store, + use_cases, + sink, + sender, + pbkdf2_salt="ADN", + pbkdf2_iterations=2000, + ) + return bridge, sink, store, sender + + +def _run_deferred(d: Deferred) -> None: + results: list[object] = [] + d.addBoth(lambda x: results.append(x) or x) + assert results, "deferred did not fire synchronously" + + +def test_self_service_settings_reads_monitor_keys() -> None: + cfg = { + "SELF_SERVICE": { + "USE_SELFSERVICE": True, + "DB_SERVER": "localhost", + "DB_USERNAME": "hbmon", + "DB_PASSWORD": "secret", + "DB_NAME": "hbmon", + "DB_PORT": 3306, + "PBKDF2_SALT": "ADN", + "PBKDF2_ITERATIONS": 2000, + } + } + ss = self_service_settings(cfg) + assert ss["enabled"] is True + assert ss["db_name"] == "hbmon" + assert ss["pbkdf2_iterations"] == 2000 + + +def test_rptc_ins_conf_and_login_opt_injects_rpto() -> None: + bridge, sink, store, _sender = _bridge() + peer = bytes_4(7300444) + store.options_by_peer[peer] = "TS2=730444;" + packet = RPTC + peer + b"CE1ILI " + b"\x00" * 85 + b"4" + bridge.before_inject(packet, ("192.168.1.10", 62031), peer) + assert ("ins_conf", peer) in store.actions + _run_deferred(bridge._login_opt(peer)) + assert sink.injected + assert sink.injected[0][0] == RPTO + peer + b"TS2=730444;" + + +def test_rpto_pass_stores_password_and_skips_inject() -> None: + bridge, sink, store, sender = _bridge() + peer = bytes_4(7300444) + packet = RPTO + peer + b"PASS=secret123" + skip = bridge.before_inject(packet, ("192.168.1.10", 62031), peer) + assert skip is True + assert sink.injected == [] + assert ("psswd", peer) in store.actions + assert sender.sent and sender.sent[0][0][:6] == b"RPTACK" + + +def test_send_opts_pushes_modified_rows_to_master() -> None: + bridge, sink, store, _sender = _bridge() + peer = bytes_4(7300444) + store.pending_modified = [(peer, "TS2=730444;")] + _run_deferred(bridge.send_opts()) + assert ("rst_mod", peer) in store.actions + assert sink.injected[0][0] == RPTO + peer + b"TS2=730444;" + + +def test_session_expired_logs_out() -> None: + bridge, _sink, store, _sender = _bridge() + peer = bytes_4(7300444) + bridge.on_session_expired(peer) + assert ("log_out", peer) in store.actions + + +def test_yaml_loader_preserves_self_service_block(tmp_path) -> None: + """SELF_SERVICE from adn-server.yaml must reach runtime (not stripped at load).""" + from adn_server.infrastructure.config_loader import YamlConfigLoader + + cfg_path = tmp_path / "adn-server.yaml" + cfg_path.write_text( + """ +GLOBAL: {} +PROXY: + LISTEN_PORT: 62031 + TARGET_SYSTEM: SYSTEM +SELF_SERVICE: + USE_SELFSERVICE: true + DB_SERVER: localhost +SYSTEMS: + SYSTEM: + MODE: MASTER + ENABLED: true +""", + encoding="utf-8", + ) + config = YamlConfigLoader(tmp_path).load(str(cfg_path)) + assert config["SELF_SERVICE"]["USE_SELFSERVICE"] is True + assert self_service_settings(config)["enabled"] is True