"""The owner's side of live tracking (AV-022): what the local development server uses behind the Mission Control panel. Two backends with one interface: * ``LocalLiveOwner`` -- the service in this process (data/local), its pages served by the development server itself: same code and rules, links reachable only from this machine. * ``RemoteLiveOwner`` -- the public server's owner API (``LIVE_TRACKING_URL`` + ``LIVE_TRACKING_ADMIN_SECRET``), links reachable from the internet. The public server keeps only the SHA-256 of a token. So that the panel can show a link again after a reload, the owner's machine keeps the issued links in its own git-ignored file (``LinkBook``) -- the owner's record, like an admin panel's list, never sent anywhere. """ from __future__ import annotations import json import os import threading import urllib.error import urllib.request from pathlib import Path from typing import Any, Optional, Sequence from urllib.parse import quote from mountain_twin.live_tracking.service import LiveTrackingError, LiveTrackingService from mountain_twin.offline import urlopen as offline_urlopen PATH_OF = {"view": "v", "push": "p"} class LinkBook: def __init__(self, path: Path): self.path = path self._lock = threading.Lock() def _read(self) -> dict[str, Any]: try: return json.loads(self.path.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError): return {} def get(self, backend: str, journey_id: str, kind: str, link_id: str) -> Optional[str]: entry = self._read().get(backend, {}).get(journey_id, {}).get(kind) return entry["url"] if entry and entry.get("link_id") == link_id else None def put(self, backend: str, journey_id: str, kind: str, link_id: str, url: str) -> None: with self._lock: book = self._read() book.setdefault(backend, {}).setdefault(journey_id, {})[kind] = { "link_id": link_id, "url": url, } self._write(book) def forget(self, backend: str, journey_id: str, kind: Optional[str] = None) -> None: with self._lock: book = self._read() journeys = book.get(backend, {}) if kind is None: journeys.pop(journey_id, None) else: journeys.get(journey_id, {}).pop(kind, None) self._write(book) def _write(self, book: dict[str, Any]) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) temporary = self.path.with_suffix(".tmp") temporary.write_text(json.dumps(book, ensure_ascii=False, indent=1), encoding="utf-8") os.chmod(temporary, 0o600) temporary.replace(self.path) class LiveOwner: """Shared by both backends: publishing, links with their URLs, status.""" kind_label = "" backend = "" def __init__(self, book: LinkBook): self.book = book # backend operations, implemented below def _status(self, journey_id: str) -> dict[str, Any]: raise NotImplementedError # noqa: E704 def _publish(self, journey_id: str, document: dict[str, Any]) -> None: raise NotImplementedError # noqa: E704 def _issue(self, journey_id: str, kind: str, password: Optional[str]) -> dict[str, Any]: raise NotImplementedError # noqa: E704 def _revoke(self, journey_id: str, kind: str) -> None: raise NotImplementedError # noqa: E704 def _delete_positions(self, journey_id: str) -> int: raise NotImplementedError # noqa: E704 def _delete_journey(self, journey_id: str) -> None: raise NotImplementedError # noqa: E704 def public_base_url(self, request_origin: str) -> str: raise NotImplementedError # noqa: E704 def status( self, journey_id: str, route_revision_id: Optional[str], request_origin: str ) -> dict[str, Any]: document = self._status(journey_id) for kind, link in document["links"].items(): if link: link["url"] = self.book.get(self.backend, journey_id, kind, link["link_id"]) document["route_is_current"] = ( document["published"] and document["route_revision_id"] == route_revision_id ) document["server"] = { "kind": self.kind_label, "base_url": self.public_base_url(request_origin), } return document def issue( self, journey_id: str, kind: str, password: Optional[str], *, title: str, route_revision_id: Optional[str], latlngs: Sequence[Sequence[float]], request_origin: str, ) -> dict[str, Any]: # The route the page shows is the Journey's current one, sent with # every new link (a few hundred KB at most). self._publish( journey_id, { "title": title, "route_revision_id": route_revision_id, "route": [list(p) for p in latlngs], }, ) issued = self._issue(journey_id, kind, password) url = f"{self.public_base_url(request_origin)}/live/{PATH_OF[kind]}/{issued.pop('token')}" self.book.put(self.backend, journey_id, kind, issued["link_id"], url) return {**issued, "url": url} def revoke(self, journey_id: str, kind: str) -> None: self._revoke(journey_id, kind) self.book.forget(self.backend, journey_id, kind) def delete_positions(self, journey_id: str) -> int: return self._delete_positions(journey_id) def delete_journey(self, journey_id: str) -> None: self._delete_journey(journey_id) self.book.forget(self.backend, journey_id) class LocalLiveOwner(LiveOwner): kind_label = "local" def __init__(self, service: LiveTrackingService, book: LinkBook): super().__init__(book) self.service = service self.backend = "local" def public_base_url(self, request_origin: str) -> str: return request_origin.rstrip("/") def _status(self, journey_id): return self.service.status(journey_id) def _publish(self, journey_id, document): self.service.publish( journey_id, document["title"], document["route_revision_id"], document["route"] ) def _issue(self, journey_id, kind, password): return self.service.issue_link(journey_id, kind, password) def _revoke(self, journey_id, kind): self.service.revoke_link(journey_id, kind) def _delete_positions(self, journey_id): return self.service.delete_positions(journey_id) def _delete_journey(self, journey_id): self.service.delete_journey(journey_id) class RemoteLiveOwner(LiveOwner): kind_label = "remote" def __init__( self, base_url: str, admin_secret: str, book: LinkBook, *, timeout: float = 20.0, opener=None, ): super().__init__(book) self.base_url = base_url.rstrip("/") self.backend = self.base_url self._secret = admin_secret self._timeout = timeout self._open = opener or offline_urlopen # AV-045: offline mode for tests def public_base_url(self, request_origin: str) -> str: return self.base_url def _call(self, method: str, path: str, document: Optional[dict[str, Any]] = None) -> Any: data = json.dumps(document).encode("utf-8") if document is not None else None request = urllib.request.Request( f"{self.base_url}/live/admin/api/journeys/{path}", data=data, method=method, headers={"Authorization": f"Bearer {self._secret}", "Content-Type": "application/json"}, ) try: with self._open(request, timeout=self._timeout) as response: return json.loads(response.read() or b"null") except urllib.error.HTTPError as error: try: body = json.loads(error.read() or b"{}") except json.JSONDecodeError: body = {} raise LiveTrackingError( error.code, body.get("error", "LIVE_SERVER_ERROR"), body.get("message") or f"Serwer śledzenia odpowiedział błędem {error.code}.", ) from None except (urllib.error.URLError, TimeoutError, OSError) as error: raise LiveTrackingError( 502, "LIVE_SERVER_UNREACHABLE", f"Serwer śledzenia ({self.base_url}) nie odpowiada: {error}.", ) from None def _status(self, journey_id): return self._call("GET", quote(journey_id, safe="")) def _publish(self, journey_id, document): self._call("PUT", quote(journey_id, safe=""), document) def _issue(self, journey_id, kind, password): return self._call( "POST", f"{quote(journey_id, safe='')}/links/{kind}", {"password": password or None} ) def _revoke(self, journey_id, kind): self._call("DELETE", f"{quote(journey_id, safe='')}/links/{kind}") def _delete_positions(self, journey_id): return self._call("DELETE", f"{quote(journey_id, safe='')}/positions")["deleted"] def _delete_journey(self, journey_id): self._call("DELETE", quote(journey_id, safe=""))