"""Validated protocol used by the map WebApp and the bot. The current map bundle intentionally does not contain the compiled graph yet. Keeping parsing and the route adapter here means the real navigator can be connected later without trusting nodes or edges supplied by the WebApp. """ from __future__ import annotations import base64 import binascii import json import re from dataclasses import dataclass from typing import Any MAX_ROUTE_DATA = 4096 MAX_REQUEST_ID = 80 MAX_ENDPOINT_TEXT = 80 _REQUEST_ID = re.compile(r"^[A-Za-z0-9_.:-]{1,80}$") class NavigatorRequestError(ValueError): """The WebApp sent an unsupported or unsafe route request.""" @dataclass(frozen=True) class RouteEndpoint: room: str floor: str | None = None @dataclass(frozen=True) class RouteRequest: version: int request_id: str start: RouteEndpoint end: RouteEndpoint def _text(value: Any, field: str, limit: int = MAX_ENDPOINT_TEXT) -> str: if not isinstance(value, str): raise NavigatorRequestError(f"{field} должен быть строкой") value = value.strip() if not value or len(value) > limit or any(char in value for char in "<>\r\n"): raise NavigatorRequestError(f"Некорректное поле {field}") return value def _endpoint(value: Any, field: str) -> RouteEndpoint: if not isinstance(value, dict): raise NavigatorRequestError(f"{field} должен быть объектом") room = _text(value.get("room"), f"{field}.room") floor_value = value.get("floor") floor = None if floor_value is None else _text(floor_value, f"{field}.floor", 16) return RouteEndpoint(room=room, floor=floor) def parse_route_request(raw: str | bytes) -> RouteRequest: if isinstance(raw, bytes): if len(raw) > MAX_ROUTE_DATA: raise NavigatorRequestError("Слишком большой запрос маршрута") raw = raw.decode("utf-8", errors="strict") elif not isinstance(raw, str) or len(raw.encode("utf-8")) > MAX_ROUTE_DATA: raise NavigatorRequestError("Слишком большой запрос маршрута") try: payload = json.loads(raw) except (UnicodeError, json.JSONDecodeError) as error: raise NavigatorRequestError("Запрос маршрута не является JSON") from error if not isinstance(payload, dict) or payload.get("v") != 1 or payload.get("type") != "route.request": raise NavigatorRequestError("Неподдерживаемый формат запроса маршрута") request_id = _text(payload.get("request_id"), "request_id", MAX_REQUEST_ID) if not _REQUEST_ID.fullmatch(request_id): raise NavigatorRequestError("Некорректный request_id") start = _endpoint(payload.get("start"), "start") end = _endpoint(payload.get("end"), "end") if start.room.casefold() == end.room.casefold() and start.floor == end.floor: raise NavigatorRequestError("Начальная и конечная аудитории совпадают") return RouteRequest(version=1, request_id=request_id, start=start, end=end) def calculate_route(request: RouteRequest) -> dict[str, Any] | None: """Return a trusted route or ``None`` until compiled graph data is added. This is an explicit mock rather than a proximity-based fallback. A real implementation should load the server-side compiled ``neighbors`` / ``graph`` data, validate both room IDs against its manifest, and return a payload with ``segments`` containing node indexes or coordinates. """ del request return None def encode_route(route: dict[str, Any]) -> str: """Encode a route for a URL without putting credentials in the URL.""" data = json.dumps(route, ensure_ascii=False, separators=(",", ":")).encode("utf-8") return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=") def decode_route(value: str) -> dict[str, Any] | None: """Decode a route in tests/tools; the WebApp performs the same operation.""" try: padded = value + "=" * ((4 - len(value) % 4) % 4) result = json.loads(base64.urlsafe_b64decode(padded).decode("utf-8")) except (ValueError, TypeError, UnicodeError, binascii.Error, json.JSONDecodeError): return None return result if isinstance(result, dict) else None