KB8PMY
/
ADN-DMR-Peer-Server
mirror of https://github.com/Amateur-Digital-Network/ADN-DMR-Peer-Server.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
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.
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '5c24ecb48b'
${ noResults }
ADN-DMR-Peer-Server/tests/infrastructure/test_obp_replay.py

401 lines
14 KiB
Raw Blame History

# ADN DMR Peer Server - tests infrastructure obp replay
#
# 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
###############################################################################
"""Replaying a capture: read the pcap, ask the engine, report the verdict."""
from __future__ import annotations
import struct
import time
import pytest
from adn_server.domain import bytes_3, bytes_4
from adn_server.infrastructure.hbp_constants import BCKA, DMRD
from adn_server.infrastructure.mesh.dmre_v5 import build_dmre
from adn_server.infrastructure.mesh.obp_v1 import build_bcka, build_dmrd_v1
from adn_server.infrastructure.obp_replay import format_report, replay, run_replay
from adn_server.infrastructure.pcap import CaptureError, read_udp
_PASSPHRASE = (b"test-passphrase" + b"\x00" * 20)[:20]
_FR = ("82.65.127.86", 62201)
_PT = ("85.241.222.7", 62268)
_NOW = 1_800_000_000.0
# --- building a capture on the fly -------------------------------------------
def _udp_packet(source: tuple[str, int], destination: tuple[str, int], payload: bytes) -> bytes:
src_ip = bytes(int(part) for part in source[0].split("."))
dst_ip = bytes(int(part) for part in destination[0].split("."))
udp = struct.pack(">HHHH", source[1], destination[1], 8 + len(payload), 0) + payload
total = 20 + len(udp)
ip = struct.pack(">BBHHHBBH", 0x45, 0, total, 0, 0, 64, 17, 0) + src_ip + dst_ip
ethernet = b"\x02" * 6 + b"\x03" * 6 + b"\x08\x00"
return ethernet + ip + udp
def write_pcap(path, frames: list[tuple[float, tuple, tuple, bytes]]) -> None:
with open(path, "wb") as fh:
fh.write(struct.pack("<IHHiIII", 0xA1B2C3D4, 2, 4, 0, 0, 262144, 1))
for when, source, destination, payload in frames:
packet = _udp_packet(source, destination, payload)
fh.write(struct.pack("<IIII", int(when), int(when % 1 * 1_000_000), len(packet), len(packet)))
fh.write(packet)
def _voice(*, src: int = 2130001, dst: int = 214, bits: int = 0x21, stream: int = 0x1234) -> bytes:
body = b"".join(
[
DMRD,
bytes([1]),
bytes_3(src),
bytes_3(dst),
bytes_4(20840),
bytes([bits]),
bytes_4(stream),
b"\x00" * 33,
]
)
return build_dmrd_v1(body, bytes_4(20840), _PASSPHRASE)
def _voice_v5(*, dst: int = 214, age: float = 0.0, now: float = _NOW) -> bytes:
body = b"".join(
[
DMRD,
bytes([1]),
bytes_3(2130001),
bytes_3(dst),
bytes_4(20840),
bytes([0x21]),
bytes_4(0x4321),
b"\x00" * 33,
]
)
packet = build_dmre(
body,
server_id=bytes_4(20840),
ber=b"\x00",
rssi=b"\x00",
embedded_ver=5,
timestamp_ns=int((now - age) * 1_000_000_000),
source_server=bytes_4(2084),
source_rptr=bytes_4(0),
hops=b"\x00",
passphrase=_PASSPHRASE,
extended_layout=True,
)
assert packet is not None
return packet
def _config() -> dict:
return {
"GLOBAL": {
"SERVER_ID": bytes_4(21310),
"USE_ACL": False,
"SUB_ACL": (True, []),
"TG1_ACL": (True, []),
},
"SYSTEMS": {
"OBP-FR": {
"MODE": "OPENBRIDGE",
"ENABLED": True,
"NETWORK_ID": bytes_4(20840),
"PASSPHRASE": _PASSPHRASE,
"TARGET_IP": _FR[0],
"TARGET_PORT": _FR[1],
"TARGET_SOCK": _FR,
"RELAX_CHECKS": False,
"VER": 1,
"ENHANCED_OBP": True,
"_REPORT_PORT": 62201,
},
"HOTSPOT": {"MODE": "MASTER", "ENABLED": True},
},
"_SERVER_IDS": set(),
}
# --- the pcap reader ----------------------------------------------------------
def test_reading_udp_out_of_a_capture(tmp_path) -> None:
capture = tmp_path / "obp.pcap"
write_pcap(capture, [(_NOW, _FR, ("10.0.0.1", 62201), b"hello")])
datagrams = list(read_udp(capture))
assert len(datagrams) == 1
assert datagrams[0].source == _FR
assert datagrams[0].destination == ("10.0.0.1", 62201)
assert datagrams[0].payload == b"hello"
assert int(datagrams[0].timestamp) == int(_NOW)
def test_a_file_that_is_not_a_capture(tmp_path) -> None:
path = tmp_path / "nope.pcap"
path.write_bytes(b"not a capture at all, really" * 2)
with pytest.raises(CaptureError):
list(read_udp(path))
def test_pcapng_says_how_to_convert_it(tmp_path) -> None:
path = tmp_path / "new.pcapng"
path.write_bytes(b"\x0a\x0d\x0d\x0a" + b"\x00" * 40)
with pytest.raises(CaptureError, match="editcap"):
list(read_udp(path))
def test_reading_a_linux_cooked_v2_capture(tmp_path) -> None:
"""``tcpdump -i any`` on a recent libpcap writes SLL2, not Ethernet."""
capture = tmp_path / "any.pcap"
packet = _udp_packet(_FR, ("10.0.0.1", 62201), b"hello")[14:] # drop the ethernet header
sll2 = struct.pack(">HHIHBB", 0x0800, 0, 2, 1, 0, 6) + b"\x02" * 8
with open(capture, "wb") as fh:
fh.write(struct.pack("<IHHiIII", 0xA1B2C3D4, 2, 4, 0, 0, 262144, 276))
frame = sll2 + packet
fh.write(struct.pack("<IIII", int(_NOW), 0, len(frame), len(frame)))
fh.write(frame)
datagrams = list(read_udp(capture))
assert [d.payload for d in datagrams] == [b"hello"]
assert datagrams[0].source == _FR
def _write_raw(path, linktype: int, frame: bytes) -> None:
with open(path, "wb") as fh:
fh.write(struct.pack("<IHHiIII", 0xA1B2C3D4, 2, 4, 0, 0, 262144, linktype))
fh.write(struct.pack("<IIII", int(_NOW), 0, len(frame), len(frame)))
fh.write(frame)
def _ip_udp() -> bytes:
return _udp_packet(_FR, ("10.0.0.1", 62201), b"hello")[14:]
def test_reading_a_raw_ip_capture(tmp_path) -> None:
capture = tmp_path / "raw.pcap"
_write_raw(capture, 101, _ip_udp())
assert [d.payload for d in read_udp(capture)] == [b"hello"]
def test_reading_a_loopback_capture(tmp_path) -> None:
capture = tmp_path / "null.pcap"
_write_raw(capture, 0, struct.pack("<I", 2) + _ip_udp())
assert [d.payload for d in read_udp(capture)] == [b"hello"]
def test_reading_a_vlan_tagged_frame(tmp_path) -> None:
capture = tmp_path / "vlan.pcap"
tagged = b"\x02" * 6 + b"\x03" * 6 + b"\x81\x00" + b"\x00\x64" + b"\x08\x00" + _ip_udp()
_write_raw(capture, 1, tagged)
assert [d.payload for d in read_udp(capture)] == [b"hello"]
def test_reading_an_ipv6_datagram(tmp_path) -> None:
capture = tmp_path / "v6.pcap"
payload = b"hello"
udp = struct.pack(">HHHH", _FR[1], 62201, 8 + len(payload), 0) + payload
header = struct.pack(">IHBB", 0x60000000, len(udp), 17, 64)
source = bytes.fromhex("20010db8000000000000000000000001")
destination = bytes.fromhex("20010db8000000000000000000000002")
_write_raw(capture, 101, header + source + destination + udp)
datagrams = list(read_udp(capture))
assert datagrams[0].payload == payload
assert datagrams[0].source == ("2001:db8:0:0:0:0:0:1", _FR[1])
def test_frames_that_are_not_udp_are_skipped(tmp_path) -> None:
capture = tmp_path / "tcp.pcap"
packet = bytearray(_ip_udp())
packet[9] = 6 # TCP
_write_raw(capture, 101, bytes(packet))
assert list(read_udp(capture)) == []
def test_an_unknown_link_layer_is_skipped(tmp_path) -> None:
capture = tmp_path / "weird.pcap"
_write_raw(capture, 999, _ip_udp())
assert list(read_udp(capture)) == []
def test_a_truncated_record_ends_the_walk(tmp_path) -> None:
capture = tmp_path / "cut.pcap"
_write_raw(capture, 101, _ip_udp())
with open(capture, "ab") as fh:
fh.write(struct.pack("<IIII", int(_NOW), 0, 200, 200) + b"\x45" * 10)
assert len(list(read_udp(capture))) == 1
# --- the replay ---------------------------------------------------------------
def _replay(frames: list[tuple], config: dict | None = None, **kwargs):
from adn_server.infrastructure.pcap import CapturedDatagram
datagrams = [
CapturedDatagram(timestamp=when, source=src, destination=dst, payload=payload)
for when, src, dst, payload in frames
]
return replay(config or _config(), datagrams, **kwargs)
def test_a_good_frame_from_the_configured_peer_is_delivered() -> None:
verdicts = _replay([(_NOW, _FR, ("10.0.0.1", 62201), _voice())])
assert [v.outcome for v in verdicts] == ["delivered"]
assert verdicts[0].system == "OBP-FR"
assert (verdicts[0].rf_src, verdicts[0].dst_id) == (2130001, 214)
assert verdicts[0].kind == "DMRD v1"
def test_a_local_talkgroup_is_reported_with_its_reason_and_quench() -> None:
verdicts = _replay([(_NOW, _FR, ("10.0.0.1", 62201), _voice(dst=9))])
assert verdicts[0].outcome == "dropped"
assert verdicts[0].reason == "tg-filter"
assert verdicts[0].quench is True
def test_a_frame_from_an_unexpected_address_is_refused_when_checks_are_strict() -> None:
verdicts = _replay([(_NOW, ("9.9.9.9", 40000), ("10.0.0.1", 62201), _voice())])
assert verdicts[0].outcome == "refused"
assert "source" in verdicts[0].reason
def test_a_frame_no_bridge_can_verify_is_left_unmatched() -> None:
config = _config()
config["SYSTEMS"]["OBP-FR"]["PASSPHRASE"] = (b"another" + b"\x00" * 20)[:20]
verdicts = _replay([(_NOW, _FR, ("10.0.0.1", 62201), _voice())], config)
assert verdicts[0].outcome == "unmatched"
assert verdicts[0].system is None
def test_a_v5_frame_is_replayed_with_the_time_it_was_captured() -> None:
config = _config()
config["SYSTEMS"]["OBP-FR"]["VER"] = 5
verdicts = _replay([(_NOW, _FR, ("10.0.0.1", 62201), _voice_v5())], config)
assert verdicts[0].kind == "DMRE v5"
assert verdicts[0].outcome == "delivered"
def test_a_v5_frame_that_was_already_late_when_captured_is_dropped() -> None:
config = _config()
config["SYSTEMS"]["OBP-FR"]["VER"] = 5
verdicts = _replay([(_NOW, _FR, ("10.0.0.1", 62201), _voice_v5(age=9.0))], config)
assert verdicts[0].reason == "stale-packet"
def test_a_v1_frame_on_a_v5_link_asks_the_peer_for_its_version() -> None:
config = _config()
config["SYSTEMS"]["OBP-FR"]["VER"] = 5
verdicts = _replay([(_NOW, _FR, ("10.0.0.1", 62201), _voice())], config)
assert verdicts[0].reason == "proto-version"
def test_control_frames_are_named_not_judged() -> None:
verdicts = _replay([(_NOW, _FR, ("10.0.0.1", 62201), build_bcka(_PASSPHRASE))])
assert verdicts[0].kind == "BCKA"
assert verdicts[0].outcome == "control"
assert verdicts[0].datagram.payload[:4] == BCKA
def test_the_bridge_is_chosen_by_the_port_the_frame_arrived_on() -> None:
"""Two bridges, one passphrase: the local port is the evidence that tells them apart."""
config = _config()
config["SYSTEMS"]["OBP-PT"] = {
**config["SYSTEMS"]["OBP-FR"],
"TARGET_IP": _PT[0],
"TARGET_PORT": _PT[1],
"TARGET_SOCK": _PT,
"RELAX_CHECKS": True,
"_REPORT_PORT": 62268,
}
verdicts = _replay([(_NOW, ("203.0.113.9", 50000), ("10.0.0.1", 62268), _voice())], config)
assert verdicts[0].system == "OBP-PT"
def test_only_the_system_that_was_asked_for() -> None:
config = _config()
config["SYSTEMS"]["OBP-PT"] = {**config["SYSTEMS"]["OBP-FR"], "_REPORT_PORT": 62268}
verdicts = _replay([(_NOW, _FR, ("10.0.0.1", 62201), _voice())], config, system="OBP-PT")
assert verdicts[0].system == "OBP-PT"
with pytest.raises(KeyError):
_replay([], config, system="OBP-NOPE")
def test_the_report_tallies_what_happened() -> None:
verdicts = _replay(
[
(_NOW, _FR, ("10.0.0.1", 62201), _voice(stream=1)),
(_NOW, _FR, ("10.0.0.1", 62201), _voice(dst=9, stream=2)),
(_NOW, _FR, ("10.0.0.1", 62201), _voice(dst=9990, stream=3)),
]
)
report = format_report(verdicts, capture="obp.pcap")
assert "3 datagram(s)" in report
assert "OBP-FR" in report
assert "delivered" in report
assert "dropped: tg-filter" in report
def test_the_report_can_skip_the_per_frame_lines() -> None:
verdicts = _replay([(_NOW, _FR, ("10.0.0.1", 62201), _voice())])
summary = format_report(verdicts, capture="obp.pcap", verbose=False)
assert "delivered" in summary
assert time.strftime("%H:%M:%S", time.localtime(_NOW)) not in summary
# --- the command ---------------------------------------------------------------
def test_run_replay_prints_a_report(tmp_path, capsys) -> None:
capture = tmp_path / "obp.pcap"
write_pcap(
capture,
[
(_NOW, _FR, ("10.0.0.1", 62201), _voice(stream=1)),
(_NOW + 1, _FR, ("10.0.0.1", 62201), _voice(dst=9, stream=2)),
],
)
assert run_replay(_config(), str(capture)) == 0
printed = capsys.readouterr().out
assert "2 datagram(s)" in printed
assert "tg-filter" in printed
def test_run_replay_stops_at_the_limit(tmp_path, capsys) -> None:
capture = tmp_path / "obp.pcap"
write_pcap(capture, [(_NOW + i, _FR, ("10.0.0.1", 62201), _voice(stream=i)) for i in range(5)])
assert run_replay(_config(), str(capture), limit=2) == 0
assert "2 datagram(s)" in capsys.readouterr().out
def test_run_replay_reports_a_bad_capture(tmp_path, capsys) -> None:
path = tmp_path / "broken.pcap"
path.write_bytes(b"x" * 64)
assert run_replay(_config(), str(path)) == 1
assert "ERROR capture" in capsys.readouterr().err
def test_run_replay_reports_an_unknown_system(tmp_path, capsys) -> None:
capture = tmp_path / "obp.pcap"
write_pcap(capture, [(_NOW, _FR, ("10.0.0.1", 62201), _voice())])
assert run_replay(_config(), str(capture), system="NOPE") == 1
assert "ERROR system" in capsys.readouterr().err
Reference in new issue View Git Blame Copy Permalink

Powered by TurnKey Linux.