viktorиClaude Opus 4.8
8e3b9144c3
Ядро протокола форкнуто из 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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XvX94Vwy7vXWEuiFm1g9Av
526 строки
32 KiB
Python
526 строки
32 KiB
Python
"""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()
|
© 2026 StackWorks