You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
135 lines
5.3 KiB
135 lines
5.3 KiB
# ADN DMR Peer Server - tests hbp ACL gate
|
|
#
|
|
# Copyright (C) 2026 Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
|
|
#
|
|
###############################################################################
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation; either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program; if not, write to the Free Software Foundation,
|
|
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
|
|
###############################################################################
|
|
|
|
"""One ACL gate, shared by the MASTER and PEER DMRD ingress paths."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import replace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from tests.harness.deterministic import PacketSpec
|
|
from tests.support.hbp_repeat_stack import RecordingTransport, build_hbp_repeat_stack
|
|
|
|
from adn_server.domain import bytes_3, bytes_4
|
|
from adn_server.infrastructure.acl_router import InMemoryAclRouter
|
|
from adn_server.infrastructure.twisted_adapters.udp_hbp import HBPProtocol
|
|
|
|
_PEER = (1234567).to_bytes(4, "big")
|
|
_ADDR = ("10.0.0.9", 54321)
|
|
_RF_SRC = 3120001
|
|
_DENY_SRC = (False, [(_RF_SRC, _RF_SRC)])
|
|
_PERMIT_ALL = (True, [(1, 4294967295)])
|
|
_NO_STREAM = b"\x00" # _make_slot_status seed: no stream seen on this slot yet
|
|
|
|
|
|
def _master_stack(**global_acl: Any):
|
|
stack = build_hbp_repeat_stack()
|
|
# the shared stack ships a permit-everything router; these tests need the real one
|
|
stack.hbp._router = InMemoryAclRouter()
|
|
# the first thing ingress does after the gate: a frame that got through is in SUB_MAP
|
|
stack.config["_SUB_MAP"] = {}
|
|
stack.config["GLOBAL"].update(global_acl)
|
|
stack.register_peer(_PEER, _ADDR)
|
|
return stack
|
|
|
|
|
|
def _voice_head(tg: int, slot: int, stream_id: int, *, call_type: str = "group") -> bytes:
|
|
return replace(
|
|
PacketSpec(
|
|
dst_id=tg, slot=slot, stream_id=stream_id, call_type=call_type,
|
|
peer_id=int.from_bytes(_PEER, "big"), rf_src=_RF_SRC,
|
|
),
|
|
frame_type=2, dtype_vseq=1,
|
|
).data()
|
|
|
|
|
|
def test_master_global_subscriber_acl_drops_and_logs_once_per_stream(
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
stack = _master_stack(USE_ACL=True, SUB_ACL=_DENY_SRC, TG1_ACL=_PERMIT_ALL, TG2_ACL=_PERMIT_ALL)
|
|
with caplog.at_level(logging.INFO):
|
|
for _ in range(3):
|
|
stack.hbp._master_datagram_received(_voice_head(214, 2, 0x44440001), _ADDR)
|
|
|
|
assert stack.config["_SUB_MAP"] == {}
|
|
dropped = [r for r in caplog.records if "BY GLOBAL ACL" in r.getMessage()]
|
|
assert len(dropped) == 1
|
|
assert "FROM SUBSCRIBER 3120001" in dropped[0].getMessage()
|
|
|
|
|
|
def test_master_system_tg_acl_applies_to_the_frame_slot_only() -> None:
|
|
stack = _master_stack()
|
|
sys_cfg = stack.config["SYSTEMS"][stack.system_name]
|
|
sys_cfg["USE_ACL"] = True
|
|
sys_cfg["SUB_ACL"] = _PERMIT_ALL
|
|
sys_cfg["TG1_ACL"] = (False, [(214, 214)])
|
|
sys_cfg["TG2_ACL"] = _PERMIT_ALL
|
|
|
|
stack.hbp._master_datagram_received(_voice_head(214, 1, 0x44440002), _ADDR)
|
|
assert stack.config["_SUB_MAP"] == {}
|
|
|
|
stack.hbp._master_datagram_received(_voice_head(214, 2, 0x44440003), _ADDR)
|
|
assert bytes_3(_RF_SRC) in stack.config["_SUB_MAP"]
|
|
|
|
|
|
def test_on_demand_unit_service_skips_tg_acl_but_not_subscriber_acl() -> None:
|
|
"""Legacy exemption: service destinations bypass the TG lists, never SUB_ACL."""
|
|
stack = _master_stack(USE_ACL=True, SUB_ACL=_PERMIT_ALL, TG1_ACL=(False, [(1, 4294967295)]),
|
|
TG2_ACL=(False, [(1, 4294967295)]))
|
|
assert not stack.hbp._acl_rejects_dmrd(
|
|
bytes_4(_RF_SRC)[1:], bytes_4(9990)[1:], 2, b"\x44\x44\x00\x04", True,
|
|
)
|
|
stack.config["GLOBAL"]["SUB_ACL"] = _DENY_SRC
|
|
assert stack.hbp._acl_rejects_dmrd(
|
|
bytes_4(_RF_SRC)[1:], bytes_4(9990)[1:], 2, b"\x44\x44\x00\x05", True,
|
|
)
|
|
|
|
|
|
def test_peer_mode_ingress_uses_the_same_gate() -> None:
|
|
"""PEER mode (client of another master) rejects on the very same rules."""
|
|
master_sockaddr = ("203.0.113.7", 62031)
|
|
config: dict[str, Any] = {
|
|
"GLOBAL": {"USE_ACL": True, "SUB_ACL": _DENY_SRC, "TG1_ACL": _PERMIT_ALL,
|
|
"TG2_ACL": _PERMIT_ALL},
|
|
"SYSTEMS": {
|
|
"PEER-1": {
|
|
"MODE": "PEER",
|
|
"ENABLED": True,
|
|
"LOOSE": True,
|
|
"RADIO_ID": _PEER,
|
|
"MASTER_SOCKADDR": master_sockaddr,
|
|
"USE_ACL": False,
|
|
},
|
|
},
|
|
}
|
|
hbp = HBPProtocol("PEER-1", config, router=InMemoryAclRouter())
|
|
hbp.transport = RecordingTransport() # type: ignore[assignment]
|
|
|
|
hbp._peer_datagram_received(_voice_head(214, 2, 0x44440006), master_sockaddr)
|
|
assert hbp.STATUS[2].get("RX_STREAM_ID") == _NO_STREAM
|
|
|
|
config["GLOBAL"]["SUB_ACL"] = _PERMIT_ALL
|
|
hbp._peer_datagram_received(_voice_head(214, 2, 0x44440007), master_sockaddr)
|
|
assert hbp.STATUS[2].get("RX_STREAM_ID") == (0x44440007).to_bytes(4, "big")
|