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
112 строки
5.7 KiB
Python
112 строки
5.7 KiB
Python
"""Проверка доступа: пароли хотспотов и списки абонентов.
|
|
|
|
Это защитная логика, поэтому проверяется не только «разрешает кого надо», но и
|
|
«НЕ разрешает кого не надо» — вторая половина важнее и обычно её забывают.
|
|
|
|
Запуск: 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)
|
© 2026 StackWorks