Port legacy adn-proxy Clients table behaviour into core PROXY fan-in: RPTC/RPTO hooks, periodic send_opts/lst_seen/clean_tbl, and SELF_SERVICE YAML block (load via config_loader). Optional mysqlclient extra.pull/1/head
Rodrigo Pérez
4 months ago
parent
18dc22b0bb
commit
bba7e43da8
@ -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)
|
||||
@ -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"]
|
||||
@ -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)
|
||||
@ -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()))
|
||||
@ -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()
|
||||
@ -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)),
|
||||
}
|
||||
@ -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
|
||||
Loading…
Reference in new issue