"""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()