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
100 строки
4.5 KiB
Python
100 строки
4.5 KiB
Python
"""Конфигурация сервера: JSON-файл, правится и руками, и из панели.
|
|
|
|
Намеренно без внешних зависимостей — на Debian 13 (PEP 668) любой pip install
|
|
в систему запрещён, а заводить venv ради пары библиотек не хочется.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import json
|
|
import os
|
|
import secrets
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
CONFIG_PATH = Path(os.environ.get("DMRSRV_CONFIG", "/etc/dmr-server/config.json"))
|
|
|
|
PARROT_TG = 9990 # общепринятый номер «попугая» в DMR-сетях
|
|
|
|
DEFAULTS: dict[str, Any] = {
|
|
"bind": "0.0.0.0",
|
|
"port": 62031, # стандартный порт Homebrew/MMDVM
|
|
"passphrase": "passw0rd", # пароль хотспотов (RPTK), меняется в панели
|
|
"panel": {
|
|
"bind": "0.0.0.0",
|
|
"port": 8080,
|
|
"user": "admin",
|
|
"password": "", # пустой -> сгенерируется при первом запуске
|
|
},
|
|
"groups": [
|
|
{"tg": 9, "name": "Локальная", "slot": 2},
|
|
{"tg": 1, "name": "Общая", "slot": 1},
|
|
],
|
|
"parrot": {"enabled": True, "tg": PARROT_TG, "delay": 2.0},
|
|
# Зарегистрированные хотспоты: у каждого может быть свой пароль.
|
|
# [{"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, # сколько последних событий держать для панели
|
|
}
|
|
|
|
|
|
def _merge(base: dict, over: dict) -> dict:
|
|
"""Рекурсивно накладывает сохранённое поверх умолчаний.
|
|
|
|
Нужно, чтобы добавление нового ключа в DEFAULTS не ломало существующий
|
|
конфиг и не требовало миграций.
|
|
"""
|
|
out = dict(base)
|
|
for k, v in over.items():
|
|
if isinstance(v, dict) and isinstance(out.get(k), dict):
|
|
out[k] = _merge(out[k], v)
|
|
else:
|
|
out[k] = v
|
|
return out
|
|
|
|
|
|
def load() -> dict[str, Any]:
|
|
cfg = dict(DEFAULTS)
|
|
if CONFIG_PATH.exists():
|
|
try:
|
|
cfg = _merge(DEFAULTS, json.loads(CONFIG_PATH.read_text("utf-8")))
|
|
except (json.JSONDecodeError, OSError) as e:
|
|
# Битый конфиг не должен ронять сервер: работаем на умолчаниях,
|
|
# но громко об этом сообщаем.
|
|
print(f"[config] не читается {CONFIG_PATH}: {e}; беру умолчания")
|
|
if not cfg["panel"]["password"]:
|
|
cfg["panel"]["password"] = secrets.token_urlsafe(12)
|
|
save(cfg)
|
|
print(f"[config] пароль панели сгенерирован: {cfg['panel']['password']}")
|
|
return cfg
|
|
|
|
|
|
def save(cfg: dict[str, Any]) -> None:
|
|
"""Атомарная запись: временный файл рядом + rename.
|
|
|
|
Без этого обрыв питания посреди сохранения из панели оставит битый JSON.
|
|
"""
|
|
CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, tmp = tempfile.mkstemp(dir=CONFIG_PATH.parent, prefix=".config.", suffix=".tmp")
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
|
json.dump(cfg, f, ensure_ascii=False, indent=2)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(tmp, CONFIG_PATH)
|
|
os.chmod(CONFIG_PATH, 0o600) # внутри пароли
|
|
except BaseException:
|
|
with contextlib.suppress(OSError):
|
|
os.unlink(tmp)
|
|
raise
|
© 2026 StackWorks