From f16ba9de2b1add7ae2db06b8168b40f668279e06 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rodrigo=20P=C3=A9rez?= Date: Thu, 9 Jul 2026 17:04:29 -0400 Subject: [PATCH] fix: OBP DMRE source-server validation without ALLOW_UNREG_ID bypass Port OPENBRIDGE.validate_id lookup for 6-7 digit source servers so OBP ingress matches legacy production config (VALIDATE_SERVER_IDS=True). --- .../twisted_adapters/udp_hbp.py | 21 +++++- .../test_obp_validate_source_server.py | 65 +++++++++++++++++++ 2 files changed, 85 insertions(+), 1 deletion(-) create mode 100644 tests/infrastructure/test_obp_validate_source_server.py diff --git a/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py b/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py index c4c9be6..9ec8b64 100644 --- a/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py +++ b/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py @@ -975,6 +975,25 @@ class HBPProtocol(DatagramProtocol): return _peer_ids[_int_peer_id] return False + def validate_obp_source_server_id(self, peer_id: bytes): + """OPENBRIDGE DMRE source-server lookup (legacy hblink.py OPENBRIDGE.validate_id). + + Unlike ``validate_id`` on HBP systems, OBP ingress always resolves 6–7 digit + source servers against alias tables with no ``ALLOW_UNREG_ID`` bypass. + Returns a callsign string on match, or ``False``. + """ + _int_peer_id = int(int_id(peer_id)) + _local_subscriber_ids = self._CONFIG.get("_LOCAL_SUBSCRIBER_IDS", {}) + _subscriber_ids = self._CONFIG.get("_SUB_IDS", {}) + _peer_ids = self._CONFIG.get("_PEER_IDS", {}) + if _int_peer_id in _local_subscriber_ids: + return _local_subscriber_ids[_int_peer_id] + if _int_peer_id in _subscriber_ids: + return _subscriber_ids[_int_peer_id] + if _int_peer_id in _peer_ids: + return _peer_ids[_int_peer_id] + return False + def proxy_IPBlackList(self, peer_id: bytes, sockaddr: tuple[str, int]) -> None: """Legacy hblink.py proxy_IPBlackList: send PRBL to proxy to blacklist a peer's IP for 5 min.""" _bltime = str(time.time() + 300) @@ -2131,7 +2150,7 @@ class HBPProtocol(DatagramProtocol): self._obp_send_bcsq(_dst_id, _stream_id) self._laststrid.append(_stream_id) return - if _src_srv_len > 5 and not self.validate_id(_source_server): + if _src_srv_len > 5 and not self.validate_obp_source_server_id(_source_server): if _stream_id not in self._laststrid: logger.warning("(%s) Source Server 6 or 7 digits but not a valid DMR ID, discarding Src: %s", self._system, _src_srv_int) self._obp_send_bcsq(_dst_id, _stream_id) diff --git a/tests/infrastructure/test_obp_validate_source_server.py b/tests/infrastructure/test_obp_validate_source_server.py new file mode 100644 index 0000000..829356b --- /dev/null +++ b/tests/infrastructure/test_obp_validate_source_server.py @@ -0,0 +1,65 @@ +# ADN DMR Peer Server - tests infrastructure obp validate source server +# +# Copyright (C) 2026 Rodrigo Pérez, CE5RPY +# +############################################################################### +# 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 +############################################################################### + +"""OBP DMRE source-server validation must not use HBP ALLOW_UNREG_ID bypass.""" + +from __future__ import annotations + +from adn_server.domain import bytes_4 +from adn_server.infrastructure.twisted_adapters.udp_hbp import HBPProtocol + +_UNKNOWN = bytes_4(3120999) +_KNOWN = bytes_4(3120001) + + +def _obp_protocol(*, allow_unreg: bool | None = None) -> HBPProtocol: + system = { + "MODE": "OPENBRIDGE", + "PASSPHRASE": b"test-passphrase\x00\x00\x00\x00\x00\x00", + "VER": 5, + "TARGET_IP": "127.0.0.1", + "TARGET_PORT": 62030, + "NETWORK_ID": bytes_4(73010), + } + if allow_unreg is not None: + system["ALLOW_UNREG_ID"] = allow_unreg + config = { + "GLOBAL": {"SERVER_ID": bytes_4(73010)}, + "_SUB_IDS": {3120001: "CE1TST"}, + "SYSTEMS": {"OBP-A": system}, + } + return HBPProtocol("OBP-A", config) + + +def test_obp_source_server_rejects_unknown_even_without_allow_unreg_id() -> None: + proto = _obp_protocol() + assert proto.validate_id(_UNKNOWN) is True + assert proto.validate_obp_source_server_id(_UNKNOWN) is False + + +def test_obp_source_server_accepts_known_subscriber_id() -> None: + proto = _obp_protocol() + assert proto.validate_obp_source_server_id(_KNOWN) == "CE1TST" + + +def test_obp_source_server_still_rejects_when_allow_unreg_disabled() -> None: + proto = _obp_protocol(allow_unreg=False) + assert proto.validate_id(_UNKNOWN) is False + assert proto.validate_obp_source_server_id(_UNKNOWN) is False