dmr-server/dmrsrv/access.py

"""Доступ: пароли хотспотов и списки абонентов, которым разрешён эфир.

Два независимых уровня — их легко перепутать, поэтому разделены явно:

  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}")