diff --git a/src/adn_server/infrastructure/mesh/__init__.py b/src/adn_server/infrastructure/mesh/__init__.py new file mode 100644 index 0000000..9664704 --- /dev/null +++ b/src/adn_server/infrastructure/mesh/__init__.py @@ -0,0 +1 @@ +# OpenBridge / DMRE wire codecs (V2-P0-005). No Twisted imports. diff --git a/src/adn_server/infrastructure/mesh/dmre_v5.py b/src/adn_server/infrastructure/mesh/dmre_v5.py new file mode 100644 index 0000000..5a5fe25 --- /dev/null +++ b/src/adn_server/infrastructure/mesh/dmre_v5.py @@ -0,0 +1,132 @@ +"""FreeBridge DMRE v4/v5: BLAKE2b wire build and parse (no Twisted).""" + +from __future__ import annotations + +from dataclasses import dataclass +from hashlib import blake2b +from hmac import compare_digest + +from ..hbp_constants import DMRE + +DMRE_DMR_PAYLOAD_LEN = 53 +DMRE_MAC_LEN = 16 + + +@dataclass(frozen=True) +class DmreTrailer: + """Fields after the 53-byte DMR payload in a DMRE datagram.""" + + embedded_version: int + ber: bytes + rssi: bytes + timestamp: bytes + source_server: bytes + source_rptr: bytes + hops: bytes + hash_len: int # authenticated prefix length (MAC starts here) + + +def dmre_blake2b_mac(packet: bytes, passphrase: bytes, hash_len: int) -> bytes: + h = blake2b(key=passphrase, digest_size=DMRE_MAC_LEN) + h.update(packet[:hash_len]) + return h.digest() + + +def verify_dmre_mac(packet: bytes, passphrase: bytes, hash_len: int) -> bool: + if len(packet) < hash_len + DMRE_MAC_LEN: + return False + expected = packet[hash_len : hash_len + DMRE_MAC_LEN] + return compare_digest(expected, dmre_blake2b_mac(packet, passphrase, hash_len)) + + +def parse_dmre_trailer(packet: bytes) -> DmreTrailer | None: + """Parse DMRE trailer; returns None if packet is too short or not DMRE.""" + if packet[:4] != DMRE or len(packet) < 69: + return None + embedded_version = packet[55] + ber = packet[53:54] + rssi = packet[54:55] + timestamp = packet[56:64] + if embedded_version > 4: + if len(packet) < 89: + return None + source_server = packet[64:68] + source_rptr = packet[68:72] + hops = packet[72:73] + hash_len = 73 + else: + if len(packet) < 85: + return None + source_server = packet[64:68] + source_rptr = b"\x00\x00\x00\x00" + hops = packet[68:69] + hash_len = 69 + return DmreTrailer( + embedded_version=embedded_version, + ber=ber, + rssi=rssi, + timestamp=timestamp, + source_server=source_server, + source_rptr=source_rptr, + hops=hops, + hash_len=hash_len, + ) + + +def build_dmre( + dmr_packet: bytes, + *, + server_id: bytes, + ber: bytes, + rssi: bytes, + embedded_ver: int, + timestamp_ns: int, + source_server: bytes, + source_rptr: bytes, + hops: bytes, + passphrase: bytes, + extended_layout: bool, +) -> bytes | None: + """Build DMRE wire packet from inner DMR voice frame. None if ver unsupported (2/3). + + ``extended_layout`` matches legacy send_system: True when config VER > 4 (89-byte + trailer with source repeater); False when config VER == 4 (85-byte trailer). + """ + if embedded_ver in (2, 3): + return None + ver_byte = embedded_ver.to_bytes(1, "big") + ts = timestamp_ns.to_bytes(8, "big") + if extended_layout: + body = b"".join( + [ + DMRE, + dmr_packet[4:11], + server_id, + dmr_packet[15:], + ber, + rssi, + ver_byte, + ts, + source_server, + source_rptr, + hops, + ] + ) + else: + body = b"".join( + [ + DMRE, + dmr_packet[4:11], + server_id, + dmr_packet[15:], + ber, + rssi, + ver_byte, + ts, + source_server, + hops, + ] + ) + hash_len = len(body) + mac = dmre_blake2b_mac(body, passphrase, hash_len) + return body + mac diff --git a/src/adn_server/infrastructure/mesh/obp_v1.py b/src/adn_server/infrastructure/mesh/obp_v1.py new file mode 100644 index 0000000..ef01e60 --- /dev/null +++ b/src/adn_server/infrastructure/mesh/obp_v1.py @@ -0,0 +1,92 @@ +"""OpenBridge protocol v1: DMRD + HMAC-SHA1 control packets (no Twisted).""" + +from __future__ import annotations + +from dataclasses import dataclass +from hashlib import sha1 +from hmac import compare_digest +from hmac import new as hmac_new + +from ..hbp_constants import BCKA, BCSQ, BCST, BCVE, DMRD + +OBP_HMAC_LEN = 20 +DMRD_V1_PAYLOAD_LEN = 53 +DMRD_V1_WIRE_LEN = DMRD_V1_PAYLOAD_LEN + OBP_HMAC_LEN + + +def obp_hmac_sha1(passphrase: bytes, data: bytes) -> bytes: + return hmac_new(passphrase, data, sha1).digest() + + +@dataclass(frozen=True) +class VerifiedDmrdV1: + payload: bytes # 53 bytes including DMRD header + + +@dataclass(frozen=True) +class VerifiedBcsq: + tgid: bytes + stream_id: bytes + + +def build_dmrd_v1(dmr_packet: bytes, server_id: bytes, passphrase: bytes) -> bytes: + """Wire DMRD v1 (53 + HMAC) from a DMR/DMRD inner voice packet.""" + packet = b"".join([DMRD, dmr_packet[4:11], server_id, dmr_packet[15:]]) + return packet + obp_hmac_sha1(passphrase, packet) + + +def verify_dmrd_v1(packet: bytes, passphrase: bytes) -> VerifiedDmrdV1 | None: + if packet[:4] != DMRD or len(packet) < DMRD_V1_WIRE_LEN: + return None + payload = packet[:DMRD_V1_PAYLOAD_LEN] + mac = packet[DMRD_V1_PAYLOAD_LEN:DMRD_V1_WIRE_LEN] + if not compare_digest(mac, obp_hmac_sha1(passphrase, payload)): + return None + return VerifiedDmrdV1(payload=payload) + + +def build_bcka(passphrase: bytes) -> bytes: + return BCKA + obp_hmac_sha1(passphrase, BCKA) + + +def verify_bcka(packet: bytes, passphrase: bytes) -> bool: + if packet[:4] != BCKA or len(packet) < 4 + OBP_HMAC_LEN: + return False + return compare_digest(packet[4:24], obp_hmac_sha1(passphrase, packet[:4])) + + +def build_bcve(ver: int, passphrase: bytes) -> bytes: + packet = BCVE + ver.to_bytes(1, "big") + return packet + obp_hmac_sha1(passphrase, packet[4:5]) + + +def verify_bcve(packet: bytes, passphrase: bytes) -> tuple[bool, int | None]: + if packet[:4] != BCVE or len(packet) < 25: + return False, None + ver = int.from_bytes(packet[4:5], "big") + ok = compare_digest(packet[5:25], obp_hmac_sha1(passphrase, packet[4:5])) + return ok, ver if ok else None + + +def build_bcsq(tgid: bytes, stream_id: bytes, passphrase: bytes) -> bytes: + packet = BCSQ + tgid + stream_id + return packet + obp_hmac_sha1(passphrase, packet) + + +def verify_bcsq(packet: bytes, passphrase: bytes) -> VerifiedBcsq | None: + if packet[:4] != BCSQ or len(packet) < 31: + return None + mac = packet[11:31] + if not compare_digest(mac, obp_hmac_sha1(passphrase, packet[:11])): + return None + return VerifiedBcsq(tgid=packet[4:7], stream_id=packet[7:11]) + + +def build_bcst(passphrase: bytes) -> bytes: + return BCST + obp_hmac_sha1(passphrase, BCST) + + +def verify_bcst(packet: bytes, passphrase: bytes) -> bool: + if packet[:4] != BCST or len(packet) < 4 + OBP_HMAC_LEN: + return False + return compare_digest(packet[4:24], obp_hmac_sha1(passphrase, packet[:4])) diff --git a/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py b/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py index a0155ff..9adb9c3 100644 --- a/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py +++ b/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py @@ -31,9 +31,7 @@ import logging import time from binascii import a2b_hex as bhex from collections import deque -from hashlib import blake2b, sha1, sha256 -from hmac import compare_digest -from hmac import new as hmac_new +from hashlib import sha256 from random import randint from typing import Any, Callable @@ -50,6 +48,18 @@ from ...domain.talker_alias import ( store_ta_block, try_buffer_ta_from_voice_fragments, ) +from ..mesh.dmre_v5 import build_dmre, parse_dmre_trailer, verify_dmre_mac +from ..mesh.obp_v1 import ( + build_bcka, + build_bcve, + build_bcsq, + build_dmrd_v1, + verify_bcka, + verify_bcsq, + verify_bcst, + verify_bcve, + verify_dmrd_v1, +) from ..hbp_constants import ( BC, BCKA, @@ -400,30 +410,46 @@ class HBPProtocol(DatagramProtocol): _server_id = _sid if isinstance(_sid, bytes) and len(_sid) >= 4 else bytes_4(int(_sid) & 0xFFFFFFFF if isinstance(_sid, int) else 0) _passphrase = _get_passphrase_bytes(self._config) _target_addr = (self._config["TARGET_IP"], self._config["TARGET_PORT"]) - if "VER" in self._config and self._config["VER"] > 4: - _ver = VER.to_bytes(1, "big") - _packet = b"".join([DMRE, _packet[4:11], _server_id, _packet[15:], _ber, _rssi, _ver, time.time_ns().to_bytes(8, "big"), _source_server, _source_rptr, _hops]) - _h = blake2b(key=_passphrase, digest_size=16) - _h.update(_packet) - _hash = _h.digest() - _packet = b"".join([_packet, _hash]) - self.transport.write(_packet, _target_addr) - elif "VER" in self._config and self._config["VER"] == 4: - _ver = VER.to_bytes(1, "big") - _packet = b"".join([DMRE, _packet[4:11], _server_id, _packet[15:], _ber, _rssi, _ver, time.time_ns().to_bytes(8, "big"), _source_server, _hops]) - _h = blake2b(key=_passphrase, digest_size=16) - _h.update(_packet) - _hash = _h.digest() - _packet = b"".join([_packet, _hash]) - self.transport.write(_packet, _target_addr) - elif "VER" in self._config and self._config["VER"] == 3: + _ver_cfg = self._config.get("VER") + if "VER" in self._config and _ver_cfg is not None and _ver_cfg > 4: + _wire = build_dmre( + _packet, + server_id=_server_id, + ber=_ber, + rssi=_rssi, + embedded_ver=VER, + timestamp_ns=time.time_ns(), + source_server=_source_server, + source_rptr=_source_rptr, + hops=_hops, + passphrase=_passphrase, + extended_layout=True, + ) + if _wire is not None: + self.transport.write(_wire, _target_addr) + elif "VER" in self._config and _ver_cfg == 4: + _wire = build_dmre( + _packet, + server_id=_server_id, + ber=_ber, + rssi=_rssi, + embedded_ver=VER, + timestamp_ns=time.time_ns(), + source_server=_source_server, + source_rptr=_source_rptr, + hops=_hops, + passphrase=_passphrase, + extended_layout=False, + ) + if _wire is not None: + self.transport.write(_wire, _target_addr) + elif "VER" in self._config and _ver_cfg == 3: logger.error("(%s) protocol version 3 no longer supported", self._system) - elif "VER" in self._config and self._config["VER"] == 2: + elif "VER" in self._config and _ver_cfg == 2: logger.error("(%s) protocol version 2 no longer supported", self._system) else: - _packet = b"".join([DMRD, _packet[4:11], _server_id, _packet[15:]]) - _packet = b"".join([_packet, hmac_new(_passphrase, _packet, sha1).digest()]) - self.transport.write(_packet, _target_addr) + _wire = build_dmrd_v1(_packet, _server_id, _passphrase) + self.transport.write(_wire, _target_addr) else: if not self._config.get("TARGET_IP"): logger.debug("(%s) Not sent packet as TARGET_IP not currently known", self._system) @@ -1205,8 +1231,7 @@ class HBPProtocol(DatagramProtocol): """Legacy send_bcka: BCKA + HMAC-SHA1 to TARGET. Uses TARGET_SOCK (IP only; hostnames resolved at startup or on first peer packet).""" _addr = self._config.get("TARGET_SOCK") if _addr and _addr[0]: - _packet = BCKA + hmac_new(self._config["PASSPHRASE"], BCKA, sha1).digest() - self.transport.write(_packet, _addr) + self.transport.write(build_bcka(_get_passphrase_bytes(self._config)), _addr) else: logger.debug("(%s) *BridgeControl* not sending KeepAlive, TARGET not currently known", self._system) @@ -1214,9 +1239,7 @@ class HBPProtocol(DatagramProtocol): """Legacy send_bcve: BCVE + VER byte + HMAC-SHA1. Uses TARGET_SOCK (IP only).""" _addr = self._config.get("TARGET_SOCK") if self._config.get("ENHANCED_OBP") and _addr and _addr[0]: - _packet = BCVE + VER.to_bytes(1, "big") - _packet = _packet + hmac_new(self._config["PASSPHRASE"], _packet[4:5], sha1).digest() - self.transport.write(_packet, _addr) + self.transport.write(build_bcve(VER, _get_passphrase_bytes(self._config)), _addr) else: logger.debug("(%s) *BridgeControl* not sending BCVE, TARGET not currently known", self._system) @@ -1253,9 +1276,10 @@ class HBPProtocol(DatagramProtocol): _addr = (tip, tport) self._config["TARGET_SOCK"] = _addr if _addr and _addr[0]: - _packet = BCSQ + _tgid + _stream_id - _packet = _packet + hmac_new(self._config["PASSPHRASE"], _packet, sha1).digest() - self.transport.write(_packet, _addr) + self.transport.write( + build_bcsq(_tgid, _stream_id, _get_passphrase_bytes(self._config)), + _addr, + ) else: logger.warning( "(%s) *BridgeControl* BCSQ not sent: no TARGET_SOCK/TARGET_IP — peer cannot be quenched", @@ -1273,9 +1297,10 @@ class HBPProtocol(DatagramProtocol): self._laststrid.append(_stream_id) self._obp_send_bcve() return - _hash = _packet[53:73] - _ckhs = hmac_new(self._config["PASSPHRASE"], _data, sha1).digest() - if compare_digest(_hash, _ckhs) and (_sockaddr == self._config.get("TARGET_SOCK") or self._config.get("RELAX_CHECKS")): + _passphrase = _get_passphrase_bytes(self._config) + _verified = verify_dmrd_v1(_packet, _passphrase) + if _verified is not None and (_sockaddr == self._config.get("TARGET_SOCK") or self._config.get("RELAX_CHECKS")): + _data = _verified.payload self._obp_sync_target_sock_from_peer(_sockaddr) _peer_id = _data[11:15] if self._config.get("NETWORK_ID") != _peer_id: @@ -1378,38 +1403,25 @@ class HBPProtocol(DatagramProtocol): logger.warning("(%s) OpenBridge HMAC failed, packet discarded - OPCODE: %s SRC: %s", self._system, _packet[:4], _sockaddr) elif _packet[:4] == DMRE: # Legacy hblink.py OPENBRIDGE: DMRE (v5) incoming – 89-byte or 85-byte format, BLAKE2b - if len(_packet) < 69: + _trailer = parse_dmre_trailer(_packet) + if _trailer is None: return _data = _packet[:53] - # Legacy hblink OPENBRIDGE DMRE: BER/RSSI before version split (`hblink.py` ~432–433) - _ber = _packet[53:54] - _rssi = _packet[54:55] - _embedded_version = _packet[55] - if _embedded_version > 4: - if len(_packet) < 89: - return - _timestamp = _packet[56:64] - _source_server = _packet[64:68] - _source_rptr = _packet[68:72] - _hops = _packet[72] - _hash = _packet[73:89] - _hash_len = 73 - else: - if len(_packet) < 85: - return - _timestamp = _packet[56:64] - _source_server = _packet[64:68] - _source_rptr = b"\x00\x00\x00\x00" - _hops = _packet[68] - _hash = _packet[69:85] - _hash_len = 69 + _ber = _trailer.ber + _rssi = _trailer.rssi + _embedded_version = _trailer.embedded_version + _timestamp = _trailer.timestamp + _source_server = _trailer.source_server + _source_rptr = _trailer.source_rptr + _hops = _trailer.hops + _hash_len = _trailer.hash_len self._config["VER"] = _embedded_version _passphrase = _get_passphrase_bytes(self._config) - _h = blake2b(key=_passphrase, digest_size=16) - _h.update(_packet[:_hash_len]) - _ckhs = _h.digest() _stream_id = _data[16:20] - if not (compare_digest(_hash, _ckhs) and (_sockaddr == self._config.get("TARGET_SOCK") or self._config.get("RELAX_CHECKS"))): + if not ( + verify_dmre_mac(_packet, _passphrase, _hash_len) + and (_sockaddr == self._config.get("TARGET_SOCK") or self._config.get("RELAX_CHECKS")) + ): logger.warning("(%s) OpenBridge DMRE BLAKE2b failed, packet discarded - SRC: %s", self._system, _sockaddr) return self._obp_sync_target_sock_from_peer(_sockaddr) @@ -1565,10 +1577,9 @@ class HBPProtocol(DatagramProtocol): elif _packet[:4] == EOBP: logger.warning("(%s) *ProtoControl* KF7EEL EOBP protocol not supported", self._system) elif self._config.get("ENHANCED_OBP") and _packet[:2] == BC: + _passphrase = _get_passphrase_bytes(self._config) if _packet[:4] == BCKA and len(_packet) >= 24: - _hash = _packet[4:24] - _ckhs = hmac_new(self._config["PASSPHRASE"], _packet[:4], sha1).digest() - if compare_digest(_hash, _ckhs): + if verify_bcka(_packet, _passphrase): self._config["_bcka"] = time.time() if _sockaddr != self._config.get("TARGET_SOCK"): logger.info("(%s) *BridgeControl* Source IP and Port has changed for OBP from %s:%s to %s:%s, updating", self._system, self._config.get("TARGET_IP"), self._config.get("TARGET_PORT"), _sockaddr[0], _sockaddr[1]) @@ -1580,11 +1591,10 @@ class HBPProtocol(DatagramProtocol): logger.info("(%s) *BridgeControl* BCKA invalid KeepAlive, packet discarded", self._system) # Source quench — legacy hblink.py OPENBRIDGE ~629-639 (sets CONFIG['_bcsq'][tgid]=stream_id) if _packet[:4] == BCSQ and len(_packet) >= 31: - _hash_bcsq = _packet[11:] - _tgid_bcsq = _packet[4:7] - _stream_bcsq = _packet[7:11] - _ckhs_bcsq = hmac_new(self._config["PASSPHRASE"], _packet[:11], sha1).digest() - if compare_digest(_hash_bcsq, _ckhs_bcsq): + _bcsq = verify_bcsq(_packet, _passphrase) + if _bcsq is not None: + _tgid_bcsq = _bcsq.tgid + _stream_bcsq = _bcsq.stream_id if "_bcsq" not in self._config: self._config["_bcsq"] = {} self._config["_bcsq"][_tgid_bcsq] = _stream_bcsq @@ -1613,9 +1623,7 @@ class HBPProtocol(DatagramProtocol): ) # STUN — must match send_bcst: HMAC-SHA1 over opcode only (hblink.py ~282-285). RX used _packet[4:] in ~647 but that does not match TX. if _packet[:4] == BCST and len(_packet) >= 24: - _hash_bcst = _packet[4:24] - _ckhs_bcst = hmac_new(self._config["PASSPHRASE"], _packet[:4], sha1).digest() - if compare_digest(_hash_bcst, _ckhs_bcst): + if verify_bcst(_packet, _passphrase): logger.trace("(%s) *BridgeControl* BCST STUN request received", self._system) self._config["_STUN"] = True else: @@ -1625,10 +1633,8 @@ class HBPProtocol(DatagramProtocol): _sockaddr, ) if _packet[:4] == BCVE and len(_packet) >= 25: - _ver = int.from_bytes(_packet[4:5], "big") - _hash = _packet[5:25] - _ckhs = hmac_new(self._config["PASSPHRASE"], _packet[4:5], sha1).digest() - if compare_digest(_hash, _ckhs): + _bcve_ok, _ver = verify_bcve(_packet, _passphrase) + if _bcve_ok and _ver is not None: if _ver in (2, 3) or _ver > 5: logger.info("(%s) *ProtoControl* BCVE Version not supported, Ver: %s", self._system, _ver) elif _ver > self._config.get("VER", 5): diff --git a/tests/infrastructure/mesh/test_dmre_v5.py b/tests/infrastructure/mesh/test_dmre_v5.py new file mode 100644 index 0000000..317a6c3 --- /dev/null +++ b/tests/infrastructure/mesh/test_dmre_v5.py @@ -0,0 +1,76 @@ +"""Unit tests for DMRE v4/v5 BLAKE2b wire codec.""" + +from __future__ import annotations + +from adn_server.domain import bytes_4 +from adn_server.domain.hbp_protocol import VER +from adn_server.infrastructure.hbp_constants import DMRD, DMRE +from adn_server.infrastructure.mesh.dmre_v5 import ( + build_dmre, + parse_dmre_trailer, + verify_dmre_mac, +) + +_PASS = b"test-passphrase\x00\x00\x00\x00\x00\x00" + + +def _sample_dmr_voice() -> bytes: + return b"".join( + [ + DMRD, + bytes([1]), + bytes_4(1001)[1:4], + bytes_4(52090)[1:4], + bytes_4(1), + bytes([0x10]), + bytes_4(0xAABBCCDD), + b"\x00" * 33, + ] + ) + + +def test_dmre_v5_extended_layout() -> None: + inner = _sample_dmr_voice() + wire = build_dmre( + inner, + server_id=bytes_4(9990), + ber=b"\x00", + rssi=b"\x00", + embedded_ver=VER, + timestamp_ns=1_700_000_000_000_000_000, + source_server=bytes_4(9990), + source_rptr=bytes_4(100), + hops=b"\x01", + passphrase=_PASS, + extended_layout=True, + ) + assert wire is not None + assert len(wire) == 89 + assert wire[:4] == DMRE + trailer = parse_dmre_trailer(wire) + assert trailer is not None + assert trailer.hash_len == 73 + assert verify_dmre_mac(wire, _PASS, trailer.hash_len) + + +def test_dmre_v4_compact_layout() -> None: + inner = _sample_dmr_voice() + wire = build_dmre( + inner, + server_id=bytes_4(9990), + ber=b"\x00", + rssi=b"\x00", + embedded_ver=4, + timestamp_ns=1_700_000_000_000_000_000, + source_server=bytes_4(9990), + source_rptr=b"\x00\x00\x00\x00", + hops=b"\x01", + passphrase=_PASS, + extended_layout=False, + ) + assert wire is not None + assert len(wire) == 85 + trailer = parse_dmre_trailer(wire) + assert trailer is not None + assert trailer.hash_len == 69 + assert verify_dmre_mac(wire, _PASS, trailer.hash_len) diff --git a/tests/infrastructure/mesh/test_obp_v1.py b/tests/infrastructure/mesh/test_obp_v1.py new file mode 100644 index 0000000..5f00359 --- /dev/null +++ b/tests/infrastructure/mesh/test_obp_v1.py @@ -0,0 +1,76 @@ +"""Unit tests for OpenBridge v1 HMAC wire codec.""" + +from __future__ import annotations + +from adn_server.domain import bytes_4 +from adn_server.infrastructure.hbp_constants import BCKA, BCSQ, BCST, BCVE, DMRD +from adn_server.infrastructure.mesh.obp_v1 import ( + DMRD_V1_WIRE_LEN, + build_bcka, + build_bcst, + build_bcve, + build_bcsq, + build_dmrd_v1, + verify_bcka, + verify_bcsq, + verify_bcst, + verify_bcve, + verify_dmrd_v1, +) + +_PASS = b"test-passphrase\x00\x00\x00\x00\x00\x00" + + +def _sample_dmr_voice() -> bytes: + return b"".join( + [ + DMRD, + bytes([1]), + bytes_4(1001)[1:4], + bytes_4(52090)[1:4], + bytes_4(1), + bytes([0x10]), + bytes_4(0xAABBCCDD), + b"\x00" * 33, + ] + ) + + +def test_dmrd_v1_roundtrip() -> None: + inner = _sample_dmr_voice() + wire = build_dmrd_v1(inner, bytes_4(9990), _PASS) + assert len(wire) == DMRD_V1_WIRE_LEN + verified = verify_dmrd_v1(wire, _PASS) + assert verified is not None + assert verified.payload[:4] == DMRD + + +def test_bcka_roundtrip() -> None: + wire = build_bcka(_PASS) + assert wire[:4] == BCKA + assert verify_bcka(wire, _PASS) + assert not verify_bcka(wire[:20], _PASS) + + +def test_bcst_roundtrip() -> None: + wire = build_bcst(_PASS) + assert wire[:4] == BCST + assert verify_bcst(wire, _PASS) + + +def test_bcve_roundtrip() -> None: + wire = build_bcve(5, _PASS) + assert wire[:4] == BCVE + ok, ver = verify_bcve(wire, _PASS) + assert ok and ver == 5 + + +def test_bcsq_roundtrip() -> None: + tgid = bytes_4(52090)[1:4] + stream = bytes_4(0x11223344) + wire = build_bcsq(tgid, stream, _PASS) + assert wire[:4] == BCSQ + verified = verify_bcsq(wire, _PASS) + assert verified is not None + assert verified.tgid == tgid + assert verified.stream_id == stream