diff --git a/dmrsrv/__main__.py b/dmrsrv/__main__.py index 74df845..fb07f7e 100644 --- a/dmrsrv/__main__.py +++ b/dmrsrv/__main__.py @@ -15,7 +15,9 @@ import time from . import config as cfgmod from . import log as logmod from . import panel as panelmod +from .access import Guard, make_password_for from .analyzer import TAG_MSTCL +from .lastheard import LastHeard from .parrot import Parrot from .router import Router, retarget from .server import HomebrewMaster, PeerState, run_server @@ -41,13 +43,14 @@ def _wire_state(master: HomebrewMaster, state: State) -> None: 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", "") + # Имена полей — ровно как у analyzer.parse_rptc_config. Раньше читались + # colorcode/latitude/longitude, которых там нет, и поля молча пустовали. + for fld in ("callsign", "location", "description", "url", + "software_id", "package_id", "rx_freq", "tx_freq", + "tx_power", "color_code", "lat", "lon", "height", "slots"): + val = cfgp.get(fld) + if val is not None: + setattr(sp, fld, val) # Мастер считает время по monotonic, панель — по time.time(). # Без пересчёта панель показывала «простой» равным времени в сети, # то есть живой хотспот выглядел мёртвым. @@ -67,6 +70,8 @@ async def _sync_loop(master: HomebrewMaster, interval: float = 1.0) -> None: await asyncio.sleep(interval) with contextlib.suppress(Exception): master._state_sync() + with contextlib.suppress(Exception): + master._lastheard_sweep() def main() -> int: @@ -76,31 +81,31 @@ def main() -> int: 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 + guard = Guard(cfg, state) master = HomebrewMaster( - password_for=password_for, + password_for=make_password_for(cfg, state), router=router, + guard=guard, # Подставляем номер ПОЛУЧАТЕЛЯ в байты 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) + lastheard = LastHeard(state) + master._lastheard_sweep = lastheard.sweep + parrot = Parrot(master) if cfg.get("parrot", {}).get("enabled", True) else None + if parrot is not None: state.event("info", f"попугай включён на TG{cfg['parrot'].get('tg', 9990)}") + def on_dmrd(peer, dmr): + lastheard.feed(peer, dmr) + if parrot is not None: + parrot.feed(dmr) + + master.on_dmrd = on_dmrd + def drop_peer(pid: int) -> bool: p = master.peers.pop(pid, None) if p is None: diff --git a/dmrsrv/access.py b/dmrsrv/access.py new file mode 100644 index 0000000..5e1c7cf --- /dev/null +++ b/dmrsrv/access.py @@ -0,0 +1,131 @@ +"""Доступ: пароли хотспотов и списки абонентов, которым разрешён эфир. + +Два независимых уровня — их легко перепутать, поэтому разделены явно: + + 1. ХОТСПОТ (оборудование) пускается по паролю на этапе RPTK. У каждого может быть + свой пароль; можно оставить общий для всех. + 2. АБОНЕНТ (владелец рации, поле src в кадре) может быть допущен или закрыт для + выхода в эфир. Проверяется на каждой передаче. + +Второе не следует из первого: на одном хотспоте работают разные люди, и запретить +конкретного абонента, не отключая всё устройство, — обычное требование. +""" +from __future__ import annotations + +from collections import Counter + +from .log import get_logger + +log = get_logger(__name__) + +# Диапазон DMR ID: 24 бита, ноль не абонент. +ID_MIN, ID_MAX = 1, 16777215 + + +def parse_ids(items) -> set[int]: + """Список ID из конфига: числа и диапазоны вида "2501000-2501999". + + Терпимо к мусору: одна кривая строка не должна ронять сервер и не должна + молча открывать доступ — просто пропускаем её с предупреждением. + """ + out: set[int] = set() + for raw in items or []: + s = str(raw).strip() + if not s: + continue + try: + if "-" in s: + a, b = s.split("-", 1) + lo, hi = int(a), int(b) + if lo > hi: + lo, hi = hi, lo + if hi - lo > 100000: # защита от «0-16777215» одной строкой + log.warning("список ID: диапазон %s слишком широкий, пропущен", s) + continue + out.update(range(lo, hi + 1)) + else: + out.add(int(s)) + except ValueError: + log.warning("список ID: не разобрал запись %r, пропущена", raw) + return out + + +def hotspot_entry(cfg: dict, peer_id: int) -> dict | None: + for h in cfg.get("hotspots", []): + try: + if int(h.get("peer_id")) == int(peer_id): + return h + except (TypeError, ValueError): + continue + return None + + +def make_password_for(cfg: dict, state): + """Колбэк для мастера: пароль хотспота или None (не пускать). + + None на этапе RPTL/RPTK = отказ, хотспот получит MSTNAK и будет пробовать снова. + """ + + def password_for(peer_id: int) -> bytes | None: + h = hotspot_entry(cfg, peer_id) + if h is not None: + if not h.get("enabled", True): + state.event("warn", f"хотспот {peer_id} ({h.get('name','')}) отключён в панели — отказ") + return None + pw = h.get("password") or cfg.get("passphrase", "") + return str(pw).encode("utf-8") + if cfg.get("allow_unknown_hotspots", True): + return str(cfg.get("passphrase", "")).encode("utf-8") + state.event("warn", f"хотспот {peer_id} не зарегистрирован — отказ " + f"(включён режим «только из списка»)") + return None + + return password_for + + +class Guard: + """Пускать ли абонента в эфир. Интерфейс задан ядром: check() -> (allow, reason).""" + + def __init__(self, cfg: dict, state) -> None: + self.cfg = cfg + self.state = state + self.dropped: Counter = Counter() + self._last_note: dict[int, int] = {} # src -> последний залогированный поток + + def _subs(self) -> dict: + return self.cfg.get("subscribers", {}) or {} + + def check(self, peer_id: int, dmr) -> tuple[bool, str]: + src = int(getattr(dmr, "src", 0) or 0) + + if not (ID_MIN <= src <= ID_MAX): + self.dropped["некорректный ID"] += 1 + return False, "некорректный ID абонента" + + # Абонент не должен выдавать себя за сам хотспот: обычно это петля или подмена. + if src == int(peer_id): + self.dropped["src совпал с хотспотом"] += 1 + return False, "src совпал с номером хотспота" + + mode = str(self._subs().get("mode", "off")).lower() + if mode not in ("allow", "deny"): + return True, "" + + ids = parse_ids(self._subs().get("list")) + if mode == "allow" and src not in ids: + self.dropped["нет в белом списке"] += 1 + self._note(src, dmr, "нет в белом списке") + return False, f"абонент {src} не в белом списке" + if mode == "deny" and src in ids: + self.dropped["в чёрном списке"] += 1 + self._note(src, dmr, "в чёрном списке") + return False, f"абонент {src} в чёрном списке" + return True, "" + + def _note(self, src: int, dmr, why: str) -> None: + """В журнал панели — раз на передачу, а не на каждый из ~17 кадров в секунду.""" + sid = getattr(dmr, "stream_id", 0) + if self._last_note.get(src) == sid: + return + self._last_note[src] = sid + self.state.event("warn", f"эфир запрещён: абонент {src} — {why}") diff --git a/dmrsrv/config.py b/dmrsrv/config.py index ea53d36..846ccc1 100644 --- a/dmrsrv/config.py +++ b/dmrsrv/config.py @@ -32,9 +32,18 @@ DEFAULTS: dict[str, Any] = { {"tg": 1, "name": "Общая", "slot": 1}, ], "parrot": {"enabled": True, "tg": PARROT_TG, "delay": 2.0}, - # Хотспоты, которым разрешено подключаться. Пустой список = пускать любой, - # кто знает пароль (удобно на старте, в панели можно перевести в белый список). - "peers_allowlist": [], + # Зарегистрированные хотспоты: у каждого может быть свой пароль. + # [{"peer_id": 2501001, "password": "...", "name": "Гараж", "enabled": true}] + # Пустой password = использовать общий passphrase. + "hotspots": [], + # true — пускать любой хотспот, знающий общий passphrase (удобно на старте); + # false — только те, что перечислены в hotspots. + "allow_unknown_hotspots": True, + # Кому разрешён ВЫХОД В ЭФИР (поле src в кадре). Это не то же, что доступ + # хотспота: на одном устройстве работают разные люди. + # mode: off | allow (белый список) | deny (чёрный список) + # list: числа и диапазоны "2501000-2501999" + "subscribers": {"mode": "off", "list": []}, "log_keep": 500, # сколько последних событий держать для панели } diff --git a/dmrsrv/lastheard.py b/dmrsrv/lastheard.py new file mode 100644 index 0000000..50e7e7e --- /dev/null +++ b/dmrsrv/lastheard.py @@ -0,0 +1,80 @@ +"""Last Heard — кто и когда выходил в эфир. + +Строится по потоку кадров, а не по хуку вызовов ядра: тот не отдаёт номер +хотспота, а без него не ответить на вопрос «через какое устройство слышали». + +Начало передачи определяется по смене stream_id (заголовки теряются в эфире), +конец — по терминатору ИЛИ по тишине: терминаторы теряются так же часто. +""" +from __future__ import annotations + +import time + +from .analyzer import CALL_GROUP +from .state import Call + +# 360 мс = 6 пропущенных голосовых кадров. Столько же ждёт HBlink, прежде чем +# считать передачу законченной. Берём с запасом, чтобы не рвать вызов на джиттере. +SILENCE_TIMEOUT = 2.0 + + +class LastHeard: + def __init__(self, state) -> None: + self.state = state + self._open: dict[tuple[int, int], Call] = {} # (peer_id, slot) -> вызов + self._sid: dict[tuple[int, int], bytes] = {} # (peer_id, slot) -> stream_id + self._seen: dict[tuple[int, int], float] = {} # (peer_id, slot) -> когда был кадр + + def feed(self, peer, dmr) -> None: + key = (int(dmr.peer_id), int(dmr.slot)) + sid = dmr.stream_id + now = time.time() + + if self._sid.get(key) != sid: # новая передача + self._close(key, "новый поток") + call = Call( + stream_id=sid if isinstance(sid, int) else int.from_bytes(bytes(sid or b"\0"), "big"), + src_id=int(dmr.src or 0), + dst_id=int(dmr.dst or 0), + slot=int(dmr.slot), + group=(dmr.call_type == CALL_GROUP), + peer_id=int(dmr.peer_id), + ) + self._open[key] = call + self._sid[key] = sid + with self.state._lock: + self.state.calls.append(call) + + call = self._open.get(key) + if call is not None: + call.frames += 1 + self._seen[key] = now + + if getattr(dmr, "is_terminator", False): + self._close(key, "терминатор") + + def sweep(self) -> None: + """Закрыть передачи, по которым перестали приходить кадры. + + Без этого зависший вызов вечно висел бы «активным»: терминатор теряется + в эфире регулярно, полагаться только на него нельзя. + """ + now = time.time() + for key, last in list(self._seen.items()): + if now - last > SILENCE_TIMEOUT: + self._close(key, "тишина") + + def _close(self, key, why: str) -> None: + call = self._open.pop(key, None) + self._sid.pop(key, None) + self._seen.pop(key, None) + if call is None or call.ended: + return + call.ended = time.time() + if call.frames > 1: # одиночный кадр — не передача, а шум + kind = "группа TG %s" % call.dst_id if call.group else "приват %s" % call.dst_id + self.state.event( + "call", + f"эфир: {call.src_id} → {kind} · TS{call.slot} · " + f"{round(call.ended - call.started, 1)} с · хотспот {call.peer_id}", + ) diff --git a/dmrsrv/page.py b/dmrsrv/page.py new file mode 100644 index 0000000..7af667f --- /dev/null +++ b/dmrsrv/page.py @@ -0,0 +1,321 @@ +"""HTML панели управления — вынесено из panel.py, чтобы серверная логика читалась. + +Одностраничное приложение с маршрутизацией по хэшу (#/overview и т.д.): страницы +разные, но файл один и без внешних зависимостей — на сервере нет ни pip, ни CDN. +Журнал приходит живым потоком (SSE), остальное обновляется опросом раз в 3 секунды. +""" + +PAGE = r""" + +DMR-сервер + + +
+ DMR-сервер + + +
+ +
+
+
+
—
хотспотов на связи
+
—
идёт передач
+
—
вызовов за сеанс
+
—
время работы
+
+

Идёт сейчас

+
+

Журнал подключаюсь…

+
+
+
+ +
+

Last Heard последние передачи в эфире

+
+
+ +
+

На связи всё, что хотспот сообщает о себе в RPTC

+
+

Зарегистрированные

+
+
+
+ + + + +
+

+
+
+
+ +
+

Разрешённые группы вызов на группу вне списка никуда не пойдёт

+
+
+ + + + +
+
+
+ +
+

Кому разрешён эфир проверяется src — владелец рации, не хотспот

+
+
+ + + + +
+ +
+

На одном хотспоте работают разные люди: запретить конкретного абонента, + не отключая всё устройство, — обычное требование. Отказы видны в журнале.

+
+
+

Незнакомые хотспоты

+
+
+ + + +
+
+
+
+ +
+

Параметры

+
+
+ + + +
+

+
+
+

Как подключить хотспот

+

+
+
+
+ + +""" diff --git a/dmrsrv/panel.py b/dmrsrv/panel.py index 4fb304d..1840c68 100644 --- a/dmrsrv/panel.py +++ b/dmrsrv/panel.py @@ -12,16 +12,26 @@ from __future__ import annotations import hmac import json import threading +import queue as _queue from base64 import b64decode from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse from . import config as cfgmod +from .page import PAGE from .log import get_logger log = get_logger(__name__) +def _valid_id_token(s: str) -> bool: + """Число или диапазон «нач-кон». Проверяем в панели, чтобы человек сразу + увидел опечатку, а не гадал, почему абонента не пускает.""" + s = s.strip() + parts = s.split("-", 1) if "-" in s else [s] + return all(p.strip().isdigit() and 1 <= int(p) <= 16777215 for p in parts) and len(parts) <= 2 + + def _make_handler(ctx): cfg, state = ctx["cfg"], ctx["state"] drop_peer = ctx["drop_peer"] @@ -79,10 +89,52 @@ def _make_handler(ctx): snap = state.snapshot() snap["groups"] = cfg.get("groups", []) snap["parrot"] = cfg.get("parrot", {}) + snap["hotspots"] = cfg.get("hotspots", []) + snap["subscribers"] = cfg.get("subscribers", {"mode": "off", "list": []}) + snap["allow_unknown"] = cfg.get("allow_unknown_hotspots", True) snap["port"] = cfg.get("port") return self._json(snap) + if path == "/api/events": + return self._sse() self.send_error(404) + def _sse(self) -> None: + """Живой поток событий (Server-Sent Events). + + Заменяет опрос раз в две секунды: события уходят в браузер сразу, + как только случились. Каждое соединение обслуживается своим потоком + (ThreadingHTTPServer), очередь ограничена — зависший браузер не + затормозит приём эфира. + """ + q = state.subscribe() + try: + self.send_response(200) + self.send_header("Content-Type", "text/event-stream; charset=utf-8") + self.send_header("Cache-Control", "no-cache") + self.send_header("Connection", "keep-alive") + self.send_header("X-Accel-Buffering", "no") # чтобы не буферизовал прокси + self.end_headers() + # Сразу отдаём хвост журнала, иначе после перезагрузки страницы пусто. + for rec in state.snapshot()["log"][-100:]: + self._sse_send(rec) + while True: + try: + rec = q.get(timeout=15) + except _queue.Empty: + self.wfile.write(b": keepalive\n\n") # держим соединение живым + self.wfile.flush() + continue + self._sse_send(rec) + except (BrokenPipeError, ConnectionResetError, OSError): + pass # вкладку закрыли — это норма + finally: + state.unsubscribe(q) + + def _sse_send(self, rec) -> None: + self.wfile.write(b"data: " + json.dumps(rec, ensure_ascii=False).encode("utf-8") + + b"\n\n") + self.wfile.flush() + def do_POST(self) -> None: if not self._authed(): return self._deny() @@ -121,6 +173,71 @@ def _make_handler(ctx): state.event("info", f"панель: удалена группа TG{tg} (слот {slot})") return self._json({"ok": True, "groups": cfg["groups"]}) + if path == "/api/hotspots/add": + try: + pid = int(data["peer_id"]) + except (KeyError, ValueError, TypeError): + return self._json({"error": "нужен peer_id (номер хотспота)"}, 400) + if not (1 <= pid <= 0xFFFFFFFF): + return self._json({"error": "peer_id вне допустимого диапазона"}, 400) + if any(int(h["peer_id"]) == pid for h in cfg["hotspots"]): + return self._json({"error": f"хотспот {pid} уже в списке"}, 400) + pw = str(data.get("password", "")) + if pw and len(pw) < 4: + return self._json({"error": "пароль слишком короткий"}, 400) + cfg["hotspots"].append({"peer_id": pid, "password": pw, + "name": str(data.get("name", ""))[:40], "enabled": True}) + cfgmod.save(cfg) + state.event("info", f"панель: добавлен хотспот {pid}" + + (" (свой пароль)" if pw else " (общий пароль)")) + return self._json({"ok": True, "hotspots": cfg["hotspots"]}) + + if path == "/api/hotspots/del": + try: + pid = int(data["peer_id"]) + except (KeyError, ValueError, TypeError): + return self._json({"error": "нужен peer_id"}, 400) + before = len(cfg["hotspots"]) + cfg["hotspots"] = [h for h in cfg["hotspots"] if int(h["peer_id"]) != pid] + if len(cfg["hotspots"]) == before: + return self._json({"error": "такого хотспота нет в списке"}, 404) + cfgmod.save(cfg) + state.event("info", f"панель: хотспот {pid} удалён из списка") + return self._json({"ok": True, "hotspots": cfg["hotspots"]}) + + if path == "/api/hotspots/toggle": + try: + pid = int(data["peer_id"]) + except (KeyError, ValueError, TypeError): + return self._json({"error": "нужен peer_id"}, 400) + for h in cfg["hotspots"]: + if int(h["peer_id"]) == pid: + h["enabled"] = not h.get("enabled", True) + cfgmod.save(cfg) + st = "включён" if h["enabled"] else "отключён" + state.event("warn", f"панель: хотспот {pid} {st}") + # отключили — рвём текущую сессию, иначе он останется в эфире + if not h["enabled"]: + drop_peer(pid) + return self._json({"ok": True, "enabled": h["enabled"]}) + return self._json({"error": "такого хотспота нет"}, 404) + + if path == "/api/subscribers": + mode = str(data.get("mode", "off")).lower() + if mode not in ("off", "allow", "deny"): + return self._json({"error": "режим: off, allow или deny"}, 400) + raw = data.get("list", []) + if isinstance(raw, str): + raw = [x for x in raw.replace(",", " ").split() if x] + bad = [x for x in raw if not _valid_id_token(str(x))] + if bad: + return self._json({"error": "непонятные записи: " + ", ".join(map(str, bad[:5]))}, 400) + cfg["subscribers"] = {"mode": mode, "list": [str(x) for x in raw]} + cfgmod.save(cfg) + names = {"off": "выключены", "allow": "белый список", "deny": "чёрный список"} + state.event("warn", f"панель: списки абонентов — {names[mode]}, записей: {len(raw)}") + return self._json({"ok": True, "subscribers": cfg["subscribers"]}) + if path == "/api/settings": changed = [] if "passphrase" in data: @@ -132,6 +249,10 @@ def _make_handler(ctx): if "parrot_enabled" in data: cfg["parrot"]["enabled"] = bool(data["parrot_enabled"]) changed.append("попугай " + ("включён" if cfg["parrot"]["enabled"] else "выключен")) + if "allow_unknown_hotspots" in data: + cfg["allow_unknown_hotspots"] = bool(data["allow_unknown_hotspots"]) + changed.append("незнакомые хотспоты: " + + ("пускать" if cfg["allow_unknown_hotspots"] else "только из списка")) if not changed: return self._json({"error": "нечего менять"}, 400) cfgmod.save(cfg) @@ -161,95 +282,3 @@ def start(cfg: dict, state, drop_peer) -> ThreadingHTTPServer: 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/state.py b/dmrsrv/state.py index 3a4e573..c2b7c4f 100644 --- a/dmrsrv/state.py +++ b/dmrsrv/state.py @@ -6,6 +6,7 @@ """ from __future__ import annotations +import queue import threading import time from collections import deque @@ -19,14 +20,22 @@ class Peer: peer_id: int addr: tuple[str, int] + # Всё, что хотспот присылает в RPTC. Имена — как у разборщика (analyzer), + # чтобы не переименовывать по дороге и не ловить пустые поля. callsign: str = "" - rx_freq: str = "" - tx_freq: str = "" - colorcode: str = "" + rx_freq: int | None = None + tx_freq: int | None = None + tx_power: int | None = None + color_code: int | None = None + lat: float | None = None + lon: float | None = None + height: int | None = None location: str = "" - description: str = "" + description: str = "" # Pi-Star кладёт сюда СТРАНУ + slots: int | None = None url: str = "" - software: str = "" + software_id: str = "" + package_id: str = "" connected_at: float = field(default_factory=time.time) last_seen: float = field(default_factory=time.time) # стадия рукопожатия: login -> auth -> config -> ready @@ -42,11 +51,18 @@ class Peer: "addr": f"{self.addr[0]}:{self.addr[1]}", "callsign": self.callsign, "location": self.location, - "description": self.description, - "software": self.software, - "colorcode": self.colorcode, + "country": self.description, # так это поле заполняет Pi-Star "rx_freq": self.rx_freq, "tx_freq": self.tx_freq, + "tx_power": self.tx_power, + "color_code": self.color_code, + "lat": self.lat, + "lon": self.lon, + "height": self.height, + "slots": self.slots, + "url": self.url, + "software_id": self.software_id, + "package_id": self.package_id, "stage": self.stage, "online_for": int(now - self.connected_at), "idle": round(now - self.last_seen, 1), @@ -93,12 +109,34 @@ class State: self.calls: deque[Call] = deque(maxlen=50) self.active: dict[tuple[int, int], Call] = {} # (slot, stream_id) -> Call self.started = time.time() + # Подписчики SSE. Очередь ограничена: медленный или зависший браузер не + # должен тормозить приём эфира — при переполнении события просто теряются + # у этого одного клиента. + self._subs: list[queue.Queue] = [] + + def subscribe(self) -> queue.Queue: + q: queue.Queue = queue.Queue(maxsize=200) + with self._lock: + self._subs.append(q) + return q + + def unsubscribe(self, q: queue.Queue) -> None: + with self._lock: + if q in self._subs: + self._subs.remove(q) # ── журнал ────────────────────────────────────────────────────────────── def event(self, kind: str, text: str) -> None: """kind: info | call | peer | warn | error — панель красит по нему.""" + rec = {"t": time.time(), "kind": kind, "text": text} with self._lock: - self.log.append({"t": time.time(), "kind": kind, "text": text}) + self.log.append(rec) + subs = list(self._subs) + for q in subs: + try: + q.put_nowait(rec) + except queue.Full: + pass # этот клиент не успевает — его проблема, эфир важнее print(f"[{kind}] {text}", flush=True) # заодно в journalctl # ── хотспоты ──────────────────────────────────────────────────────────── diff --git a/tests/test_access.py b/tests/test_access.py new file mode 100644 index 0000000..4d915ad --- /dev/null +++ b/tests/test_access.py @@ -0,0 +1,111 @@ +"""Проверка доступа: пароли хотспотов и списки абонентов. + +Это защитная логика, поэтому проверяется не только «разрешает кого надо», но и +«НЕ разрешает кого не надо» — вторая половина важнее и обычно её забывают. + +Запуск: python3 tests/test_access.py +""" +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from dmrsrv.access import Guard, make_password_for, parse_ids # noqa: E402 + +ok_n = fail_n = 0 + + +def ok(m): + global ok_n + ok_n += 1 + print(f" \033[32mOK\033[0m {m}") + + +def bad(m): + global fail_n + fail_n += 1 + print(f" \033[31mFAIL\033[0m {m}") + + +def check(cond, good, bad_msg): + ok(good) if cond else bad(bad_msg) + + +class FakeState: + def __init__(self): + self.events = [] + + def event(self, kind, text): + self.events.append((kind, text)) + + +class Frame: + def __init__(self, src, dst=9, stream_id=1): + self.src, self.dst, self.stream_id = src, dst, stream_id + self.call_type = "group" + + +print("==> разбор списков ID") +check(parse_ids([1, "2", "5-7"]) == {1, 2, 5, 6, 7}, "числа и диапазоны разбираются", + f"разобрано неверно: {parse_ids([1, '2', '5-7'])}") +check(parse_ids(["7-5"]) == {5, 6, 7}, "перевёрнутый диапазон нормализуется", "7-5 не разобран") +check(parse_ids(["мусор", "", None]) == set(), "мусор пропускается без падения", "мусор не отфильтрован") +check(parse_ids(["0-16777215"]) == set(), "слишком широкий диапазон отвергнут (защита от «пустить всех»)", + "широкий диапазон принят — так можно случайно открыть всё") + +print("==> пароли хотспотов") +st = FakeState() +cfg = {"passphrase": "obshiy", "allow_unknown_hotspots": True, + "hotspots": [{"peer_id": 100, "password": "lichniy", "enabled": True}, + {"peer_id": 200, "password": "", "enabled": True}, + {"peer_id": 300, "password": "x", "enabled": False}]} +pf = make_password_for(cfg, st) +check(pf(100) == b"lichniy", "у хотспота свой пароль", f"получено {pf(100)!r}") +check(pf(200) == b"obshiy", "пустой пароль = общий", f"получено {pf(200)!r}") +check(pf(300) is None, "отключённый хотспот не пускается", "отключённый хотспот пустили!") +check(pf(999) == b"obshiy", "незнакомый пускается по общему паролю (режим по умолчанию)", + f"получено {pf(999)!r}") + +cfg["allow_unknown_hotspots"] = False +check(pf(999) is None, "режим «только из списка»: незнакомый не пускается", + "незарегистрированный хотспот пустили при закрытом режиме!") +check(pf(100) == b"lichniy", "в закрытом режиме свои продолжают работать", "свой хотспот перестал пускаться") + +print("==> списки абонентов") +st = FakeState() +cfg = {"subscribers": {"mode": "off", "list": []}} +g = Guard(cfg, st) +check(g.check(100, Frame(2501001))[0], "режим off — пускает всех", "off не пускает") + +cfg["subscribers"] = {"mode": "allow", "list": ["2501001", "2502000-2502010"]} +check(g.check(100, Frame(2501001))[0], "белый список: свой проходит", "свой не прошёл") +check(g.check(100, Frame(2502005))[0], "белый список: диапазон работает", "диапазон не сработал") +check(not g.check(100, Frame(2509999))[0], "белый список: чужой НЕ проходит", + "чужой прошёл через белый список!") + +cfg["subscribers"] = {"mode": "deny", "list": ["2501001"]} +check(not g.check(100, Frame(2501001))[0], "чёрный список: запрещённый НЕ проходит", + "запрещённый абонент прошёл!") +check(g.check(100, Frame(2501002))[0], "чёрный список: остальные проходят", "лишнего заблокировали") + +print("==> некорректные и подозрительные ID") +cfg["subscribers"] = {"mode": "off", "list": []} +check(not g.check(100, Frame(0))[0], "нулевой ID отвергается", "нулевой ID пропущен") +check(not g.check(100, Frame(99999999))[0], "ID за пределами 24 бит отвергается", "слишком большой ID пропущен") +check(not g.check(100, Frame(100))[0], "src, равный номеру хотспота, отвергается (петля/подмена)", + "src == peer_id пропущен") + +print("==> журнал для пользователя") +st2 = FakeState() +cfg2 = {"subscribers": {"mode": "deny", "list": ["777"]}} +g2 = Guard(cfg2, st2) +for _ in range(20): + g2.check(100, Frame(777, stream_id=42)) # одна передача = один кадр за другим +g2.check(100, Frame(777, stream_id=43)) # следующая передача +check(len(st2.events) == 2, "запись в журнал раз на передачу, а не на каждый кадр", + f"записей {len(st2.events)}, ожидалось 2 — журнал захлебнётся при 17 кадрах в секунду") + +print(f"\n \033[1m{ok_n} OK, {fail_n} FAIL\033[0m\n") +sys.exit(0 if fail_n == 0 else 1) diff --git a/tests/test_panel.py b/tests/test_panel.py new file mode 100644 index 0000000..c64054f --- /dev/null +++ b/tests/test_panel.py @@ -0,0 +1,196 @@ +"""Проверка панели: страницы, API, SSE и Last Heard. + +Поднимает настоящий сервер и говорит с ним по HTTP, как браузер. +Запуск: python3 tests/test_panel.py +""" +from __future__ import annotations + +import json +import os +import socket +import subprocess +import sys +import time +import urllib.error +import urllib.request +from base64 import b64encode +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +CFG = "/tmp/dmrsrv-panel-test.json" +PORT = 18080 +UDP = 18031 +ok_n = fail_n = 0 + + +def ok(m): + global ok_n + ok_n += 1 + print(f" \033[32mOK\033[0m {m}") + + +def bad(m): + global fail_n + fail_n += 1 + print(f" \033[31mFAIL\033[0m {m}") + + +def check(c, g, b): + ok(g) if c else bad(b) + + +Path(CFG).write_text(json.dumps({ + "bind": "127.0.0.1", "port": UDP, "passphrase": "testpass", + "panel": {"bind": "127.0.0.1", "port": PORT, "user": "admin", "password": "sekret"}, + "groups": [{"tg": 9, "name": "Локальная", "slot": 2}], + "parrot": {"enabled": False, "tg": 9990}, + "hotspots": [], "allow_unknown_hotspots": True, + "subscribers": {"mode": "off", "list": []}, +}, ensure_ascii=False)) + +env = dict(os.environ, DMRSRV_CONFIG=CFG) +srv = subprocess.Popen([sys.executable, "-m", "dmrsrv"], cwd=ROOT, env=env, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) +AUTH = "Basic " + b64encode(b"admin:sekret").decode() + + +def req(path, data=None, auth=True, timeout=6): + r = urllib.request.Request(f"http://127.0.0.1:{PORT}{path}") + if auth: + r.add_header("Authorization", AUTH) + if data is not None: + r.data = json.dumps(data).encode() + r.add_header("Content-Type", "application/json") + return urllib.request.urlopen(r, timeout=timeout) + + +try: + for _ in range(60): + try: + req("/api/state").read() + break + except Exception: + time.sleep(0.25) + else: + print("сервер не поднялся:", srv.stdout.read() if srv.stdout else "") + sys.exit(1) + + print("==> авторизация") + try: + req("/", auth=False) + bad("панель отдаётся без пароля!") + except urllib.error.HTTPError as e: + check(e.code == 401, "без пароля — 401", f"код {e.code}") + check(req("/").status == 200, "с паролем — 200", "с паролем не пускает") + + print("==> страницы") + html = req("/").read().decode() + for pid, name in [("p-overview", "Обзор"), ("p-lastheard", "Last Heard"), + ("p-hotspots", "Хотспоты"), ("p-groups", "Группы"), + ("p-access", "Доступ"), ("p-settings", "Настройки")]: + check(f'id={pid}' in html, f"страница «{name}» есть", f"нет страницы {pid}") + check("EventSource" in html, "журнал подключён живым потоком (SSE)", "SSE не подключён") + + print("==> управление хотспотами") + req("/api/hotspots/add", {"peer_id": 2501001, "name": "Гараж", "password": "hotpass"}) + st = json.load(req("/api/state")) + check(any(h["peer_id"] == 2501001 for h in st["hotspots"]), "хотспот добавлен", "не добавился") + try: + req("/api/hotspots/add", {"peer_id": 2501001}) + bad("дубль хотспота принят") + except urllib.error.HTTPError as e: + check(e.code == 400, "повторный номер отклонён", f"код {e.code}") + req("/api/hotspots/toggle", {"peer_id": 2501001}) + st = json.load(req("/api/state")) + check(st["hotspots"][0]["enabled"] is False, "хотспот отключается", "не отключился") + + print("==> списки абонентов") + req("/api/subscribers", {"mode": "deny", "list": "2501111 2502000-2502005"}) + st = json.load(req("/api/state")) + check(st["subscribers"]["mode"] == "deny", "режим сохранён", "режим не сохранён") + check(len(st["subscribers"]["list"]) == 2, "список сохранён", "список не сохранён") + try: + req("/api/subscribers", {"mode": "allow", "list": "не-число"}) + bad("мусор в списке принят — оператор не увидит опечатку") + except urllib.error.HTTPError as e: + check(e.code == 400, "мусор в списке отклонён с объяснением", f"код {e.code}") + try: + req("/api/subscribers", {"mode": "странно", "list": ""}) + bad("неизвестный режим принят") + except urllib.error.HTTPError as e: + check(e.code == 400, "неизвестный режим отклонён", f"код {e.code}") + + print("==> Last Heard наполняется из эфира") + # шлём настоящее рукопожатие и голосовой вызов + import hashlib + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.settimeout(3) + PEER = 2509999 + + def rt(p): + s.sendto(p, ("127.0.0.1", UDP)) + return s.recv(2048) + + salt = rt(b"RPTL" + PEER.to_bytes(4, "big"))[6:10] + rt(b"RPTK" + PEER.to_bytes(4, "big") + hashlib.sha256(salt + b"testpass").digest()) + body = (f"{'LHTEST':<8.8s}{438000000:09d}{430000000:09d}{5:02d}{1:02d}" + f"{'51.66600':8.8s}{'-1.170000':9.9s}{10:03d}{'Irkutsk':<20.20s}" + f"{'Russia':<19.19s}4{'https://x':<124.124s}{'MMDVM_v1':<40.40s}{'MMDVM_HS':<40.40s}") + rt(b"RPTC" + PEER.to_bytes(4, "big") + body.encode()) + + def frame(seq, dtype, ftype=0b10): + bits = 0x80 | (ftype << 4) | dtype + return (b"DMRD" + bytes([seq]) + (2501234).to_bytes(3, "big") + (9).to_bytes(3, "big") + + PEER.to_bytes(4, "big") + bytes([bits]) + b"\xaa\xbb\xcc\xdd" + + bytes(range(33)) + b"\x00\x00") + + s.sendto(frame(0, 0x01), ("127.0.0.1", UDP)) # начало + for i in range(1, 6): + s.sendto(frame(i, i % 6, 0b00), ("127.0.0.1", UDP)) + s.sendto(frame(6, 0x02), ("127.0.0.1", UDP)) # терминатор + time.sleep(1.5) + + st = json.load(req("/api/state")) + calls = st.get("calls", []) + check(len(calls) >= 1, "вызов попал в Last Heard", "Last Heard пуст после реального вызова") + if calls: + c = calls[0] + check(c["src_id"] == 2501234, "источник записан верно", f"src={c['src_id']}") + check(c["dst_id"] == 9, "назначение записано верно", f"dst={c['dst_id']}") + check(c["peer_id"] == PEER, "видно, через какой хотспот", f"peer={c['peer_id']}") + check(not c["active"], "вызов закрыт по терминатору", "вызов остался активным") + check(c["frames"] >= 6, "кадры посчитаны", f"кадров {c['frames']}") + + print("==> поля RPTC доходят до панели") + p = st["peers"][0] if st["peers"] else {} + check(p.get("country") == "Russia", "страна (Pi-Star кладёт её в description)", + f"country={p.get('country')!r}") + check(p.get("location") == "Irkutsk", "местоположение", f"location={p.get('location')!r}") + check(p.get("color_code") == 1, "color code (раньше читался как colorcode и был пуст)", + f"color_code={p.get('color_code')!r}") + check(p.get("tx_power") == 5, "мощность", f"tx_power={p.get('tx_power')!r}") + check(p.get("rx_freq") == 438000000, "частота RX", f"rx={p.get('rx_freq')!r}") + check(p.get("lat") == 51.666, "широта", f"lat={p.get('lat')!r}") + check(p.get("package_id") == "MMDVM_HS", "модель устройства", f"pkg={p.get('package_id')!r}") + + print("==> живой поток событий (SSE)") + r = urllib.request.Request(f"http://127.0.0.1:{PORT}/api/events") + r.add_header("Authorization", AUTH) + resp = urllib.request.urlopen(r, timeout=6) + check(resp.headers.get("Content-Type", "").startswith("text/event-stream"), + "заголовок text/event-stream", f"тип {resp.headers.get('Content-Type')}") + line = resp.readline().decode() + check(line.startswith("data: "), "события приходят в формате SSE", f"строка {line!r}") + resp.close() + +finally: + srv.terminate() + try: + srv.wait(timeout=5) + except subprocess.TimeoutExpired: + srv.kill() + +print(f"\n \033[1m{ok_n} OK, {fail_n} FAIL\033[0m\n") +sys.exit(0 if fail_n == 0 else 1)