"""DMRhub v2 · M1 — полный анализатор DMR/Homebrew-пакета (оффлайн, без сети). Задача: по ЛЮБОМУ UDP-датаграмму отдать полностью разобранный объект. • Homebrew (HBP) управляющий пакет → тип + peer_id (RPTL/RPTK/RPTC/RPTPING/…). • DMRD (бинарный кадр DMR) → src, dst, peer_id, слот, call_type, frame_type, data_type (с именем), stream_id, seq, BER/RSSI + классификация (старт/конец голоса, CSBK, данные/SMS). Глубокий декод полезной нагрузки (текст SMS, embedded Link Control / Talker Alias, CSBK-опкод) — шаги M1b/M1c. Всё big-endian. Референс: ETSI TS 102 361-1, MMDVMHost/HBlink3, наш обкатанный master/homebrew.py. M1a — ТОЛЬКО заголовки/классификация, детерминированно и без внешних зависимостей → легко тестировать на дампах. """ from __future__ import annotations import hashlib import struct from dataclasses import dataclass, field from enum import Enum # ─────────────────────────── Homebrew (HBP) теги ─────────────────────────── # peer → master TAG_LOGIN = b"RPTL" # RPTL + peer_id(4) TAG_AUTH = b"RPTK" # RPTK + peer_id(4) + sha256(32) TAG_CONFIG = b"RPTC" # RPTC + peer_id(4) + config(~306) TAG_CLOSE = b"RPTCL" # RPTCL + peer_id(4) TAG_PING = b"RPTPING" # RPTPING + peer_id(4) TAG_OPTIONS = b"RPTO" # RPTO + peer_id(4) + options(...) TAG_DMRD = b"DMRD" # DMRD + 51-байт payload (всего 55) # master → peer TAG_ACK = b"RPTACK" # RPTACK + 4 байта (salt при логине / id при успехе) TAG_NAK = b"MSTNAK" # MSTNAK + peer_id(4) TAG_PONG = b"MSTPONG" # MSTPONG + peer_id(4) TAG_MSTCL = b"MSTCL" # MSTCL + peer_id(4) class Kind(str, Enum): LOGIN = "login"; AUTH = "auth"; CONFIG = "config"; CLOSE = "close" PING = "ping"; OPTIONS = "options"; DMRD = "dmrd" ACK = "ack"; NAK = "nak"; PONG = "pong"; MSTCL = "mstcl" UNKNOWN = "unknown" # ─────────────────────────── DMR: типы кадров ────────────────────────────── class FrameType(int, Enum): VOICE = 0x0 # голос-бёрст БЕЗ sync (B–E/F): несёт EMB + фрагмент embedded LC / Talker Alias VOICE_SYNC = 0x1 # голос-бёрст A с sync-паттерном (embedded signalling тут НЕТ) DATA_SYNC = 0x2 # данные/сигнализация (header, terminator, CSBK, data) # data_type (младший ниббл при FrameType.DATA_SYNC) — ETSI TS 102 361-1 class DataType(int, Enum): PI_HEADER = 0x0 # индикатор шифрования VOICE_LC_HEADER = 0x1 # старт голосового вызова (Link Control) TERMINATOR_LC = 0x2 # конец передачи (Link Control) CSBK = 0x3 # Control Signalling Block (Call Alert/Radio Check/Stun/…) MBC_HEADER = 0x4 # Multi-Block Control header MBC_CONT = 0x5 # MBC continuation DATA_HEADER = 0x6 # старт пакетных данных (SMS/GPS) RATE12_DATA = 0x7 # блок данных rate-1/2 RATE34_DATA = 0x8 # блок данных rate-3/4 IDLE = 0x9 RATE1_DATA = 0xA # блок данных rate-1 CALL_GROUP = "group" CALL_UNIT = "unit" # приват (рация↔рация) def auth_hash(salt: bytes, passphrase: bytes) -> bytes: """sha256(salt + passphrase) — значение для RPTK.""" return hashlib.sha256(salt + passphrase).digest() # ─────────────────────────── разобранные объекты ─────────────────────────── @dataclass class HBP: """Разобранный управляющий (не-DMRD) Homebrew-пакет.""" kind: Kind peer_id: int | None raw: bytes # для AUTH: присланный sha256; для CONFIG/OPTIONS: сырой хвост auth: bytes | None = None tail: bytes | None = None # для CONFIG: разобранные поля RPTC (частоты, координаты, софт — «что шлёт Pi-Star») config: dict | None = None # ── RPTC config: фикс-ширины ASCII (MMDVMHost/HBlink3). Смещения от начала пакета # ('RPTC'[0:4] + id[4:8]). Старый мастер это НЕ парсил — теперь ловим всё. ── _RPTC_FIELDS = [ ("callsign", 8, 16, "s"), ("rx_freq", 16, 25, "i"), ("tx_freq", 25, 34, "i"), ("tx_power", 34, 36, "i"), ("color_code", 36, 38, "i"), ("lat", 38, 46, "f"), ("lon", 46, 55, "f"), ("height", 55, 58, "i"), ("location", 58, 78, "s"), ("description", 78, 97, "s"), ("slots", 97, 98, "i"), ("url", 98, 222, "s"), ("software_id", 222, 262, "s"), ("package_id", 262, 302, "s"), ] def parse_rptc_config(raw: bytes) -> dict: """Разобрать конфиг хотспота из RPTC: частоты, координаты, локация, софт/версия.""" out: dict = {} for name, a, b, typ in _RPTC_FIELDS: chunk = raw[a:b] if len(chunk) < (b - a): continue s = chunk.decode("ascii", "ignore").strip().strip("\x00").strip() try: if typ == "i": out[name] = int(s) if s else None elif typ == "f": out[name] = float(s) if s and s not in ("-", ".") else None else: out[name] = s or None except ValueError: out[name] = s or None return out @dataclass class DMR: """Полностью разобранный DMRD-кадр (заголовок + классификация). Глубокий декод нагрузки заполняется на M1b/M1c (поля decoded/*).""" raw: bytes seq: int src: int # ID рации-отправителя (talker) dst: int # TG (group) или ID рации (unit) peer_id: int # ID хотспота/репитера-источника slot: int # 1 | 2 call_type: str # group | unit frame_type: FrameType data_type: int # DataType при DATA_SYNC, иначе voice-seq (0..5) stream_id: int # одна непрерывная передача ber: int | None = None # трейлер: BER (%) rssi: int | None = None # трейлер: RSSI (−dBm) decoded: dict = field(default_factory=dict) # M1b/M1c: sms/lc/ta/csbk # ── классификация (что это за кадр) ── @property def data_type_name(self) -> str: if self.frame_type == FrameType.DATA_SYNC: try: return DataType(self.data_type).name except ValueError: return f"DATA_0x{self.data_type:X}" return f"VOICE_SEQ_{self.data_type}" # A..F голос-кадры несут vseq 0..5 @property def is_voice_start(self) -> bool: return self.frame_type == FrameType.DATA_SYNC and self.data_type == DataType.VOICE_LC_HEADER @property def is_terminator(self) -> bool: return self.frame_type == FrameType.DATA_SYNC and self.data_type == DataType.TERMINATOR_LC @property def is_voice(self) -> bool: return self.frame_type in (FrameType.VOICE, FrameType.VOICE_SYNC) @property def is_csbk(self) -> bool: return self.frame_type == FrameType.DATA_SYNC and self.data_type == DataType.CSBK @property def is_data(self) -> bool: """Пакетные данные/SMS-поток (header + rate-блоки), НЕ голос и НЕ CSBK.""" return self.frame_type == FrameType.DATA_SYNC and self.data_type in ( DataType.DATA_HEADER, DataType.RATE12_DATA, DataType.RATE34_DATA, DataType.RATE1_DATA, DataType.MBC_HEADER, DataType.MBC_CONT) @property def dmr_payload(self) -> bytes: """33 байта DMR-слота (FEC-кодировано; декод — по data_type на M1b).""" return self.raw[20:53] def summary(self) -> str: who = f"src={self.src}→{'TG' if self.call_type==CALL_GROUP else 'unit'} {self.dst}" q = f" ber={self.ber}%" if self.ber is not None else "" return (f"DMRD peer={self.peer_id} TS{self.slot} {self.call_type} " f"{self.frame_type.name}/{self.data_type_name} {who} " f"stream={self.stream_id:08x} seq={self.seq}{q}") # ─────────────────────────────── анализатор ──────────────────────────────── def _peer4(data: bytes, off: int) -> int | None: return struct.unpack(">I", data[off:off + 4])[0] if len(data) >= off + 4 else None def parse_dmrd(data: bytes) -> DMR: """DMRD (≥53 байта). Раскладка: [0:4]'DMRD' [4]seq [5:8]src [8:11]dst [11:15]peer [15]bits [16:20]stream [20:53]DMR-данные [53]BER? [54]RSSI? Байт 15: slot=bit7, call(unit)=bit6, frame_type=bits5-4, data_type=bits3-0.""" seq = data[4] src = int.from_bytes(data[5:8], "big") dst = int.from_bytes(data[8:11], "big") peer_id = int.from_bytes(data[11:15], "big") bits = data[15] slot = 2 if (bits & 0x80) else 1 call_type = CALL_UNIT if (bits & 0x40) else CALL_GROUP frame_type = FrameType((bits & 0x30) >> 4) data_type = bits & 0x0F stream_id = int.from_bytes(data[16:20], "big") ber = data[53] if len(data) >= 54 else None rssi = data[54] if len(data) >= 55 else None return DMR(raw=data, seq=seq, src=src, dst=dst, peer_id=peer_id, slot=slot, call_type=call_type, frame_type=frame_type, data_type=data_type, stream_id=stream_id, ber=ber, rssi=rssi) def build_dmrd(dmr: "DMR", burst33: bytes) -> bytes: """Собрать 55-байт Homebrew DMRD-кадр из DMR-модели (для пересылки Homebrew-пиру).""" b15 = ((0x80 if dmr.slot == 2 else 0) | (0x40 if dmr.call_type == CALL_UNIT else 0) | ((int(dmr.frame_type) & 0x3) << 4) | (dmr.data_type & 0x0F)) return (TAG_DMRD + bytes([dmr.seq & 0xFF]) + dmr.src.to_bytes(3, "big") + dmr.dst.to_bytes(3, "big") + dmr.peer_id.to_bytes(4, "big") + bytes([b15]) + (dmr.stream_id & 0xFFFFFFFF).to_bytes(4, "big") + bytes(burst33[:33]).ljust(33, b"\x00") + bytes([dmr.ber or 0, dmr.rssi or 0])) def analyze(data: bytes) -> HBP | DMR: """Классифицировать любой входящий UDP-датаграмм.""" if len(data) < 4: return HBP(Kind.UNKNOWN, None, data) if data[:4] == TAG_DMRD: if len(data) < 53: return HBP(Kind.UNKNOWN, _peer4(data, 11), data) return parse_dmrd(data) # управляющие: тег + peer_id(4) (у RPTPING/RPTCL/RPTACK/… тег разной длины) if data[:4] == TAG_LOGIN: return HBP(Kind.LOGIN, _peer4(data, 4), data) if data[:4] == TAG_AUTH: return HBP(Kind.AUTH, _peer4(data, 4), data, auth=data[8:40] if len(data) >= 40 else None) # ВНИМАНИЕ: RPTCL (close, 5 байт) начинается с 'RPTC' — проверяем ДО RPTC (config). if data[:5] == TAG_CLOSE: return HBP(Kind.CLOSE, _peer4(data, 5), data) if data[:4] == TAG_CONFIG: return HBP(Kind.CONFIG, _peer4(data, 4), data, tail=data[8:], config=parse_rptc_config(data)) if data[:4] == TAG_OPTIONS: return HBP(Kind.OPTIONS, _peer4(data, 4), data, tail=data[8:]) if data[:7] == TAG_PING: return HBP(Kind.PING, _peer4(data, 7), data) if data[:6] == TAG_ACK: return HBP(Kind.ACK, _peer4(data, 6), data) if data[:6] == TAG_NAK: return HBP(Kind.NAK, _peer4(data, 6), data) if data[:7] == TAG_PONG: return HBP(Kind.PONG, _peer4(data, 7), data) if data[:5] == TAG_MSTCL: return HBP(Kind.MSTCL, _peer4(data, 5), data) return HBP(Kind.UNKNOWN, None, data)