viktorиClaude Opus 4.8
cb4e8a9bb1
Панель: • шесть страниц с маршрутизацией по хэшу: Обзор, Last Heard, Хотспоты, Группы, Доступ, Настройки; • журнал приходит живым потоком (SSE) вместо опроса раз в две секунды; • Last Heard: кто, куда, когда, сколько, через какой хотспот; • в таблице хотспотов — ВСЁ, что устройство сообщает в RPTC. Контроль доступа (два независимых уровня, их легко перепутать): • хотспот пускается по паролю — у каждого может быть свой, плюс режим «только из списка» и отключение конкретного устройства; • абонент (поле src, владелец рации) допускается белым/чёрным списком с поддержкой диапазонов. Встроено в штатный хук guard ядра. Два бага по дороге: • поля RPTC читались как colorcode/latitude/longitude, а разборщик отдаёт color_code/lat/lon — половина таблицы молча пустовала; • столбец «описание» на деле содержит СТРАНУ: именно туда Pi-Star кладёт country. Переименован. tests/test_access.py 20/20, tests/test_panel.py 31/31 (включая наполнение Last Heard настоящим рукопожатием и голосовым вызовом по UDP). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XvX94Vwy7vXWEuiFm1g9Av
285 строки
15 KiB
Python
285 строки
15 KiB
Python
"""Веб-панель: хотспоты, группы, живой журнал, настройки.
|
|
|
|
Работает на stdlib `http.server` в отдельном потоке — так проще и надёжнее, чем
|
|
писать разбор HTTP поверх asyncio. Нагрузка тут никакая: несколько хотспотов и
|
|
одна-две открытые вкладки.
|
|
|
|
Авторизация — HTTP Basic. Для внутренней сети этого достаточно, а формы с
|
|
сессиями потребовали бы куки, CSRF и хранение сессий ради того же результата.
|
|
"""
|
|
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"]
|
|
|
|
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["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()
|
|
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/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:
|
|
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 "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)
|
|
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
|
© 2026 StackWorks