commit 8e3b9144c337bcafe05a428bb0a7b087376c13f0 Author: viktor Date: Tue Jul 28 23:32:04 2026 +0900 DMR-сервер: мастер Homebrew на stdlib для Pi-Star Ядро протокола форкнуто из dmrhubv2 (~1300 строк из 31700): analyzer, server, parrot, dmr_voice, bptc. Отрезаны портал, БД, мосты, алерты, динамические подписки. Своё: конфиг, состояние, маршрутизатор, панель, установщик. Ноль зависимостей — на Debian 13 действует PEP 668, pip в систему запрещён. Две правки под Pi-Star, которых в исходном коде не было: • peer_id переписывается на получателя (байты 11-14) через хук encode_for — так делают HBlink3/HBlink4 и BrandMeister, а DMRGateway на Pi-Star стоит между MMDVMHost и сетью; • приватный вызов при неизвестном абоненте уходит всем, а не в никуда (взято из HBlink4: user cache with broadcast fallback). docs/PROTOCOL.md фиксирует, что тест-вектор auth-хеша в спецификации 2015 года ОШИБОЧЕН (посчитан от ASCII-строки соли вместо четырёх байт). Правка кода по нему сломала бы вход всем хотспотам. tests/test_e2e.py: 15 проверок настоящими UDP-кадрами MMDVMHost. Все проходят. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01XvX94Vwy7vXWEuiFm1g9Av diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b7b743f --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.pyc +config.json diff --git a/README.md b/README.md new file mode 100644 index 0000000..3fad6b5 --- /dev/null +++ b/README.md @@ -0,0 +1,72 @@ +# DMR-сервер + +Простой Homebrew/MMDVM мастер для внутренней сети: несколько хотспотов **Pi-Star**, +веб-панель с группами и живым журналом, попугай (эхотест) на TG 9990. + +**Ноль зависимостей** — только `python3` (3.10+). На Debian 13 действует PEP 668, +и `pip install` в систему запрещён; venv ради пары библиотек — лишний слой, который +однажды сломается. Поэтому всё на stdlib. + +## Установка + +```bash +sudo bash install.sh +``` + +Ставит код в `/opt/dmr-server`, конфиг в `/etc/dmr-server/config.json`, заводит юнит +systemd под отдельным пользователем и печатает адрес панели с паролем. Повторный запуск +безопасен — конфиг не перезаписывается. + +## Настройка Pi-Star + +`Configuration → DMR Master`: адрес сервера, порт **62031**, пароль — значение +`passphrase` из конфига (**обязательно смени умолчание `passw0rd`**). + +## Что умеет + +- Рукопожатие Homebrew: `RPTL → RPTACK+salt → RPTK → RPTC → RPTPING/MSTPONG`. +- Групповые вызовы: кадр от одного хотспота уходит всем остальным, если TG есть в списке групп. +- Приватные вызовы: туда, где абонента слышали последним; если не слышали ни разу — + всем хотспотам (как в HBlink4). +- Попугай на 9990, групповой и приватный: записывает передачу и проигрывает обратно. +- Панель: список хотспотов, добавление/удаление групп, последние вызовы, живой журнал, + принудительное отключение хотспота. + +## Проверка + +```bash +python3 tests/test_e2e.py +``` + +Поднимает настоящий мастер и говорит с ним по UDP кадрами, которые шлёт MMDVMHost: +рукопожатие, отказ по неверному паролю, keepalive, доставка группового вызова между +двумя хотспотами, отсутствие эха отправителю, фильтр по разрешённым группам, приват. +Проверяет и то, что легко испортить: длину кадра 55 байт, переписывание `peer_id`, +дословную копию `stream_id` и неизменность бёрста. + +## Устройство + +| Файл | Что делает | +|---|---| +| `dmrsrv/analyzer.py` | разбор и сборка кадров Homebrew/DMR | +| `dmrsrv/server.py` | ядро мастера: рукопожатие, реестр хотспотов, маршрут в эфир | +| `dmrsrv/router.py` | кому пересылать кадр + подстановка `peer_id` получателя | +| `dmrsrv/parrot.py` | эхотест 9990 | +| `dmrsrv/dmr_voice.py`, `bptc.py` | пересборка LC — нужна только приватному попугаю | +| `dmrsrv/panel.py` | веб-панель на `http.server` | +| `dmrsrv/config.py` | конфиг с атомарным сохранением | +| `dmrsrv/state.py` | состояние и журнал для панели | + +Ядро протокола форкнуто из `viktor/dmrhubv2` (~1300 строк из 31700) — там оно уже +обкатано на живом эфире. Отрезаны портал, БД, мосты, алерты, динамические подписки. + +**Перед правкой протокола прочти [`docs/PROTOCOL.md`](docs/PROTOCOL.md).** Спецификация +2015 года врёт в нескольких местах, включая собственный тест-вектор auth-хеша; +там записано, где именно и почему прав код, а не бумажка. + +## Ограничения + +- Один сервер = одна сеть, без транков между серверами (OpenBridge не реализован). +- Группы статические, из панели. Динамических подписок через `RPTO` нет. +- Панель по HTTP с Basic-авторизацией — рассчитана на доверенную внутреннюю сеть. + Наружу выставлять только за HTTPS-прокси. diff --git a/dmrsrv/__init__.py b/dmrsrv/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/dmrsrv/__main__.py b/dmrsrv/__main__.py new file mode 100644 index 0000000..e9100b1 --- /dev/null +++ b/dmrsrv/__main__.py @@ -0,0 +1,138 @@ +"""Точка входа: python3 -m dmrsrv + +Связывает мастер, маршрутизатор, попугая, состояние и панель. Всё на stdlib — +на Debian 13 действует PEP 668, ставить что-либо через pip в систему нельзя, +а venv ради пары строк кода — лишний слой, который однажды сломается. +""" +from __future__ import annotations + +import asyncio +import contextlib +import signal +import sys + +from . import config as cfgmod +from . import log as logmod +from . import panel as panelmod +from .analyzer import TAG_MSTCL +from .parrot import Parrot +from .router import Router, retarget +from .server import HomebrewMaster, PeerState, run_server +from .state import Peer as StatePeer +from .state import State + + +def _wire_state(master: HomebrewMaster, state: State) -> None: + """Зеркалим реестр пиров мастера в State для панели. + + Мастер — форк из DMRhub, его не трогаем: снимаем состояние снаружи, + чтобы обновление форка не конфликтовало. + """ + + def sync() -> None: + for pid, p in list(master.peers.items()): + sp = state.peer_get(pid) + cfgp = p.config or {} + if sp is None: + sp = StatePeer(peer_id=pid, addr=p.addr) + state.peer_add(sp) + sp.addr = p.addr + sp.stage = {PeerState.NEW: "вход", PeerState.CHALLENGE: "пароль", + PeerState.AUTHED: "конфигурация", + PeerState.CONNECTED: "в сети"}.get(p.state, p.state) + sp.callsign = cfgp.get("callsign", "") + sp.location = cfgp.get("location", "") + sp.description = cfgp.get("description", "") + sp.software = cfgp.get("package_id", "") or cfgp.get("software_id", "") + sp.colorcode = cfgp.get("colorcode", "") + sp.rx_freq = cfgp.get("rx_freq", "") + sp.tx_freq = cfgp.get("tx_freq", "") + for pid in list(state.peers): + if pid not in master.peers: + gone = state.peer_del(pid) + if gone: + state.event("peer", f"хотспот {pid} ({gone.callsign}) отключился") + + master._state_sync = sync # зовём из фоновой задачи ниже + + +async def _sync_loop(master: HomebrewMaster, interval: float = 1.0) -> None: + while True: + await asyncio.sleep(interval) + with contextlib.suppress(Exception): + master._state_sync() + + +def main() -> int: + cfg = cfgmod.load() + state = State(log_keep=int(cfg.get("log_keep", 500))) + logmod.setup() + logmod.set_sink(state) + + router = Router(cfg, state) + passphrase = cfg["passphrase"].encode("utf-8") + allow = {int(x) for x in cfg.get("peers_allowlist", [])} + + def password_for(peer_id: int) -> bytes | None: + """Пароль хотспота. Пустой список разрешённых = пускать всех, кто знает пароль.""" + if allow and peer_id not in allow: + state.event("warn", f"хотспот {peer_id} не в списке разрешённых — отказ") + return None + return passphrase + + master = HomebrewMaster( + password_for=password_for, + router=router, + # Подставляем номер ПОЛУЧАТЕЛЯ в байты 11-14. Так делают HBlink3/HBlink4 + # и BrandMeister; DMRGateway на Pi-Star стоит между MMDVMHost и сетью. + encode_for=lambda tgt, dmr: retarget(dmr.raw, tgt.peer_id), + ) + _wire_state(master, state) + + parrot = None + if cfg.get("parrot", {}).get("enabled", True): + parrot = Parrot(master) + master.on_dmrd = lambda peer, dmr: parrot.feed(dmr) + state.event("info", f"попугай включён на TG{cfg['parrot'].get('tg', 9990)}") + + def drop_peer(pid: int) -> bool: + p = master.peers.pop(pid, None) + if p is None: + return False + with contextlib.suppress(Exception): + if master._send: + master._send(TAG_MSTCL + pid.to_bytes(4, "big"), p.addr) + return True + + panelmod.start(cfg, state, drop_peer) + + state.event("info", f"мастер слушает UDP {cfg['bind']}:{cfg['port']}, " + f"групп в списке: {len(cfg.get('groups', []))}") + if not cfg.get("groups"): + state.event("warn", "групп нет — групповые вызовы никуда не пойдут, добавь в панели") + + async def runner() -> None: + loop = asyncio.get_running_loop() + stop = loop.create_future() + for sig in (signal.SIGINT, signal.SIGTERM): + with contextlib.suppress(NotImplementedError): + loop.add_signal_handler(sig, lambda: stop.done() or stop.set_result(None)) + tasks = [ + asyncio.create_task(run_server(master, cfg["bind"], int(cfg["port"]))), + asyncio.create_task(_sync_loop(master)), + ] + await stop + state.event("info", "останавливаюсь") + for t in tasks: + t.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + try: + asyncio.run(runner()) + except KeyboardInterrupt: + pass + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/dmrsrv/analyzer.py b/dmrsrv/analyzer.py new file mode 100644 index 0000000..cc7e5e7 --- /dev/null +++ b/dmrsrv/analyzer.py @@ -0,0 +1,252 @@ +"""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) diff --git a/dmrsrv/bptc.py b/dmrsrv/bptc.py new file mode 100755 index 0000000..408d882 --- /dev/null +++ b/dmrsrv/bptc.py @@ -0,0 +1,140 @@ +""" +RadioStar — DMR BPTC(196,96) codec (pure Python). + +Ported from dmr_utils3 (N0MJS, GPL) + ETSI TS 102 361-1. Used to decode/encode +the 96-bit payloads carried by DATA_HEADER and CSBK bursts (data headers, CSBK +preambles, etc.). Lets the server READ and BUILD those control frames. + +A burst's 196 info bits live at burst-bit positions [0:98] + [166:264]; the +middle 68 bits (slot type + sync) are spliced from a real burst of the same type. +""" +from __future__ import annotations + +# interleaver index (on-air[INDEX_181[i]] = logical[i]) +INDEX_181 = ( + 0, 181, 166, 151, 136, 121, 106, 91, 76, 61, 46, 31, 16, 1, 182, 167, 152, 137, + 122, 107, 92, 77, 62, 47, 32, 17, 2, 183, 168, 153, 138, 123, 108, 93, 78, 63, + 48, 33, 18, 3, 184, 169, 154, 139, 124, 109, 94, 79, 64, 49, 34, 19, 4, 185, 170, + 155, 140, 125, 110, 95, 80, 65, 50, 35, 20, 5, 186, 171, 156, 141, 126, 111, 96, + 81, 66, 51, 36, 21, 6, 187, 172, 157, 142, 127, 112, 97, 82, 67, 52, 37, 22, 7, + 188, 173, 158, 143, 128, 113, 98, 83, 68, 53, 38, 23, 8, 189, 174, 159, 144, 129, + 114, 99, 84, 69, 54, 39, 24, 9, 190, 175, 160, 145, 130, 115, 100, 85, 70, 55, 40, + 25, 10, 191, 176, 161, 146, 131, 116, 101, 86, 71, 56, 41, 26, 11, 192, 177, 162, + 147, 132, 117, 102, 87, 72, 57, 42, 27, 12, 193, 178, 163, 148, 133, 118, 103, 88, + 73, 58, 43, 28, 13, 194, 179, 164, 149, 134, 119, 104, 89, 74, 59, 44, 29, 14, + 195, 180, 165, 150, 135, 120, 105, 90, 75, 60, 45, 30, 15) + +# the 96 data-bit positions in the deinterleaved matrix (72 LC/header + 24 RS/CRC) +DATA_POS = [ + 136, 121, 106, 91, 76, 61, 46, 31, + 152, 137, 122, 107, 92, 77, 62, 47, 32, 17, 2, + 123, 108, 93, 78, 63, 48, 33, 18, 3, 184, 169, + 94, 79, 64, 49, 34, 19, 4, 185, 170, 155, 140, + 65, 50, 35, 20, 5, 186, 171, 156, 141, 126, 111, + 36, 21, 6, 187, 172, 157, 142, 127, 112, 97, 82, + 7, 188, 173, 158, 143, 128, 113, 98, 83, + 68, 53, 174, 159, 144, 129, 114, 99, 84, 69, 54, 39, + 24, 145, 130, 115, 100, 85, 70, 55, 40, 25, 10, 191, +] + + +def _enc_15113(d): + return [ + d[0] ^ d[1] ^ d[2] ^ d[3] ^ d[5] ^ d[7] ^ d[8], + d[1] ^ d[2] ^ d[3] ^ d[4] ^ d[6] ^ d[8] ^ d[9], + d[2] ^ d[3] ^ d[4] ^ d[5] ^ d[7] ^ d[9] ^ d[10], + d[0] ^ d[1] ^ d[2] ^ d[4] ^ d[6] ^ d[7] ^ d[10], + ] + + +def _enc_1393(d): + return [ + d[0] ^ d[1] ^ d[3] ^ d[5] ^ d[6], + d[0] ^ d[1] ^ d[2] ^ d[4] ^ d[6] ^ d[7], + d[0] ^ d[1] ^ d[2] ^ d[3] ^ d[5] ^ d[7] ^ d[8], + d[0] ^ d[2] ^ d[4] ^ d[5] ^ d[8], + ] + + +def _bytes_to_bits(data): + out = [] + for byte in data: + for i in range(7, -1, -1): + out.append((byte >> i) & 1) + return out + + +def _bits_to_bytes(bits): + out = bytearray() + for i in range(0, len(bits), 8): + b = 0 + for j in range(8): + b = (b << 1) | bits[i + j] + out.append(b) + return bytes(out) + + +def _burst_info_bits(burst33): + """Extract the 196 on-air info bits from a 33-byte burst ([0:98]+[166:264]).""" + allbits = _bytes_to_bits(burst33) + return allbits[0:98] + allbits[166:264] + + +def decode(burst33: bytes) -> bytes: + """Decode a BPTC(196,96) burst -> 12 bytes (96 bits). No error correction. + DATA_POS index the on-air (interleaved) info bits directly (per dmr_utils).""" + onair = _burst_info_bits(burst33) # 196 on-air bits + bits = [onair[p] for p in DATA_POS] # extract 96 data bits + return _bits_to_bytes(bits) + + +def encode(data12: bytes) -> list: + """Encode 12 bytes -> 196 on-air info bits (list).""" + bits = _bytes_to_bits(data12) # 96 bits + for _ in range(4): + bits.insert(0, 0) # R0-R3 + for index in range(9): + spos = index * 15 + 1 + epos = spos + 11 + par = _enc_15113(bits[spos:epos]) + for pbit in range(4): + bits.insert(epos + pbit, par[pbit]) + for _ in range(60): + bits.append(0) + for col in range(15): + spos = col + 1 + column = [] + for _ in range(9): + column.append(bits[spos]) + spos += 15 + colp = _enc_1393(column) + cpar = 136 + col + for pbit in range(4): + bits[cpar] = colp[pbit] + cpar += 15 + onair = [0] * 196 + for i in range(196): + onair[INDEX_181[i]] = bits[i] # interleave + return onair + + +def splice_into_burst(template33: bytes, onair196: list) -> bytes: + """Build a burst: info bits from onair196, sync+slot-type from a template burst + of the same data type (positions [98:166]).""" + allbits = _bytes_to_bits(template33) + allbits[0:98] = onair196[0:98] + allbits[166:264] = onair196[98:196] + return _bits_to_bytes(allbits) + + +if __name__ == "__main__": + # decode the captured RT-4D DATA_HEADER (TEST123, 1400001 -> 1234567) + hdr = bytes.fromhex("157a07fb5a6681a8b2828e4b058d5d7f77fd7573388475f5803549e54a87813418") + dec = decode(hdr) + print("DATA_HEADER 12 bytes:", dec.hex()) + # round-trip: encode(decode) info bits must match the burst's info bits + rt = splice_into_burst(hdr, encode(dec)) + print("round-trip burst == original:", rt == hdr) + # decode a captured CSBK preamble + csbk = bytes.fromhex("47e7160d92e007ac4e398c7244cd5d7f77fd757ac996047099099220839a4a9b4e") + print("CSBK 12 bytes:", decode(csbk).hex()) \ No newline at end of file diff --git a/dmrsrv/config.py b/dmrsrv/config.py new file mode 100644 index 0000000..ea53d36 --- /dev/null +++ b/dmrsrv/config.py @@ -0,0 +1,90 @@ +"""Конфигурация сервера: JSON-файл, правится и руками, и из панели. + +Намеренно без внешних зависимостей — на Debian 13 (PEP 668) любой pip install +в систему запрещён, а заводить venv ради пары библиотек не хочется. +""" +from __future__ import annotations + +import contextlib +import json +import os +import secrets +import tempfile +from pathlib import Path +from typing import Any + +CONFIG_PATH = Path(os.environ.get("DMRSRV_CONFIG", "/etc/dmr-server/config.json")) + +PARROT_TG = 9990 # общепринятый номер «попугая» в DMR-сетях + +DEFAULTS: dict[str, Any] = { + "bind": "0.0.0.0", + "port": 62031, # стандартный порт Homebrew/MMDVM + "passphrase": "passw0rd", # пароль хотспотов (RPTK), меняется в панели + "panel": { + "bind": "0.0.0.0", + "port": 8080, + "user": "admin", + "password": "", # пустой -> сгенерируется при первом запуске + }, + "groups": [ + {"tg": 9, "name": "Локальная", "slot": 2}, + {"tg": 1, "name": "Общая", "slot": 1}, + ], + "parrot": {"enabled": True, "tg": PARROT_TG, "delay": 2.0}, + # Хотспоты, которым разрешено подключаться. Пустой список = пускать любой, + # кто знает пароль (удобно на старте, в панели можно перевести в белый список). + "peers_allowlist": [], + "log_keep": 500, # сколько последних событий держать для панели +} + + +def _merge(base: dict, over: dict) -> dict: + """Рекурсивно накладывает сохранённое поверх умолчаний. + + Нужно, чтобы добавление нового ключа в DEFAULTS не ломало существующий + конфиг и не требовало миграций. + """ + out = dict(base) + for k, v in over.items(): + if isinstance(v, dict) and isinstance(out.get(k), dict): + out[k] = _merge(out[k], v) + else: + out[k] = v + return out + + +def load() -> dict[str, Any]: + cfg = dict(DEFAULTS) + if CONFIG_PATH.exists(): + try: + cfg = _merge(DEFAULTS, json.loads(CONFIG_PATH.read_text("utf-8"))) + except (json.JSONDecodeError, OSError) as e: + # Битый конфиг не должен ронять сервер: работаем на умолчаниях, + # но громко об этом сообщаем. + print(f"[config] не читается {CONFIG_PATH}: {e}; беру умолчания") + if not cfg["panel"]["password"]: + cfg["panel"]["password"] = secrets.token_urlsafe(12) + save(cfg) + print(f"[config] пароль панели сгенерирован: {cfg['panel']['password']}") + return cfg + + +def save(cfg: dict[str, Any]) -> None: + """Атомарная запись: временный файл рядом + rename. + + Без этого обрыв питания посреди сохранения из панели оставит битый JSON. + """ + CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=CONFIG_PATH.parent, prefix=".config.", suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(cfg, f, ensure_ascii=False, indent=2) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp, CONFIG_PATH) + os.chmod(CONFIG_PATH, 0o600) # внутри пароли + except BaseException: + with contextlib.suppress(OSError): + os.unlink(tmp) + raise diff --git a/dmrsrv/dmr_voice.py b/dmrsrv/dmr_voice.py new file mode 100644 index 0000000..b6f2949 --- /dev/null +++ b/dmrsrv/dmr_voice.py @@ -0,0 +1,287 @@ +"""DMRhub v2 · сборка DMR-голоса ИЗ ВОКОДЕРА в эфир (без parrot-шаблонов). + +Нужно для инжекта чужого голоса (P25→DMR, TTS, приложение) DMR-пирам: собираем +голосовой суперкадр целиком в коде, с корректным FEC, без утечки чужой идентичности: + • Voice LC Header / Terminator = BPTC(196,96) над [LC9 + RS(12,9) 3Б] (+маска типа); + • Voice burst A = AMBE + BS voice sync; + • Voice bursts B,C,D,E = AMBE + EMB + фрагменты Embedded LC того же вызова; + • Voice burst F = AMBE + EMB (null embedded). + +Порт проверенного энкодера RadioStar (бит-в-бит против реального захвата): RS(12,9), +Embedded LC vBPTC(128,72) с Hamming(16,11)+column parity, EMB QR(16,7,6). BPTC(196,96) +берём из v2 `bptc.encode`. Slot type/sync — BS-каркас; на RF MMDVMHost перегенерит под +свой Color Code. Чистый stdlib + bptc. Самотест: `python dmr_voice.py`. +""" +from __future__ import annotations + +from . import bptc + +# ── RS(12,9) над GF(256), prim.poly 0x11D, корни a^1..a^3 (сверено с захватом) ── +_EXP = [0] * 512 +_LOG = [0] * 256 +_a = 1 +for _i in range(255): + _EXP[_i] = _a + _LOG[_a] = _i + _a <<= 1 + if _a & 0x100: + _a ^= 0x11D +for _i in range(255, 512): + _EXP[_i] = _EXP[_i - 255] + + +def _gmul(x: int, y: int) -> int: + return 0 if (x == 0 or y == 0) else _EXP[_LOG[x] + _LOG[y]] + + +def _rs_gen(): + g = [1] + for i in range(3): + root = _EXP[(1 + i) % 255] + ng = [0] * (len(g) + 1) + for j in range(len(g)): + ng[j] ^= g[j] + ng[j + 1] ^= _gmul(g[j], root) + g = ng + return g + + +_RSG = _rs_gen() + + +def rs129(lc9: bytes) -> bytes: + """9 байт LC → 3 байта RS(12,9) чётности (raw, без маски типа).""" + out = list(lc9) + [0, 0, 0] + for i in range(9): + c = out[i] + if c: + for j in range(1, 4): + out[i + j] ^= _gmul(_RSG[j], c) + return bytes(out[9:]) + + +# маски «полного LC» по типу данных (MMDVMHost CDMRFullLC) +_MASK_HEADER = (0x96, 0x96, 0x96) +_MASK_TERM = (0x99, 0x99, 0x99) + +# ── Sync (BS-источник, ETSI TS 102 361-1). На RF MMDVMHost перегенерит ── +SYNC_BS_VOICE = 0x755FD7DF75F7 +SYNC_BS_DATA = 0xDFF57D75DF5D + +# slot type (CC1-каркас; MMDVMHost подставит свой CC на RF) +_SLOTTYPE_CC1 = {1: 0x11B8C, 2: 0x12A59} + + +def _slottype(cc: int, dt: int) -> int: + return _SLOTTYPE_CC1[dt] + + +# ── EMB QR(16,7,6) — таблица g4klx MMDVMHost ── +_ENC_1676 = ( + 0x0000, 0x0273, 0x04E5, 0x0696, 0x09C9, 0x0BBA, 0x0D2C, 0x0F5F, 0x11E2, 0x1391, 0x1507, 0x1774, + 0x182B, 0x1A58, 0x1CCE, 0x1EBD, 0x21B7, 0x23C4, 0x2552, 0x2721, 0x287E, 0x2A0D, 0x2C9B, 0x2EE8, + 0x3055, 0x3226, 0x34B0, 0x36C3, 0x399C, 0x3BEF, 0x3D79, 0x3F0A, 0x411E, 0x436D, 0x45FB, 0x4788, + 0x48D7, 0x4AA4, 0x4C32, 0x4E41, 0x50FC, 0x528F, 0x5419, 0x566A, 0x5935, 0x5B46, 0x5DD0, 0x5FA3, + 0x60A9, 0x62DA, 0x644C, 0x663F, 0x6960, 0x6B13, 0x6D85, 0x6FF6, 0x714B, 0x7338, 0x75AE, 0x77DD, + 0x7882, 0x7AF1, 0x7C67, 0x7E14, 0x804F, 0x823C, 0x84AA, 0x86D9, 0x8986, 0x8BF5, 0x8D63, 0x8F10, + 0x91AD, 0x93DE, 0x9548, 0x973B, 0x9864, 0x9A17, 0x9C81, 0x9EF2, 0xA1F8, 0xA38B, 0xA51D, 0xA76E, + 0xA831, 0xAA42, 0xACD4, 0xAEA7, 0xB01A, 0xB269, 0xB4FF, 0xB68C, 0xB9D3, 0xBBA0, 0xBD36, 0xBF45, + 0xC151, 0xC322, 0xC5B4, 0xC7C7, 0xC898, 0xCAEB, 0xCC7D, 0xCE0E, 0xD0B3, 0xD2C0, 0xD456, 0xD625, + 0xD97A, 0xDB09, 0xDD9F, 0xDFEC, 0xE0E6, 0xE295, 0xE403, 0xE670, 0xE92F, 0xEB5C, 0xEDCA, 0xEFB9, + 0xF104, 0xF377, 0xF5E1, 0xF792, 0xF8CD, 0xFABE, 0xFC28, 0xFE5B, +) + +# LCSS (EMB): связь фрагментов Embedded LC (ETSI) +LCSS_SINGLE = 0b00 +LCSS_FIRST = 0b01 +LCSS_CONT = 0b11 +LCSS_LAST = 0b10 + + +def emb16(color_code: int, lcss: int, pi: int = 0) -> list[int]: + """16-бит EMB: CC(4)+PI(1)+LCSS(2) + QR(16,7,6) parity → 16 бит MSB-first.""" + value = ((color_code & 0x0F) << 3) | ((pi & 1) << 2) | (lcss & 0x03) + code = _ENC_1676[value & 0x7F] + return [(code >> i) & 1 for i in range(15, -1, -1)] + + +# ── bit-хелперы ── +def _bytes_to_bits(d): + o = [] + for by in d: + for i in range(7, -1, -1): + o.append((by >> i) & 1) + return o + + +def _bits_to_bytes(bits): + out = bytearray() + for i in range(0, len(bits), 8): + b = 0 + for j in range(8): + b = (b << 1) | bits[i + j] + out.append(b) + return bytes(out) + + +def _int_bits(v, n): + return [(v >> i) & 1 for i in range(n - 1, -1, -1)] + + +# ── Embedded LC vBPTC(128,72): Hamming(16,11)+column parity → 4 фрагмента B..E ── +def _csum5(lc9: bytes): + accum = sum(lc9[:9]) % 31 + return [(accum >> i) & 1 for i in range(4, -1, -1)] + + +def _enc_16114(d): + """Hamming(16,11,4) — 5 чётностей по 11 битам (dmr_utils3/hamming.py).""" + return [ + d[0] ^ d[1] ^ d[2] ^ d[3] ^ d[5] ^ d[7] ^ d[8], + d[1] ^ d[2] ^ d[3] ^ d[4] ^ d[6] ^ d[8] ^ d[9], + d[2] ^ d[3] ^ d[4] ^ d[5] ^ d[7] ^ d[9] ^ d[10], + d[0] ^ d[1] ^ d[2] ^ d[4] ^ d[6] ^ d[7] ^ d[10], + d[0] ^ d[2] ^ d[5] ^ d[6] ^ d[8] ^ d[9] ^ d[10], + ] + + +def encode_emblc(lc9: bytes) -> dict: + """9-байтный Embedded LC → 4 фрагмента по 32 бита (бёрсты B,C,D,E).""" + cs = _csum5(lc9) + bits = _bytes_to_bits(lc9) # 72 бита + for pos, bit in zip((32, 43, 54, 65, 76), cs): + bits.insert(pos, bit) + for index in range(0, 112, 16): # Hamming(16,11) по 7 строкам + par = _enc_16114(bits[index:index + 11]) + for hindex, hbit in zip(range(index + 11, index + 16), par): + bits.insert(hindex, hbit) + for index in range(16): # строка 7 = XOR строк 0..6 + bits.insert(index + 112, + bits[index] ^ bits[index + 16] ^ bits[index + 32] ^ bits[index + 48] + ^ bits[index + 64] ^ bits[index + 80] ^ bits[index + 96]) + + def frag(cols): + return [bits[c + r * 16] for c in cols for r in range(8)] + + return {1: frag([0, 1, 2, 3]), 2: frag([4, 5, 6, 7]), + 3: frag([8, 9, 10, 11]), 4: frag([12, 13, 14, 15])} + + +# ── Link Control + бёрсты ── +def lc9(dst: int, src: int, unit: bool = False) -> bytes: + """9-байтный LC голосового вызова: FLCO + FID + svcOpts + dst(3) + src(3).""" + flco = 0x03 if unit else 0x00 + return bytes([flco, 0x00, 0x00]) + dst.to_bytes(3, "big") + src.to_bytes(3, "big") + + +def _data_burst(info196, sync48: int, slottype20: int) -> bytes: + """264-битный data-бёрст: info[0:98]+[166:264], slot type[98:108]+[156:166], sync[108:156].""" + bits = [0] * 264 + bits[0:98] = info196[0:98] + bits[166:264] = info196[98:196] + st = _int_bits(slottype20, 20) + bits[98:108] = st[0:10] + bits[156:166] = st[10:20] + bits[108:156] = _int_bits(sync48, 48) + return _bits_to_bytes(bits) + + +def _full_lc_burst(dst, src, unit, cc, dt, mask) -> bytes: + l = lc9(dst, src, unit) + par = bytes(p ^ m for p, m in zip(rs129(l), mask)) + info = bptc.encode(l + par) # 196 on-air info bits + return _data_burst(info, SYNC_BS_DATA, _slottype(cc, dt)) + + +def voice_lc_header(dst, src, unit=False, cc=1) -> bytes: + """Бёрст Voice LC Header (data type 1).""" + return _full_lc_burst(dst, src, unit, cc, 1, _MASK_HEADER) + + +def terminator(dst, src, unit=False, cc=1) -> bytes: + """Бёрст Terminator with LC (data type 2).""" + return _full_lc_burst(dst, src, unit, cc, 2, _MASK_TERM) + + +def _put_ambe(ambe27: bytes) -> bytearray: + """27 байт AMBE (3×9) → 33-байтный бёрст: AMBE в [0:108]+[156:264], середина = 0.""" + out = bytearray(33) + out[0:13] = ambe27[0:13] + out[13] = ambe27[13] & 0xF0 + out[19] = ambe27[13] & 0x0F + out[20:33] = ambe27[14:27] + return out + + +def ambe_from_burst(burst33: bytes) -> bytes: + """Инверсия _put_ambe: из 33-байтного DMR голос-бёрста достать 27Б AMBE (3×9). + Середина ([13] млад.ниббл / [19] стар.ниббл) — sync/EMB, не AMBE.""" + b = bytes(burst33[:33]).ljust(33, b"\x00") + out = bytearray(27) + out[0:13] = b[0:13] + out[13] = (b[13] & 0xF0) | (b[19] & 0x0F) + out[14:27] = b[20:33] + return bytes(out) + + +_LCSS_BY_VSEQ = {1: LCSS_FIRST, 2: LCSS_CONT, 3: LCSS_CONT, 4: LCSS_LAST} + + +def voice_burst(ambe27: bytes, vseq: int, dst, src, unit=False, cc=1) -> bytes: + """Голосовой бёрст. vseq: 0=A(voice sync), 1..4=B..E(EMB+Embedded LC), 5=F(EMB null).""" + buf = _put_ambe(ambe27) + bits = _bytes_to_bits(buf) + if vseq % 6 == 0: # burst A — голосовой sync + bits[108:156] = _int_bits(SYNC_BS_VOICE, 48) + else: + v = vseq % 6 + if v in (1, 2, 3, 4): # B,C,D,E — EMB + фрагмент Embedded LC + frag = encode_emblc(lc9(dst, src, unit))[v] + emb = emb16(cc, _LCSS_BY_VSEQ[v]) + else: # F — EMB single, пустой фрагмент + frag = [0] * 32 + emb = emb16(cc, LCSS_SINGLE) + bits[108:116] = emb[0:8] + bits[116:148] = list(frag) + bits[148:156] = emb[8:16] + return _bits_to_bytes(bits) + + +# ── Самотест: RS-вектор + round-trip LC-заголовка через bptc.decode ── +if __name__ == "__main__": + ok = True + L = bytes.fromhex("000000002706155cc1") + raw = rs129(L) + hdr = bytes(p ^ m for p, m in zip(raw, _MASK_HEADER)) + trm = bytes(p ^ m for p, m in zip(raw, _MASK_TERM)) + for name, got, exp in (("RS raw", raw.hex(), "505575"), + ("RS header", hdr.hex(), "c6c3e3"), + ("RS term", trm.hex(), "c9ccec")): + good = got == exp + ok = ok and good + print(f"{name:12}= {got} (эталон {exp}) {'OK' if good else 'FAIL'}") + + # header round-trip: собрать бёрст → bptc.decode → те же LC9+parity + DST, SRC = 250, 1400001 + h = voice_lc_header(DST, SRC, cc=1) + dec = bptc.decode(h[:33] if len(h) >= 33 else h) # 12 байт: LC9 + masked parity + l = lc9(DST, SRC) + exp12 = l + bytes(p ^ m for p, m in zip(rs129(l), _MASK_HEADER)) + good = bytes(dec) == exp12 + ok = ok and good + print(f"{'HDR bptc rt':12}= {bytes(dec).hex()} (эталон {exp12.hex()}) {'OK' if good else 'FAIL'}") + + # Embedded LC B-E: 4 фрагмента по 32 бита + frags = encode_emblc(lc9(DST, SRC)) + good = all(len(frags[i]) == 32 for i in (1, 2, 3, 4)) + ok = ok and good + print(f"{'EmbLC frags':12}= {[len(frags[i]) for i in (1,2,3,4)]} {'OK' if good else 'FAIL'}") + + # AMBE round-trip: voice_burst(ambe27) → ambe_from_burst == ambe27 (для A и B-бёрстов) + for vseq in (0, 1, 5): + a27 = bytes(range(27)) + rt = ambe_from_burst(voice_burst(a27, vseq, DST, SRC)) + g = rt == a27 + ok = ok and g + print(f"{'AMBE rt v'+str(vseq):12}= {'OK' if g else 'FAIL ' + rt.hex()}") + print("=== ИТОГО:", "ВСЁ OK" if ok else "ЕСТЬ ОШИБКИ", "===") diff --git a/dmrsrv/log.py b/dmrsrv/log.py new file mode 100644 index 0000000..793058f --- /dev/null +++ b/dmrsrv/log.py @@ -0,0 +1,52 @@ +"""Логирование: в journald (stdout) и одновременно в живой журнал панели. + +Перенесённые из DMRhub модули зовут `get_logger(__name__)` и пишут через него — +эта прослойка позволяет не трогать их код и при этом показывать всё в панели. +""" +from __future__ import annotations + +import logging +import sys + +_sink = None # выставляется в main(): объект State с методом .event(kind, text) + + +def set_sink(state) -> None: + global _sink + _sink = state + + +class _PanelHandler(logging.Handler): + """Дублирует записи в журнал панели. Уровень -> цвет строки в интерфейсе.""" + + _KIND = { + logging.DEBUG: "info", + logging.INFO: "info", + logging.WARNING: "warn", + logging.ERROR: "error", + logging.CRITICAL: "error", + } + + def emit(self, record: logging.LogRecord) -> None: + if _sink is None: + return + try: + _sink.event(self._KIND.get(record.levelno, "info"), record.getMessage()) + except Exception: # логирование не должно ронять сервер + pass + + +def get_logger(name: str) -> logging.Logger: + return logging.getLogger(name) + + +def setup(level: str = "INFO") -> None: + root = logging.getLogger() + root.setLevel(getattr(logging, level.upper(), logging.INFO)) + if not any(isinstance(h, logging.StreamHandler) for h in root.handlers): + h = logging.StreamHandler(sys.stdout) + h.setFormatter(logging.Formatter("%(asctime)s %(levelname)-7s %(name)s: %(message)s", + datefmt="%H:%M:%S")) + root.addHandler(h) + if not any(isinstance(h, _PanelHandler) for h in root.handlers): + root.addHandler(_PanelHandler()) diff --git a/dmrsrv/panel.py b/dmrsrv/panel.py new file mode 100644 index 0000000..4fb304d --- /dev/null +++ b/dmrsrv/panel.py @@ -0,0 +1,255 @@ +"""Веб-панель: хотспоты, группы, живой журнал, настройки. + +Работает на stdlib `http.server` в отдельном потоке — так проще и надёжнее, чем +писать разбор HTTP поверх asyncio. Нагрузка тут никакая: несколько хотспотов и +одна-две открытые вкладки. + +Авторизация — HTTP Basic. Для внутренней сети этого достаточно, а формы с +сессиями потребовали бы куки, CSRF и хранение сессий ради того же результата. +""" +from __future__ import annotations + +import hmac +import json +import threading +from base64 import b64decode +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import urlparse + +from . import config as cfgmod +from .log import get_logger + +log = get_logger(__name__) + + +def _make_handler(ctx): + cfg, state = ctx["cfg"], ctx["state"] + drop_peer = ctx["drop_peer"] + + class Handler(BaseHTTPRequestHandler): + server_version = "dmrsrv" + + def log_message(self, fmt, *args): # не засорять journal каждым GET + pass + + # ── авторизация ───────────────────────────────────────────────────── + def _authed(self) -> bool: + hdr = self.headers.get("Authorization", "") + if not hdr.startswith("Basic "): + return False + try: + user, _, pwd = b64decode(hdr[6:]).decode("utf-8").partition(":") + except Exception: + return False + p = cfg["panel"] + # сравнение постоянного времени: иначе пароль подбирается по таймингу + return (hmac.compare_digest(user, p["user"]) + and hmac.compare_digest(pwd, p["password"])) + + def _deny(self) -> None: + self.send_response(401) + self.send_header("WWW-Authenticate", 'Basic realm="DMR"') + self.send_header("Content-Length", "0") + self.end_headers() + + def _json(self, obj, code: int = 200) -> None: + body = json.dumps(obj, ensure_ascii=False).encode("utf-8") + self.send_response(code) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _html(self, text: str) -> None: + body = text.encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + # ── маршруты ──────────────────────────────────────────────────────── + def do_GET(self) -> None: + if not self._authed(): + return self._deny() + path = urlparse(self.path).path + if path == "/": + return self._html(PAGE) + if path == "/api/state": + snap = state.snapshot() + snap["groups"] = cfg.get("groups", []) + snap["parrot"] = cfg.get("parrot", {}) + snap["port"] = cfg.get("port") + return self._json(snap) + self.send_error(404) + + def do_POST(self) -> None: + if not self._authed(): + return self._deny() + path = urlparse(self.path).path + try: + n = int(self.headers.get("Content-Length", "0")) + data = json.loads(self.rfile.read(n) or b"{}") + except (ValueError, json.JSONDecodeError): + return self._json({"error": "плохой JSON"}, 400) + + if path == "/api/groups/add": + try: + tg, slot = int(data["tg"]), int(data.get("slot", 2)) + except (KeyError, ValueError, TypeError): + return self._json({"error": "нужны tg и slot"}, 400) + if slot not in (1, 2): + return self._json({"error": "слот только 1 или 2"}, 400) + if any(int(g["tg"]) == tg and int(g["slot"]) == slot for g in cfg["groups"]): + return self._json({"error": f"TG{tg} на слоте {slot} уже есть"}, 400) + cfg["groups"].append({"tg": tg, "name": str(data.get("name", ""))[:40], "slot": slot}) + cfgmod.save(cfg) + state.event("info", f"панель: добавлена группа TG{tg} (слот {slot})") + return self._json({"ok": True, "groups": cfg["groups"]}) + + if path == "/api/groups/del": + try: + tg, slot = int(data["tg"]), int(data["slot"]) + except (KeyError, ValueError, TypeError): + return self._json({"error": "нужны tg и slot"}, 400) + before = len(cfg["groups"]) + cfg["groups"] = [g for g in cfg["groups"] + if not (int(g["tg"]) == tg and int(g["slot"]) == slot)] + if len(cfg["groups"]) == before: + return self._json({"error": "такой группы нет"}, 404) + cfgmod.save(cfg) + state.event("info", f"панель: удалена группа TG{tg} (слот {slot})") + return self._json({"ok": True, "groups": cfg["groups"]}) + + if path == "/api/settings": + changed = [] + if "passphrase" in data: + pw = str(data["passphrase"]) + if len(pw) < 4: + return self._json({"error": "пароль хотспотов слишком короткий"}, 400) + cfg["passphrase"] = pw + changed.append("пароль хотспотов (применится при следующем входе)") + if "parrot_enabled" in data: + cfg["parrot"]["enabled"] = bool(data["parrot_enabled"]) + changed.append("попугай " + ("включён" if cfg["parrot"]["enabled"] else "выключен")) + if not changed: + return self._json({"error": "нечего менять"}, 400) + cfgmod.save(cfg) + state.event("info", "панель: " + "; ".join(changed)) + return self._json({"ok": True, "changed": changed}) + + if path == "/api/peer/kick": + try: + pid = int(data["peer_id"]) + except (KeyError, ValueError, TypeError): + return self._json({"error": "нужен peer_id"}, 400) + if drop_peer(pid): + state.event("warn", f"панель: хотспот {pid} отключён вручную") + return self._json({"ok": True}) + return self._json({"error": "нет такого хотспота"}, 404) + + self.send_error(404) + + return Handler + + +def start(cfg: dict, state, drop_peer) -> ThreadingHTTPServer: + p = cfg["panel"] + srv = ThreadingHTTPServer((p["bind"], int(p["port"])), _make_handler( + {"cfg": cfg, "state": state, "drop_peer": drop_peer})) + srv.daemon_threads = True + threading.Thread(target=srv.serve_forever, name="panel", daemon=True).start() + log.info("панель: http://%s:%s (пользователь %s)", p["bind"], p["port"], p["user"]) + return srv + + +PAGE = """ + +DMR-сервер + +
+

DMR-сервер

+ подключение… +
+
+

Хотспоты

+

Группы

+
+
+ + + + +
+
+

Последние вызовы

+

Журнал

+
+ +""" diff --git a/dmrsrv/parrot.py b/dmrsrv/parrot.py new file mode 100644 index 0000000..c0ce6fd --- /dev/null +++ b/dmrsrv/parrot.py @@ -0,0 +1,150 @@ +"""DMRhub v2 · Паррот (эхотест) на TG 9990. + +Записывает групповой вызов на 9990 и после отпускания PTT проигрывает его ОБРАТНО — +абонент слышит себя («слышно ли в эфире»). + +Ключ: проигрываем ЗАПИСАННЫЕ кадры ДОСЛОВНО (payload [20:53] как есть), только с новым +stream_id. У кадров рации уже корректные sync/EMB/slot-type/AMBE — ничего не пересобираем +(реконструкция фрейминга давала «секундное молчание»: рация видела вызов, но не декодировала +голос). Это бит-в-бит повтор принятого — гарантированно декодируется. + +Доставка — НАПРЯМУЮ в хотспот вызвавшего (master._send на его addr), минуя роутер/гвард: + • не идёт через _route_dmr → не попадает обратно в feed (петли исключены by design); + • не портит last-heard рации и не триггерит анти-эхо мостов; + • слышит только тестирующий. Темп 60мс/кадр (как реальный слот TDMA). +""" +from __future__ import annotations + +import asyncio +import os +import random + +from .analyzer import DMR, FrameType, CALL_GROUP, CALL_UNIT, build_dmrd +from .server import PeerState +from .log import get_logger +from . import dmr_voice + +log = get_logger(__name__) + +# ДИАГНОСТИКА (env DMRHUB_PARROT_RECON): изоляция причины «тишины» v1. Дефолт пусто — +# дословный повтор (рабочий). Режимы: +# "voice" — пересобрать payload голос-бёрстов (voice_burst), заголовок/терм дословно; +# "header" — пересобрать заголовок/терминатор (voice_lc_header/terminator), голос дословно. +RECON = os.getenv("DMRHUB_PARROT_RECON", "") +RECON_VOICE = RECON in ("1", "voice") +RECON_HEADER = RECON == "header" + +PARROT_TG = 9990 +PARROT_SLOT = 2 # слот эхотеста (переопределяется из конфига) +PARROT_PEER = 0x0F990000 # peer_id в собранных DMRD (для direct-send не критичен) +FRAME_S = 0.06 # 60мс на кадр (шаг слота TDMA) +GAP_S = 1.0 # пауза после отпускания PTT перед проигрыванием +MAX_FRAMES = 4000 # предохранитель (~4 мин) от бесконечной записи + + +class _Rec: + __slots__ = ("caller_peer", "src", "frames", "group") + + def __init__(self, caller_peer: int, src: int, group: bool = True): + self.caller_peer = caller_peer + self.src = src + self.group = group # True=групповой эхотест на 9990, False=приватный + self.frames: list[tuple] = [] # (frame_type, data_type, payload33) в порядке приёма + + +class Parrot: + """Эхотест TG9990. `feed(dmr)` цепляется на on_dmrd мастера (как P25-мост).""" + + def __init__(self, master): + self.master = master + self._rec: dict[tuple, _Rec] = {} # (peer_id, stream_id) → запись + + def feed(self, dmr) -> None: + if dmr.peer_id == PARROT_PEER: # свой replay не писать (страховка) + return + if dmr.dst != PARROT_TG or dmr.slot != PARROT_SLOT: + return + if dmr.call_type not in (CALL_GROUP, CALL_UNIT): # групповой И приватный эхотест + return + is_group = dmr.call_type == CALL_GROUP + key = (dmr.peer_id, dmr.stream_id) + rec = self._rec.get(key) + if rec is None: + if dmr.is_terminator: # хвост без записи — игнор + return + rec = self._rec[key] = _Rec(dmr.peer_id, dmr.src, is_group) + + payload = bytes(dmr.raw[20:53]) + if len(payload) >= 33 and len(rec.frames) < MAX_FRAMES: + rec.frames.append((dmr.frame_type, int(dmr.data_type), payload[:33])) + + if dmr.is_terminator: + self._rec.pop(key, None) + if any(ft in (FrameType.VOICE, FrameType.VOICE_SYNC) for ft, _, _ in rec.frames): + self._schedule(rec) # проигрываем, только если был голос + + + def _schedule(self, rec: _Rec) -> None: + try: + asyncio.ensure_future(self._replay(rec)) + except RuntimeError: + log.warning("паррот: нет event-loop для проигрывания") + + # ── сборка кадров эха из записанных payload (тестируемо отдельно от пейсинга) ── + def build_frames(self, rec: _Rec, sid: int) -> list[DMR]: + """Пересобрать DMR-кадры эха. Группа — дословно (тот же payload, новый stream_id). + Приват — обратным адресом (src=9990 → dst=вызывающий, unit): дословный повтор не + годится, там dst=9990, рация не примет как «мне».""" + from analyzer import DataType + out: list[DMR] = [] + if not rec.group: # ── приватный эхотест ── + dst, src = rec.src, PARROT_TG # эхо приватом ОБРАТНО вызывающему + for i, (ft, dt, payload33) in enumerate(rec.frames): + if ft in (FrameType.VOICE, FrameType.VOICE_SYNC): + pl = dmr_voice.voice_burst(dmr_voice.ambe_from_burst(payload33), + dt, dst, src, unit=True) + elif ft == FrameType.DATA_SYNC and dt == DataType.VOICE_LC_HEADER: + pl = dmr_voice.voice_lc_header(dst, src, unit=True) + elif ft == FrameType.DATA_SYNC and dt == DataType.TERMINATOR_LC: + pl = dmr_voice.terminator(dst, src, unit=True) + else: + pl = payload33 + d = DMR(raw=b"", seq=i & 0xFF, src=src, dst=dst, peer_id=PARROT_PEER, + slot=PARROT_SLOT, call_type=CALL_UNIT, frame_type=ft, + data_type=dt, stream_id=sid) + d.raw = build_dmrd(d, pl) + out.append(d) + return out + for i, (ft, dt, payload33) in enumerate(rec.frames): + pl = payload33 + is_voice = ft in (FrameType.VOICE, FrameType.VOICE_SYNC) + if RECON_VOICE and is_voice: # пересобрать голос-бёрст (dt = vseq 0..5) + pl = dmr_voice.voice_burst(dmr_voice.ambe_from_burst(payload33), + dt, PARROT_TG, rec.src) + elif RECON_HEADER and ft == FrameType.DATA_SYNC and dt == DataType.VOICE_LC_HEADER: + pl = dmr_voice.voice_lc_header(PARROT_TG, rec.src) + elif RECON_HEADER and ft == FrameType.DATA_SYNC and dt == DataType.TERMINATOR_LC: + pl = dmr_voice.terminator(PARROT_TG, rec.src) + d = DMR(raw=b"", seq=i & 0xFF, src=rec.src, dst=PARROT_TG, peer_id=PARROT_PEER, + slot=PARROT_SLOT, call_type=CALL_GROUP, frame_type=ft, + data_type=dt, stream_id=sid) + d.raw = build_dmrd(d, pl) + out.append(d) + return out + + async def _replay(self, rec: _Rec) -> None: + peer = self.master.peers.get(rec.caller_peer) + if peer is None or peer.state != PeerState.CONNECTED or not self.master._send: + return + await asyncio.sleep(GAP_S) # дать рации перейти на приём + frames = self.build_frames(rec, random.getrandbits(32)) + log.info("паррот: эхо src=%s → хотспот %s, %d кадров", + rec.src, rec.caller_peer, len(frames)) + for i, d in enumerate(frames): + try: + self.master._send(d.raw, peer.addr) + except Exception: + log.exception("паррот: ошибка отправки эха") + return + if i < len(frames) - 1: + await asyncio.sleep(FRAME_S) # шаг слота 60мс между кадрами diff --git a/dmrsrv/router.py b/dmrsrv/router.py new file mode 100644 index 0000000..3a0e4e7 --- /dev/null +++ b/dmrsrv/router.py @@ -0,0 +1,102 @@ +"""Маршрутизация эфира: кому пересылать принятый кадр. + +Сервер намеренно простой — несколько хотспотов во внутренней сети: + • групповой вызов на разрешённую группу → всем остальным подключённым хотспотам; + • приватный вызов → только тому хотспоту, где абонента слышали последним; + • вызов на группу, которой нет в списке → никуда (в панели будет видно отказ). + +Здесь же переписывается peer_id кадра под каждого получателя — см. retarget(). +""" +from __future__ import annotations + +import time + +from .analyzer import CALL_GROUP +from .log import get_logger + +log = get_logger(__name__) + +RADIO_LOC_TTL = 15 * 60 # сколько помним, через какой хотспот слышали абонента + + +def retarget(raw: bytes, peer_id: int) -> bytes: + """Подставить в кадр номер ПОЛУЧАТЕЛЯ (байты 11-14). + + Смысл поля — «ретранслятор, которого касается этот кадр на этом соединении», + поэтому принимающий хотспот должен увидеть там свой номер. Так делают HBlink3 + (`send_peer`) и BrandMeister. MMDVMHost поле на приёме не проверяет, но + DMRGateway на многих Pi-Star стоит между ним и сетью, так что делаем как эталон. + """ + if len(raw) < 15: + return raw + return raw[:11] + peer_id.to_bytes(4, "big") + raw[15:] + + +class Router: + def __init__(self, cfg: dict, state) -> None: + self.cfg = cfg + self.state = state + self._radio_loc: dict[int, tuple[int, float]] = {} # dmr_id -> (peer_id, когда) + + # ── группы из конфига ─────────────────────────────────────────────────── + def allowed(self, tg: int, slot: int) -> bool: + for g in self.cfg.get("groups", []): + if int(g["tg"]) == tg and int(g.get("slot", slot)) == slot: + return True + return False + + def group_names(self) -> dict[int, str]: + return {int(g["tg"]): g.get("name", "") for g in self.cfg.get("groups", [])} + + # ── где слышали абонента (для приватных вызовов) ──────────────────────── + def set_radio_loc(self, dmr_id: int, peer_id: int) -> None: + self._radio_loc[dmr_id] = (peer_id, time.time()) + + def clear_radio_loc(self, dmr_id: int, peer_id: int) -> None: + cur = self._radio_loc.get(dmr_id) + if cur and cur[0] == peer_id: + del self._radio_loc[dmr_id] + + def locate(self, dmr_id: int) -> int | None: + cur = self._radio_loc.get(dmr_id) + if cur is None: + return None + peer_id, when = cur + if time.time() - when > RADIO_LOC_TTL: + del self._radio_loc[dmr_id] + return None + return peer_id + + def radio_locs(self) -> dict[int, int]: + return {rid: pid for rid, (pid, _) in self._radio_loc.items()} + + def active_calls(self) -> list: + return [] # состояние вызовов ведёт State; здесь пусто для совместимости + + # ── собственно маршрут ────────────────────────────────────────────────── + def route(self, peer_id: int, dmr, connected: list[int]) -> list[int]: + # запоминаем, через какой хотспот слышали передающего + if dmr.src: + self.set_radio_loc(dmr.src, peer_id) + + parrot_tg = int(self.cfg.get("parrot", {}).get("tg", 9990)) + if dmr.dst == parrot_tg: + return [] # попугай отвечает сам, напрямую вызывавшему + + if dmr.call_type == CALL_GROUP: + if not self.allowed(dmr.dst, dmr.slot): + return [] + # всем, кроме источника: возврат кадра отправителю вызывает петлю в эфире + return [pid for pid in connected if pid != peer_id] + + # Приватный вызов — туда, где абонента слышали последним. Если не слышали + # никогда (рация ещё не выходила в эфир) — рассылаем всем, иначе вызов молча + # уходит в никуда. Так же поступает HBlink4 («user cache with broadcast + # fallback»). Лишний трафик тут копеечный: хотспотов единицы. + target = self.locate(dmr.dst) + if target is None: + log.debug("приват %s→%s: абонент не известен, рассылаю всем", dmr.src, dmr.dst) + return [pid for pid in connected if pid != peer_id] + if target == peer_id: + return [] # абонент на том же хотспоте — эфир и так его слышит + return [target] if target in connected else [] diff --git a/dmrsrv/server.py b/dmrsrv/server.py new file mode 100644 index 0000000..ca62307 --- /dev/null +++ b/dmrsrv/server.py @@ -0,0 +1,525 @@ +"""DMRhub v2 · M2 — Homebrew-мастер: UDP-петля, рукопожатие, реестр пиров. + +Реализует серверную сторону протокола Homebrew (HBP) для хотспотов MMDVMHost: + RPTL(login) → RPTACK+salt → RPTK(auth sha256(salt+pass)) → RPTACK → RPTC(config) + → RPTACK → RPTPING↔MSTPONG (keepalive) + DMRD (эфир). RPTCL/таймаут → отключение. + +Ядро `HomebrewMaster.handle(data, addr)` — чистое (без сокета): принимает датаграмму, +возвращает список ответов-датаграмм. Это позволяет юнит-тестировать без сети. +Аутентификация — через колбэк `password_for(peer_id) → bytes|None` (позже: Postgres, +поиск хотспота по dmr_id + secret, [[dmrhub-id-base-essid]]). DMRD-маршрут — M3 (хук). +""" +from __future__ import annotations + +import hashlib +import os +import time +from dataclasses import dataclass, field +from typing import Callable + +from .analyzer import analyze, HBP, DMR, Kind, auth_hash, TAG_ACK, TAG_NAK, TAG_PONG +from .log import get_logger + +log = get_logger(__name__) + +# человекочитаемые причины отклонения гвардом (для лога панели) +_GUARD_REASON_RU = { + "peer_spoof": "подмена peer_id (кадр не от владельца этого ID)", + "own_id": "src совпал с peer_id (эхо/петля собственного ID)", + "src_out_of_range": "src вне допустимого диапазона DMR ID", + "unregistered_src": "незарегистрированный абонент (закрытая сеть)", +} + + +class PeerState: + NEW = "new" + CHALLENGE = "challenge" # выслан salt, ждём RPTK + AUTHED = "authed" # пароль принят, ждём RPTC + CONNECTED = "connected" # сконфигурирован, в сети + + +@dataclass +class Peer: + peer_id: int + addr: tuple + state: str = PeerState.NEW + salt: bytes = b"" + password: bytes = b"" + config: dict | None = None + last_seen: float = 0.0 + logins: int = 0 + transport: str = "homebrew" # homebrew | hytera + extra: dict = field(default_factory=dict) # адреса портов Hytera и пр. + + +def _ack(peer_id: int, extra: bytes = b"") -> bytes: + """RPTACK + 4-байт peer_id (или salt/иное в extra).""" + return TAG_ACK + (extra if extra else peer_id.to_bytes(4, "big")) + + +def _nak(peer_id: int) -> bytes: + return TAG_NAK + peer_id.to_bytes(4, "big") + + +def _pong(peer_id: int) -> bytes: + return TAG_PONG + peer_id.to_bytes(4, "big") + + +class HomebrewMaster: + """Ядро мастера. `password_for(peer_id)` → пароль (bytes) или None (неизвестен). + `on_dmrd(peer, dmr)` — хук маршрутизации эфира (M3). `now`/`rand` — для тестов.""" + + def __init__(self, password_for: Callable[[int], bytes | None], + router=None, guard=None, data_hub=None, ta_injector=None, + on_dmrd: Callable[[Peer, DMR], None] | None = None, + send: Callable[[bytes, tuple], None] | None = None, + encode_for: Callable[[Peer, DMR], bytes] | None = None, + timeout: float = 90.0, + on_unknown: Callable[[int], None] | None = None, + now: Callable[[], float] = time.monotonic, + rand: Callable[[int], bytes] = os.urandom): + self.password_for = password_for + # Незнакомый пир → просим перечитать реестр (как было в v1). Регистрация создаёт + # вирт. хотспот приложения, а Store перечитывается раз в 30с — без этого новичок + # до полуминуты получает отказы и может сдаться. + self.on_unknown = on_unknown + self._unknown_at: float = 0.0 + self.router = router + self.guard = guard + self.data_hub = data_hub # M5: сборка SMS/GPS/TA из эфира + self.ta_injector = ta_injector # подмешивание TA (позывной из БД) в эфир без TA + self.bridges: dict[int, object] = {} # peer_id → Bridge (внешние сети) + self.on_dmrd = on_dmrd + self.encode_for = encode_for # (target_peer, dmr) → байты в формате транспорта цели + self._send = send # доставка кадров чужим пирам (asyncio: transport.sendto) + self._send_hytera = None # доставка IPSC Hytera-пирам (сокет 50001); ставит hytera_run + self.timeout = timeout + self._now = now + self._rand = rand + self.peers: dict[int, Peer] = {} + self._last_guard_drop: dict[int, int] = {} # peer_id → последний дропнутый stream (лог по потоку) + self._last_call_log: dict[int, int] = {} # peer_id → последний залогированный stream (CALL start) + self._last_spoof_at: float = 0.0 # троттл лога анти-спуфа DMRD (по времени, без роста dict) + # Кандидатские логины: RPTL по живой сессии, но с ДРУГОГО адреса (смена IP/NAT либо + # спуф). Живую сессию не трогаем — держим челлендж отдельно и переносим пира на новый + # адрес ТОЛЬКО после верного RPTK. Ограничены по TTL и размеру (без роста под флудом). + self._pending: dict[int, dict] = {} + self._pending_ttl: float = 30.0 + self._pending_max: int = 64 + self._call_stats: dict[tuple, dict] = {} # (peer_id,slot) → статистика текущего потока (длит./BER/loss/тип) + + # ── обработка датаграммы: вернуть список ответов ── + def handle(self, data: bytes, addr: tuple) -> list[bytes]: + pkt = analyze(data) + if isinstance(pkt, DMR): + return self._on_dmrd(pkt, addr) + if not isinstance(pkt, HBP): + return [] + pid = pkt.peer_id + if pkt.kind == Kind.LOGIN: + return self._on_login(pid, addr) + if pkt.kind == Kind.AUTH: + return self._on_auth(pid, addr, pkt.auth or b"") + if pkt.kind == Kind.CONFIG: + return self._on_config(pid, addr, pkt.config) + if pkt.kind == Kind.PING: + return self._on_ping(pid, addr) + if pkt.kind in (Kind.CLOSE,): + # АНТИ-СПУФ управляющих (аудит 2026-07-25): RPTCL принимался БЕЗ проверки + # адреса — знающий peer_id (он выводится из публичного DMR ID) выбивал чужой + # хотспот из сети одним UDP-пакетом, а повторяя — держал офлайн. Для app-пиров + # это дополнительно стирало last-heard, и приваты абоненту переставали доходить. + # Логика та же, что у DMRD: закрыть сессию может только тот адрес, с которого + # пир прошёл рукопожатие. У app/bridge-пиров addr фиктивный (("app", pid) и + # т.п.) и по UDP они не закрываются вовсе — их снимают свои обработчики. + p = self.peers.get(pid) + if p is None: + return [] + if p.transport in ("app", "bridge"): + log.info("RPTCL peer=%s addr=%s — игнор: пир транспорта %s не закрывается по UDP", + pid, addr, p.transport) + return [] + if p.addr != addr: + log.warning("RPTCL peer=%s с ЧУЖОГО адреса %s (пир на %s) — игнор", + pid, addr, p.addr) + return [] + self.peers.pop(pid, None) + if pid % 100 == 99 and self.router is not None: + self.router.clear_radio_loc(pid // 100, pid) + return [] + return [] + + def _on_login(self, pid: int, addr: tuple) -> list[bytes]: + pw = self.password_for(pid) + if pw is None: + # v1-поведение: refresh + NAK. Пир ретрайнет через несколько секунд и пройдёт, + # не дожидаясь планового перечитывания Store. + # ТРОТТЛИНГ ОБЯЗАТЕЛЕН: без него спам чужими peer_id = шторм запросов в PG. + if self.on_unknown is not None: + t = self._now() + if t - self._unknown_at > 10.0: + self._unknown_at = t + try: + self.on_unknown(pid) + except Exception: + log.exception("on_unknown") + log.warning("LOGIN отказ: незарегистрированный peer=%s addr=%s (реестр перечитан)", pid, addr) + return [_nak(pid)] # незарегистрированный ID — отказ + salt = self._rand(4) + p = self.peers.get(pid) + # ЗАЩИТА ЖИВОЙ СЕССИИ (аудит 2026-07-25): раньше RPTL с ЛЮБОГО адреса сразу + # перетирал p.addr/p.state работающего пира — его DMRD после этого отбрасывались + # анти-спуфом, эфир хотспота пропадал (для этого даже пароль не требовался). + # Теперь логин с чужого адреса по живой сессии = кандидат: сессию не трогаем, + # переносим пира на новый адрес только после верного RPTK (см. _on_auth). + if (p is not None and p.state == PeerState.CONNECTED and p.addr != addr + and p.transport not in ("app", "bridge")): + now = self._now() + for k in [k for k, v in self._pending.items() if now - v["ts"] > self._pending_ttl]: + self._pending.pop(k, None) + if len(self._pending) < self._pending_max: + self._pending[pid] = {"addr": addr, "salt": salt, "password": pw, "ts": now} + log.info("LOGIN peer=%s с НОВОГО адреса %s (живая сессия на %s) → челлендж-кандидат", + pid, addr, p.addr) + return [_ack(pid, salt)] + p = p or Peer(pid, addr) + p.addr = addr + p.salt = salt + p.password = pw + p.state = PeerState.CHALLENGE + p.last_seen = self._now() + p.logins += 1 + self.peers[pid] = p + log.info("LOGIN peer=%s addr=%s → выслан salt (челлендж)", pid, addr) + return [_ack(pid, salt)] # RPTACK + salt + + def _on_auth(self, pid: int, addr: tuple, sha_recv: bytes) -> list[bytes]: + # Кандидатский логин (RPTL с нового адреса по живой сессии): проверяем пароль, + # НЕ трогая работающего пира. Верный → переносим сессию на новый адрес (легитимная + # смена IP/NAT). Неверный → выбрасываем только кандидата, эфир пира не страдает. + pend = self._pending.get(pid) + if pend is not None and pend["addr"] == addr: + self._pending.pop(pid, None) + if sha_recv[:32] != auth_hash(pend["salt"], pend["password"]): + log.warning("AUTH отказ (кандидат): неверный пароль peer=%s addr=%s — " + "живая сессия сохранена", pid, addr) + return [_nak(pid)] + p = self.peers.get(pid) or Peer(pid, addr) + old = p.addr + p.addr = addr + p.salt = pend["salt"] + p.password = pend["password"] + p.state = PeerState.AUTHED + p.last_seen = self._now() + p.logins += 1 + self.peers[pid] = p + log.info("AUTH ok (кандидат): peer=%s переехал %s → %s", pid, old, addr) + return [_ack(pid)] + p = self.peers.get(pid) + if p is None or p.state != PeerState.CHALLENGE: + return [_nak(pid)] + expected = auth_hash(p.salt, p.password) + if sha_recv[:32] != expected: + self.peers.pop(pid, None) # неверный пароль — сброс + log.warning("AUTH отказ: неверный пароль peer=%s addr=%s", pid, addr) + return [_nak(pid)] + p.state = PeerState.AUTHED + p.last_seen = self._now() + log.info("AUTH ok: хотспот peer=%s addr=%s подтвердил пароль → жду конфиг (RPTC)", pid, addr) + return [_ack(pid)] + + def _on_config(self, pid: int, addr: tuple, config: dict | None) -> list[bytes]: + p = self.peers.get(pid) + if p is None or p.state not in (PeerState.AUTHED, PeerState.CONNECTED): + return [_nak(pid)] + p.config = config + p.state = PeerState.CONNECTED + p.last_seen = self._now() + cs = (config or {}).get("callsign", "?") if config else "?" + log.info("CONFIG peer=%s callsign=%s → CONNECTED (пиров: %d)", + pid, cs, len(self.connected_peers())) + # Онлайн-приложение (essid 99 = виртуальная рация) регистрирует свою локацию + # СРАЗУ на коннекте, чтобы приватный вызов доходил, даже если оно ещё не + # передавало (route() ставит локацию лишь при передаче). База = pid//100. + if pid % 100 == 99 and self.router is not None: + self.router.set_radio_loc(pid // 100, pid) + return [_ack(pid)] + + def _on_ping(self, pid: int, addr: tuple) -> list[bytes]: + p = self.peers.get(pid) + if p is None or p.state != PeerState.CONNECTED: + return [_nak(pid)] + # Адрес пинга должен совпадать с адресом сессии (аудит 2026-07-25). Раньше пинг с + # ЛЮБОГО адреса продлевал сессию — при смене порта NAT хотспот получал PONG и считал + # себя в сети, а его DMRD резались анти-спуфом: пир «онлайн», но глухонемой навсегда. + # NAK заставляет его перелогиниться (и переехать на новый адрес по паролю, см. + # кандидатский путь в _on_login/_on_auth). Спуферу NAK не даёт ничего: чужая сессия + # не продлевается и не рвётся. + if p.addr != addr and p.transport not in ("app", "bridge"): + log.info("PING peer=%s с чужого/нового адреса %s (сессия на %s) → NAK, пусть перелогинится", + pid, addr, p.addr) + return [_nak(pid)] + p.last_seen = self._now() + return [_pong(pid)] + + def _on_dmrd(self, dmr: DMR, addr: tuple) -> list[bytes]: + # АНТИ-СПУФ (C1): DMRD принимаем ТОЛЬКО с адреса, с которого пир прошёл рукопожатие + # RPTL/RPTK (пароль). Сами DMRD-кадры не подписаны — единственная привязка к + # аутентификации это исходный addr. Без этой проверки любой хост интернета, зная + # peer_id хотспота (7 цифр, светятся в lastheard) и любой зарегистрированный src, + # инжектит голос/данные в любой TG и ПЕРЕХВАТЫВАЕТ приваты (router на каждом кадре + # пишет _radio_loc[src]=peer). peers keyed by peer_id → addr==p.addr гарантирует, + # что кадр пришёл именно от владельца этого peer_id (и peer_id не подставлен). + p = self.peers.get(dmr.peer_id) + if p is None or p.state != PeerState.CONNECTED or p.addr != addr: + t = self._now() + if t - self._last_spoof_at > 5.0: # троттл по времени (без роста dict под флудом) + self._last_spoof_at = t + log.warning("DMRD отброшен (спуф/чужой addr): peer=%s src=%s→%s от %s (ожидался %s)", + dmr.peer_id, dmr.src, dmr.dst, addr, p.addr if p else None) + return [] + self._route_dmr(dmr) # Homebrew DMRD-путь = общий маршрут + return [] + + # общий вход для эфира из ЛЮБОГО транспорта (Homebrew DMRD или мост Hytera→DMR) + def ingest_dmr(self, dmr: DMR) -> None: + """Прогнать готовый DMR-кадр через гвард+маршрут (для Hytera-транспорта).""" + self._route_dmr(dmr) + + def _route_dmr(self, dmr: DMR) -> None: + p = self.peers.get(dmr.peer_id) + if p is None or p.state != PeerState.CONNECTED: + return + p.last_seen = self._now() + # гвард НЕ трогает мосты: их трафик фильтрует сам мост (only_registered), а не мастер + # ([[dmrhub-closed-network-src]]). Иначе SMK/ADN-юзеры без нашей регистрации режутся. + if self.guard is not None and p.transport != "bridge": + allow, reason = self.guard.check(dmr.peer_id, dmr) + if not allow: + if self._last_guard_drop.get(dmr.peer_id) != dmr.stream_id: # лог по потоку, не по кадру + self._last_guard_drop[dmr.peer_id] = dmr.stream_id + tgt = ("TG %s" % dmr.dst) if dmr.call_type == "group" else ("приват %s" % dmr.dst) + log.warning("ГВАРД отклонил кадр: %s — src=%s → %s через хотспот peer=%s " + "(код=%s, поток %08x)", _GUARD_REASON_RU.get(reason, reason), + dmr.src, tgt, dmr.peer_id, reason, dmr.stream_id) + return + if dmr.is_voice_start and self._last_call_log.get(dmr.peer_id) != dmr.stream_id: + self._last_call_log[dmr.peer_id] = dmr.stream_id # лог РАЗ В ПОТОК (не на каждый повторный заголовок late-entry) + tgt = ("группа TG %s" % dmr.dst) if dmr.call_type == "group" else ("приват на %s" % dmr.dst) + log.info("ЭФИР старт: src=%s → %s · слот TS%s · хотспот peer=%s (поток %08x)", + dmr.src, tgt, dmr.slot, dmr.peer_id, dmr.stream_id) + ctype = "voice" if dmr.is_voice_start else "data" + if self.data_hub is not None and getattr(self.data_hub, "on_call", None): + try: + self.data_hub.on_call({"src": dmr.src, "dst": dmr.dst, + "is_group": dmr.call_type == "group", "slot": dmr.slot, + "ctype": ctype}) + except Exception: + log.debug("on_call emit не выполнен", exc_info=True) + self._call_stats[(dmr.peer_id, dmr.slot)] = { + "stream_id": dmr.stream_id, "src": dmr.src, "dst": dmr.dst, + "is_group": dmr.call_type == "group", "slot": dmr.slot, "ctype": ctype, + "start": self._now(), "last": self._now(), "frames": 0, "ber_sum": 0.0, "ber_n": 0} + if self.data_hub is not None: # M5: собрать SMS/GPS/TA + self.data_hub.feed(dmr) + try: # статистика потока (длит./BER/loss) + финал по терминатору + self._accumulate_call(dmr) + except Exception: + log.debug("accumulate_call сбой", exc_info=True) + # TA-инжект: рация без своего TA → подмешиваем позывной из БД в embedded LC. + # ПОСЛЕ data_hub.feed (разбор видит оригинал источника) и ДО рассылки/tap — + # чтобы и пиры, и voice_bridge, и Hytera-encode_for получили кадр с TA. + if self.ta_injector is not None: + try: + self.ta_injector.process(dmr) + except Exception: + log.debug("ta_inject сбой", exc_info=True) + if self.on_dmrd: + self.on_dmrd(p, dmr) + if self.router is not None: + connected = [pid for pid, pp in self.peers.items() if pp.state == PeerState.CONNECTED] + targets = self.router.route(dmr.peer_id, dmr, connected) + for tid in targets: + if tid in self.bridges: # мост во внешнюю сеть + try: + self.bridges[tid].feed_from_master(dmr) + except Exception: + log.exception("мост %s: ошибка отправки", tid) + continue + tgt = self.peers.get(tid) + if tgt and tgt.transport == "app": + continue # виртуальный app-пир: доставку делает voice_bridge (tap) + if tgt and self._send: + frame = self.encode_for(tgt, dmr) if self.encode_for else dmr.raw + if getattr(tgt, "transport", None) == "hytera" and self._send_hytera: + self._send_hytera(frame, tgt.addr) # IPSC через Hytera-сокет 50001 + else: + self._send(frame, tgt.addr) + if targets and (dmr.is_voice_start or dmr.data_type == 6): + log.debug("route peer=%s src=%s→%s → %d получателям %s", + dmr.peer_id, dmr.src, dmr.dst, len(targets), targets) + + def _accumulate_call(self, dmr) -> None: + key = (dmr.peer_id, dmr.slot) + st = self._call_stats.get(key) + if st is None or st["stream_id"] != dmr.stream_id: + return + st["frames"] += 1 + st["last"] = self._now() + if dmr.ber is not None: + st["ber_sum"] += dmr.ber; st["ber_n"] += 1 + if dmr.is_terminator: + self._finalize_call(key) + + def _finalize_call(self, key) -> None: + st = self._call_stats.pop(key, None) + if not st or st["frames"] <= 0: + return + dur_ms = int(max(0.0, st["last"] - st["start"]) * 1000) + ber = round(st["ber_sum"] / st["ber_n"], 1) if st["ber_n"] else None + loss = None + if st["ctype"] == "voice" and dur_ms > 0: + expected = int(round(dur_ms / 60.0)) + 1 # голос ~60 мс/кадр + if expected > 0: + loss = max(0, min(100, round((expected - st["frames"]) / expected * 100))) + if self.data_hub is not None and getattr(self.data_hub, "on_call_end", None): + try: + self.data_hub.on_call_end({"src": st["src"], "dst": st["dst"], + "is_group": st["is_group"], "slot": st["slot"], + "ctype": st["ctype"], "duration_ms": dur_ms, + "ber": ber, "loss": loss}) + except Exception: + log.debug("on_call_end emit не выполнен", exc_info=True) + + # ── обслуживание: отключить протухшие пиры ── + def reap(self) -> list[int]: + """Удалить пиры без активности дольше timeout. Вернуть их id. + Мосты (transport=bridge) НЕ реапим: они исходящие клиенты и не пингуют наш мастер, + их last_seen не обновляется — иначе мост-пир пропадает из маршрута (эфир us→сеть рвётся).""" + now_ts = self._now() # финализировать потоки без терминатора (потеря UDP) + for k in list(self._call_stats): + if now_ts - self._call_stats[k]["last"] > 1.5: + self._finalize_call(k) + cutoff = self._now() - self.timeout + dead = [pid for pid, p in self.peers.items() + if p.last_seen < cutoff and p.transport not in ("bridge", "app")] + for pid in dead: + self.peers.pop(pid, None) + if pid % 100 == 99 and self.router is not None: + self.router.clear_radio_loc(pid // 100, pid) # app ушёл — снять локацию + # Динамическую подписку НЕ снимаем: для отключённого пира она безвредна (subscribers() + # фильтрует по connected) и переживает реконнект → приложению/хотспоту не нужен повторный + # subscribe-берст (кёрчанк), который слышат все на TG. Подписка также в БД (dynamic_tg). + return dead + + def connected_peers(self) -> list[Peer]: + return [p for p in self.peers.values() if p.state == PeerState.CONNECTED] + + def hotspots_by_ip(self, ip: str) -> list[int]: + """peer_id подключённых homebrew-хотспотов с данным IP — для привязки P25/NXDN/YSF к + аутентифицированному DMR-хотспоту (эти протоколы своей аутентификации не имеют). + Пусто = IP неизвестен (не наш хотспот) → трафик не принимаем. Несколько = NAT + (неоднозначно) → атрибутируем на общий шлюз, но принимаем (IP известен).""" + return [pid for pid, p in self.peers.items() + if p.state == PeerState.CONNECTED and p.transport == "homebrew" + and p.addr and p.addr[0] == ip] + + # ── виртуальный «хотспот» онлайн-приложения (id пользователя×100+essid) ── + # Даёт системе единый вид: приложение онлайн как рация+хотспот; приватный поток + # роутится к нему (router.locate). Доставку кадров делает voice_bridge (tap), _send пропускается. + def add_app_peer(self, peer_id: int, callsign: str | None = None) -> "Peer | None": + p = self.peers.get(peer_id) + if p is not None and p.transport != "app": + return None # на этом essid реальный хотспот — не трогаем + if p is None: + p = Peer(peer_id, ("app", peer_id)) + self.peers[peer_id] = p + p.transport = "app" + p.state = PeerState.CONNECTED + p.last_seen = self._now() + p.config = {"callsign": callsign, "location": "Приложение"} + return p + + def remove_app_peer(self, peer_id: int) -> None: + p = self.peers.get(peer_id) + if p is not None and p.transport == "app": + self.peers.pop(peer_id, None) + + def snapshot(self) -> dict: + """Срез состояния для статус-панели (пиры, счётчики, гвард).""" + peers = [] + for p in self.peers.values(): + cfg = p.config or {} + peers.append({ + "peer_id": p.peer_id, "state": p.state, "transport": p.transport, + "addr": f"{p.addr[0]}:{p.addr[1]}" if p.addr else None, + "callsign": cfg.get("callsign"), "location": cfg.get("location"), + "last_seen": round(self._now() - p.last_seen, 1), + }) + peers.sort(key=lambda x: (x["state"] != "connected", x["peer_id"])) + bridges = [] + for nid, br in self.bridges.items(): + cfg = getattr(br, "cfg", None) + cl = getattr(br, "client", None) + bridges.append({ + "network_id": nid, + "name": getattr(cfg, "name", "?"), + "type": getattr(cfg, "type", "?"), + "host": f"{getattr(cfg,'host','?')}:{getattr(cfg,'port','?')}", + "state": getattr(cl, "state", "idle") if cl else "idle", + "maps": [{"our": m.our_tg, "remote": m.remote_tg, "slot": m.our_slot} + for m in getattr(cfg, "maps", [])], + }) + return { + "peers_total": len(self.peers), + "peers_connected": len([p for p in self.connected_peers() if p.transport not in ("bridge", "app")]), + "apps_online": len([p for p in self.peers.values() if p.transport == "app"]), + "peers": [p for p in peers if p["transport"] not in ("bridge", "app")], + "bridges": bridges, + "guard_dropped": dict(self.guard.dropped) if self.guard else {}, + "active_calls": self.router.active_calls() if self.router else [], + # рация → peer_id последнего приёма (для карты: кого «слышит» каждый хотспот) + "radio_locs": ({str(k): v for k, v in self.router.radio_locs().items()} + if self.router else {}), + } + + +# ── asyncio UDP-обвязка (сетевой слой поверх чистого ядра) ── +import asyncio + + +class _MasterProtocol(asyncio.DatagramProtocol): + def __init__(self, master: HomebrewMaster): + self.master = master + self.transport = None + + def connection_made(self, transport): + self.transport = transport + self.master._send = transport.sendto # пересылка кадров чужим пирам + + def datagram_received(self, data: bytes, addr: tuple): + try: + for reply in self.master.handle(data, addr): + self.transport.sendto(reply, addr) + except Exception: # один битый пакет не роняет мастер + pass + + def error_received(self, exc): + pass + + def connection_lost(self, exc): + pass + + +async def run_server(master: HomebrewMaster, host: str = "0.0.0.0", + port: int = 62031, reap_interval: float = 30.0): + """Запустить UDP-мастер: слушать host:port, периодически чистить протухшие пиры.""" + loop = asyncio.get_running_loop() + transport, _ = await loop.create_datagram_endpoint( + lambda: _MasterProtocol(master), local_addr=(host, port)) + try: + while True: + await asyncio.sleep(reap_interval) + master.reap() + finally: + transport.close() diff --git a/dmrsrv/state.py b/dmrsrv/state.py new file mode 100644 index 0000000..3a4e573 --- /dev/null +++ b/dmrsrv/state.py @@ -0,0 +1,130 @@ +"""Общее состояние сервера: подключённые хотспоты, журнал событий, счётчики. + +Читается из потока веб-панели, пишется из цикла asyncio, поэтому всё под +одним замком. Объём данных крошечный (десятки хотспотов, сотни строк лога), +так что простой Lock дешевле и надёжнее любых очередей. +""" +from __future__ import annotations + +import threading +import time +from collections import deque +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class Peer: + """Подключённый хотспот/ретранслятор.""" + + peer_id: int + addr: tuple[str, int] + callsign: str = "" + rx_freq: str = "" + tx_freq: str = "" + colorcode: str = "" + location: str = "" + description: str = "" + url: str = "" + software: str = "" + connected_at: float = field(default_factory=time.time) + last_seen: float = field(default_factory=time.time) + # стадия рукопожатия: login -> auth -> config -> ready + stage: str = "login" + salt: int = 0 + rx_packets: int = 0 + tx_packets: int = 0 + + def as_dict(self) -> dict[str, Any]: + now = time.time() + return { + "peer_id": self.peer_id, + "addr": f"{self.addr[0]}:{self.addr[1]}", + "callsign": self.callsign, + "location": self.location, + "description": self.description, + "software": self.software, + "colorcode": self.colorcode, + "rx_freq": self.rx_freq, + "tx_freq": self.tx_freq, + "stage": self.stage, + "online_for": int(now - self.connected_at), + "idle": round(now - self.last_seen, 1), + "rx_packets": self.rx_packets, + "tx_packets": self.tx_packets, + } + + +@dataclass +class Call: + """Идущая или завершившаяся передача.""" + + stream_id: int + src_id: int + dst_id: int + slot: int + group: bool + peer_id: int + started: float = field(default_factory=time.time) + ended: float = 0.0 + frames: int = 0 + + def as_dict(self) -> dict[str, Any]: + end = self.ended or time.time() + return { + "stream_id": self.stream_id, + "src_id": self.src_id, + "dst_id": self.dst_id, + "slot": self.slot, + "type": "группа" if self.group else "приват", + "peer_id": self.peer_id, + "started": self.started, + "duration": round(end - self.started, 1), + "frames": self.frames, + "active": self.ended == 0.0, + } + + +class State: + def __init__(self, log_keep: int = 500) -> None: + self._lock = threading.Lock() + self.peers: dict[int, Peer] = {} + self.log: deque[dict[str, Any]] = deque(maxlen=log_keep) + self.calls: deque[Call] = deque(maxlen=50) + self.active: dict[tuple[int, int], Call] = {} # (slot, stream_id) -> Call + self.started = time.time() + + # ── журнал ────────────────────────────────────────────────────────────── + def event(self, kind: str, text: str) -> None: + """kind: info | call | peer | warn | error — панель красит по нему.""" + with self._lock: + self.log.append({"t": time.time(), "kind": kind, "text": text}) + print(f"[{kind}] {text}", flush=True) # заодно в journalctl + + # ── хотспоты ──────────────────────────────────────────────────────────── + def peer_add(self, peer: Peer) -> None: + with self._lock: + self.peers[peer.peer_id] = peer + + def peer_del(self, peer_id: int) -> Peer | None: + with self._lock: + return self.peers.pop(peer_id, None) + + def peer_get(self, peer_id: int) -> Peer | None: + with self._lock: + return self.peers.get(peer_id) + + def peers_ready(self) -> list[Peer]: + with self._lock: + return [p for p in self.peers.values() if p.stage == "ready"] + + # ── снимок для панели ─────────────────────────────────────────────────── + def snapshot(self, log_since: float = 0.0) -> dict[str, Any]: + with self._lock: + return { + "uptime": int(time.time() - self.started), + "peers": [p.as_dict() for p in self.peers.values()], + "calls": [c.as_dict() for c in reversed(self.calls)], + "log": [e for e in self.log if e["t"] > log_since], + "now": time.time(), + } diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md new file mode 100644 index 0000000..bbeddec --- /dev/null +++ b/docs/PROTOCOL.md @@ -0,0 +1,93 @@ +# Homebrew/MMDVM — что реально в эфире + +Сверено с исходниками MMDVMHost (`DMRDirectNetwork.cpp`), DMRGateway, HBlink3 (`hblink.py`) +и спецификацией DL5DI/G4KLX/DG1HT 2015 года. **Где спецификация расходится с кодом — +авторитетен код:** именно он крутится на каждом Pi-Star, а спецификацию никто до конца +не реализовал. + +## ⚠️ Тест-вектор из спецификации ОШИБОЧЕН — не «чините» по нему код + +Спецификация приводит: salt `0A7ED498`, пароль `DL5DI` → +`cbf0e29abbd11c6573825d36a664e3064441f91fc815ac5dc5ca570d4c4a5f85`. + +Это значение получается из **ASCII-строки** `"0A7ED498"`, хотя тремя абзацами выше та же +спецификация требует «трактовать 32-битное значение как битовый шаблон, а не как число». +Проверено: + +``` +sha256(b'\x0a\x7e\xd4\x98' + b'DL5DI') = a763d5c7… ← так делают ВСЕ реализации +sha256(b'0A7ED498' + b'DL5DI') = cbf0e29a… ← так написано в тест-векторе +``` + +Сырые байты используют MMDVMHost (`memcpy(in, m_salt, 4)`), HBlink3 (`bytes_4(SALT) + PASSPHRASE`) +и наш `analyzer.auth_hash`. **Выбивается только бумажка.** Если «поправить» код под +тест-вектор — вход перестанут проходить все хотспоты. + +## Расхождения спецификации и реальности + +| | спецификация | на самом деле | +|---|---|---| +| Ответ мастера | `MSTACK` | **`RPTACK`** | +| Keepalive | хотспот `MSTPING`, мастер `RPTPONG` | **хотспот `RPTPING`, мастер `MSTPONG`** | +| RPTL | «семь байт» | **8 байт** | +| Порядок в RPTC | callsign, затем id | **id@4, затем callsign@8** | +| Описание/слоты | description 20 байт, поля слотов нет | **description 19@78 + slots 1@97** | +| Длина DMRD | 53 | **53 ядро + BER/RSSI = 55 в эфире** | +| Маски байта 15 | `0x01/0x02/0x0C/0xF0` | **`0x80/0x40/0x30/0x0F`** | +| Период пинга | «раз в минуту» | **5–15 с** | + +## Кадр DMRD + +| Смещение | Длина | Поле | +|---:|---:|---| +| 0 | 4 | `DMRD` | +| 4 | 1 | последовательность | +| 5 | 3 | src (uint24 BE) | +| 8 | 3 | dst — talkgroup или абонент (uint24 BE) | +| 11 | 4 | **peer_id (uint32 BE) — переписывается на получателя** | +| 15 | 1 | биты | +| 16 | 4 | stream_id — **непрозрачные байты, не число** | +| 20 | 33 | бёрст DMR | +| 53 | 1 | BER (расширение G4KLX) | +| 54 | 1 | RSSI (расширение G4KLX) | + +Байт 15: `0x80` слот (0=TS1, 1=TS2) · `0x40` тип вызова (0=группа, 1=приват) · +`0x30` тип кадра (`00` голос, `01` voice sync, `10` data sync) · `0x0F` тип данных +либо номер голосового кадра. Начало передачи — data sync + `0x01`, конец — data sync + `0x02`. + +## Три вещи, которые легко испортить + +1. **Переписывать peer_id (11–14) на номер получателя.** Смысл поля — «ретранслятор, + которого касается кадр на этом соединении». Делает `router.retarget()` через хук + `encode_for`. MMDVMHost поле на приёме не проверяет, но DMRGateway на многих Pi-Star + стоит между ним и сетью. +2. **Не возвращать кадр отправителю** — мгновенная петля в эфире на дуплексном + ретрансляторе. Исключение — попугай, он отвечает намеренно и напрямую. +3. **Отправлять 55 байт, принимать 53/55/71.** DMRGateway молча выбрасывает 53-байтные + кадры (`length == HOMEBREW_DATA_PACKET_LENGTH`). + +Терминаторы и заголовки регулярно теряются в эфире, поэтому начало передачи определяется +ещё и сменой `stream_id`, а конец — таймаутом 360 мс (6 пропущенных голосовых кадров). + +## Попугай (TG 9990) + +Записанные кадры проигрываются **дословно** — `payload[20:53]` как есть, меняется только +`stream_id`. Это не оптимизация, а необходимость: в DMRhub пробовали пересобирать фрейминг, +и получалось «секундное молчание» — рация видела вызов, но голос не декодировала. У кадров +из рации уже корректные sync/EMB/slot-type/AMBE, трогать их нечего. + +Исключение — **приватный** эхотест: там нужен обратный адрес (`src=9990 → dst=вызывающий`), +дословный повтор не годится, потому что в нём `dst=9990` и рация не примет вызов как «мне». +Только для этого случая пересобираются LC-заголовок и терминатор (`dmr_voice` + `bptc`). + +Темп проигрывания — 60 мс на кадр (шаг слота TDMA), пауза перед началом — 1 с. +Доставка — напрямую в хотспот вызывавшего, минуя маршрутизатор: так исключены петли +и не портится last-heard. + +## Источники + +- Спецификация: `DMRplus_IPSC_Protocol_for_HB_repeater.pdf` (DL5DI/G4KLX/DG1HT, 2015-07-26) +- MMDVMHost `DMRDirectNetwork.cpp` @ `7008c62` (до удаления прямого доступа к мастерам) +- DMRGateway `DMRNetwork.cpp`, HBlink3 `hblink.py`/`const.py`/`playback.py` +- Проверенная реализация: `viktor/dmrhubv2`, откуда взяты `analyzer.py`, `server.py`, + `parrot.py`, `dmr_voice.py`, `bptc.py` diff --git a/install.sh b/install.sh new file mode 100644 index 0000000..18a00d3 --- /dev/null +++ b/install.sh @@ -0,0 +1,136 @@ +#!/usr/bin/env bash +# Установка DMR-сервера. Debian/Ubuntu, systemd. Зависимостей нет — только python3. +# +# sudo bash install.sh +# +# Повторный запуск безопасен: конфиг и пароли не перезаписываются. +set -euo pipefail + +APP=dmr-server +DEST=/opt/$APP +CFGDIR=/etc/$APP +USER=dmrsrv +SRC="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +[ "$(id -u)" = 0 ] || { echo "нужен root: sudo bash install.sh" >&2; exit 1; } + +command -v python3 >/dev/null || { + echo "==> ставлю python3" + DEBIAN_FRONTEND=noninteractive apt-get update -qq + DEBIAN_FRONTEND=noninteractive apt-get install -y -qq python3 +} +PYV=$(python3 -c 'import sys;print("%d.%d"%sys.version_info[:2])') +python3 - <<'PY' || { echo "нужен Python 3.10+, найден $PYV" >&2; exit 1; } +import sys; sys.exit(0 if sys.version_info >= (3, 10) else 1) +PY +echo "==> python3 $PYV" + +echo "==> служебный пользователь $USER" +id "$USER" >/dev/null 2>&1 || useradd --system --no-create-home --shell /usr/sbin/nologin "$USER" + +echo "==> код в $DEST" +install -d -m 755 "$DEST" +rm -rf "$DEST/dmrsrv" +cp -r "$SRC/dmrsrv" "$DEST/dmrsrv" +[ -d "$SRC/tests" ] && { rm -rf "$DEST/tests"; cp -r "$SRC/tests" "$DEST/tests"; } +find "$DEST" -name '__pycache__' -type d -exec rm -rf {} + 2>/dev/null || true +chown -R root:root "$DEST" + +echo "==> конфигурация в $CFGDIR" +install -d -m 750 -o "$USER" -g "$USER" "$CFGDIR" +if [ -f "$CFGDIR/config.json" ]; then + echo " config.json уже есть — не трогаю" +else + # Пароль панели сервер сгенерирует сам при первом запуске и напишет в журнал. + cat > "$CFGDIR/config.json" <<'JSON' +{ + "bind": "0.0.0.0", + "port": 62031, + "passphrase": "passw0rd", + "panel": { "bind": "0.0.0.0", "port": 8080, "user": "admin", "password": "" }, + "groups": [ + { "tg": 9, "name": "Локальная", "slot": 2 }, + { "tg": 1, "name": "Общая", "slot": 1 } + ], + "parrot": { "enabled": true, "tg": 9990, "delay": 2.0 }, + "peers_allowlist": [], + "log_keep": 500 +} +JSON + chown "$USER:$USER" "$CFGDIR/config.json" + chmod 600 "$CFGDIR/config.json" + echo " создан конфиг по умолчанию (ОБЯЗАТЕЛЬНО смени passphrase в панели)" +fi + +echo "==> systemd" +cat > /etc/systemd/system/$APP.service < служба работает" +else + echo "==> СЛУЖБА НЕ ПОДНЯЛАСЬ:"; journalctl -u $APP -n 20 --no-pager; exit 1 +fi + +PORT=$(python3 -c "import json;print(json.load(open('$CFGDIR/config.json'))['port'])") +PPORT=$(python3 -c "import json;print(json.load(open('$CFGDIR/config.json'))['panel']['port'])") +PPASS=$(python3 -c "import json;print(json.load(open('$CFGDIR/config.json'))['panel']['password'])" 2>/dev/null || true) +IP=$(hostname -I 2>/dev/null | awk '{print $1}') + +cat <} + └──────────────────────────────────────────────────────────── + + Журнал: journalctl -u $APP -f + Перезапуск: systemctl restart $APP + + В Pi-Star: Configuration -> DMR Master -> DMRGateway/"DMR Master" + указать этот адрес и порт $PORT, пароль — passphrase из конфига. + +INFO diff --git a/tests/test_e2e.py b/tests/test_e2e.py new file mode 100644 index 0000000..9bfb189 --- /dev/null +++ b/tests/test_e2e.py @@ -0,0 +1,255 @@ +"""Сквозной тест: имитируем хотспот Pi-Star (MMDVMHost) против нашего мастера. + +Гоняем настоящие UDP-датаграммы через настоящий asyncio-мастер — так проверяется +вся цепочка сразу: разбор кадров, sha256-рукопожатие, разбор RPTC, keepalive, +маршрутизация группового вызова между двумя хотспотами и переписывание peer_id. + +Кадры собираются ровно так, как их шлёт MMDVMHost (DMRDirectNetwork.cpp): + RPTL 8б · RPTK 40б · RPTC 302б · RPTPING 11б · DMRD 55б. + +Запуск: python3 tests/test_e2e.py +""" +from __future__ import annotations + +import asyncio +import hashlib +import os +import socket +import sys +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +os.environ["DMRSRV_CONFIG"] = "/tmp/dmrsrv-test-config.json" + +from dmrsrv.router import Router, retarget # noqa: E402 +from dmrsrv.server import HomebrewMaster, PeerState, run_server # noqa: E402 + +PASS = b"testpass" +PEER_A, PEER_B = 2501001, 2501002 +TG = 9 +ok_n = fail_n = 0 + + +def ok(msg: str) -> None: + global ok_n + ok_n += 1 + print(f" \033[32mOK\033[0m {msg}") + + +def bad(msg: str) -> None: + global fail_n + fail_n += 1 + print(f" \033[31mFAIL\033[0m {msg}") + + +def rptc_config(peer_id: int, callsign: str) -> bytes: + """302 байта ровно как MMDVMHost: RPTC + id(4) + 294 байта ASCII фикс-ширины.""" + body = ( + f"{callsign:<8.8s}" # 8 позывной + f"{438000000:09d}" # 9 RX Гц + f"{430000000:09d}" # 9 TX Гц + f"{1:02d}" # 2 мощность + f"{1:02d}" # 2 colorcode + f"{'51.66600':8.8s}" # 8 широта + f"{'-1.170000':9.9s}" # 9 долгота + f"{10:03d}" # 3 высота + f"{'Test Location':<20.20s}" + f"{'Test Repeater':<19.19s}" + f"4" # 1 слоты: 4 = симплекс/хотспот + f"{'https://example.org':<124.124s}" + f"{'20250101_MMDVM':<40.40s}" + f"{'MMDVM_MMDVM_HS_Hat':<40.40s}" + ) + assert len(body) == 294, f"конфиг {len(body)} байт вместо 294" + return b"RPTC" + peer_id.to_bytes(4, "big") + body.encode("ascii") + + +def dmrd(peer_id: int, src: int, dst: int, *, slot: int = 2, group: bool = True, + seq: int = 0, stream: bytes = b"\x11\x22\x33\x44", + frame_type: int = 0b10, dtype: int = 0x01) -> bytes: + """55 байт: 53 ядро + BER + RSSI (как MMDVMHost).""" + bits = (0x80 if slot == 2 else 0) | (0 if group else 0x40) | (frame_type << 4) | dtype + return (b"DMRD" + bytes([seq]) + src.to_bytes(3, "big") + dst.to_bytes(3, "big") + + peer_id.to_bytes(4, "big") + bytes([bits]) + stream + + bytes(range(33)) + b"\x00\x00") + + +class Hotspot: + """Минимальный клиент Homebrew — ровно то, что делает MMDVMHost.""" + + def __init__(self, peer_id: int, callsign: str, port: int): + self.peer_id, self.callsign, self.port = peer_id, callsign, port + self.s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + self.s.bind(("127.0.0.1", 0)) + self.s.settimeout(2.0) + + def send(self, data: bytes) -> None: + self.s.sendto(data, ("127.0.0.1", self.port)) + + def recv(self) -> bytes: + return self.s.recv(2048) + + def drain(self) -> int: + """Выбросить всё, что уже пришло. Без этого следующая проверка вычитает + кадр от предыдущей и делает ложный вывод.""" + self.s.settimeout(0.05) + n = 0 + try: + while True: + self.s.recv(2048) + n += 1 + except (OSError, socket.timeout): + pass + return n + + def login(self) -> tuple[bool, str]: + self.send(b"RPTL" + self.peer_id.to_bytes(4, "big")) + r = self.recv() + if r[:6] != b"RPTACK": + return False, f"на RPTL пришло {r[:6]!r}, ждали RPTACK" + salt = r[6:10] + + # sha256(СЫРЫЕ 4 байта соли + пароль) — не hex-строка (см. docs/PROTOCOL.md) + self.send(b"RPTK" + self.peer_id.to_bytes(4, "big") + + hashlib.sha256(salt + PASS).digest()) + r = self.recv() + if r[:6] != b"RPTACK": + return False, f"на RPTK пришло {r[:6]!r} — пароль не принят" + if r[6:10] != self.peer_id.to_bytes(4, "big"): + return False, "в RPTACK после RPTK не наш peer_id" + + self.send(rptc_config(self.peer_id, self.callsign)) + r = self.recv() + if r[:6] != b"RPTACK": + return False, f"на RPTC пришло {r[:6]!r}" + return True, "ok" + + +async def main() -> int: + port = 62000 + (os.getpid() % 900) + cfg = {"groups": [{"tg": TG, "name": "Тест", "slot": 2}], + "parrot": {"enabled": False, "tg": 9990}} + router = Router(cfg, state=None) + master = HomebrewMaster( + password_for=lambda pid: PASS, + router=router, + encode_for=lambda tgt, d: retarget(d.raw, tgt.peer_id), + ) + server = asyncio.create_task(run_server(master, "127.0.0.1", port)) + await asyncio.sleep(0.3) + + a = Hotspot(PEER_A, "TEST1", port) + b = Hotspot(PEER_B, "TEST2", port) + + print("==> рукопожатие (RPTL → RPTK → RPTC)") + for hs in (a, b): + good, why = await asyncio.get_running_loop().run_in_executor(None, hs.login) + ok(f"хотспот {hs.peer_id} вошёл") if good else bad(f"{hs.peer_id}: {why}") + + p = master.peers.get(PEER_A) + if p and p.state == PeerState.CONNECTED: + ok("мастер считает хотспот подключённым") + else: + bad(f"состояние хотспота: {p.state if p else 'нет в реестре'}") + if p and (p.config or {}).get("callsign") == "TEST1": + ok("позывной из RPTC разобран верно") + else: + bad(f"позывной разобран как {(p.config or {}).get('callsign') if p else '—'!r}") + if p and (p.config or {}).get("package_id") == "MMDVM_MMDVM_HS_Hat": + ok("модель хотспота из RPTC разобрана (смещение 262)") + else: + bad(f"package_id: {(p.config or {}).get('package_id') if p else '—'!r}") + + print("==> неверный пароль отвергается") + # ВАЖНО: recv() блокирующий — звать его напрямую из корутины нельзя, он застопорит + # event loop, и мастер не успеет ответить. Только через run_in_executor. + c = Hotspot(2501003, "BAD", port) + c.send(b"RPTL" + (2501003).to_bytes(4, "big")) + await asyncio.get_running_loop().run_in_executor(None, c.recv) + c.send(b"RPTK" + (2501003).to_bytes(4, "big") + hashlib.sha256(b"wrong").digest()) + r = await asyncio.get_running_loop().run_in_executor(None, c.recv) + ok("неверный пароль → MSTNAK") if r[:6] == b"MSTNAK" else bad(f"пришло {r[:6]!r}, ждали MSTNAK") + + print("==> keepalive") + a.send(b"RPTPING" + PEER_A.to_bytes(4, "big")) + r = await asyncio.get_running_loop().run_in_executor(None, a.recv) + if r[:7] == b"MSTPONG" and r[7:11] == PEER_A.to_bytes(4, "big"): + ok("RPTPING → MSTPONG с нашим peer_id") + else: + bad(f"на RPTPING пришло {r[:7]!r}") + + print("==> групповой вызов A → B") + b.s.settimeout(2.0) + a.send(dmrd(PEER_A, src=2501101, dst=TG)) + try: + got = await asyncio.get_running_loop().run_in_executor(None, b.recv) + except socket.timeout: + got = b"" + if got[:4] == b"DMRD": + ok("кадр доставлен второму хотспоту") + if len(got) == 55: + ok("длина кадра 55 байт (DMRGateway отбрасывает 53)") + else: + bad(f"длина кадра {len(got)}, а не 55") + if int.from_bytes(got[11:15], "big") == PEER_B: + ok("peer_id переписан на получателя (байты 11-14)") + else: + bad(f"peer_id остался {int.from_bytes(got[11:15], 'big')}, ждали {PEER_B}") + if got[16:20] == b"\x11\x22\x33\x44": + ok("stream_id скопирован дословно") + else: + bad(f"stream_id изменён: {got[16:20].hex()}") + if got[20:53] == bytes(range(33)): + ok("бёрст не тронут") + else: + bad("бёрст изменён при пересылке") + else: + bad("второй хотспот кадр не получил") + + print("==> отправителю кадр не возвращается") + a.drain() + a.s.settimeout(0.6) + a.send(dmrd(PEER_A, src=2501101, dst=TG, seq=1, stream=b"\xaa\xbb\xcc\xdd")) + try: + echo = await asyncio.get_running_loop().run_in_executor(None, a.recv) + except socket.timeout: + echo = b"" + ok("эха отправителю нет (иначе петля в эфире)") if echo[:4] != b"DMRD" \ + else bad("кадр вернулся отправителю — будет петля") + + print("==> вызов на неразрешённую группу") + b.drain() + b.s.settimeout(0.6) + a.send(dmrd(PEER_A, src=2501101, dst=777, seq=2, stream=b"\x01\x02\x03\x04")) + try: + leak = await asyncio.get_running_loop().run_in_executor(None, b.recv) + except socket.timeout: + leak = b"" + if leak[:4] != b"DMRD": + ok("TG777 нет в списке → не маршрутизируется") + else: + bad(f"кадр на неразрешённую группу ушёл (dst={int.from_bytes(leak[8:11], 'big')})") + + print("==> приватный вызов идёт туда, где слышали абонента") + b.drain() + router.set_radio_loc(2501202, PEER_B) + b.s.settimeout(2.0) + a.send(dmrd(PEER_A, src=2501101, dst=2501202, group=False, seq=3, + stream=b"\x09\x08\x07\x06")) + try: + priv = await asyncio.get_running_loop().run_in_executor(None, b.recv) + except socket.timeout: + priv = b"" + ok("приват доставлен на нужный хотспот") if priv[:4] == b"DMRD" \ + else bad("приватный вызов не доставлен") + + server.cancel() + await asyncio.gather(server, return_exceptions=True) + print(f"\n \033[1m{ok_n} OK, {fail_n} FAIL\033[0m\n") + return 0 if fail_n == 0 else 1 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main()))