"""Secure live-stream access. Camera credentials are never exposed. A customer can request short-lived, HMAC-signed access tokens for streams authorized for their own orders; the `verify` endpoint validates the token and returns a protocol-aware playable stream descriptor (URL rewritten through a media gateway when needed). """ from __future__ import annotations import base64 import hashlib import hmac import json import time from typing import Any from urllib.parse import urlencode, urlsplit, urlunsplit from django.conf import settings from apps.core.settings_service import SettingsService from apps.live_monitoring.models import Camera, WorkflowStepCamera from apps.live_monitoring.protocols import ( DEFAULT_PROTOCOL, STREAM_PROTOCOLS, detect_protocol_from_url, get_protocol, ) from apps.orders.models import ProcessingOrder from apps.workflows.models import WorkflowStep class StreamAccessService: def can_access(self, user, camera: Camera) -> bool: if not user or not user.is_authenticated: return False if user.is_superuser: return True version_ids = ProcessingOrder.objects.filter(customer__user=user).values_list( "workflow_version_id", flat=True ) step_ids = WorkflowStep.objects.filter( version_id__in=version_ids, live_monitoring_enabled=True ).values_list("id", flat=True) return WorkflowStepCamera.objects.filter(camera=camera, step_id__in=step_ids).exists() def _ttl(self, ttl: int | None) -> int: if ttl is not None: return int(ttl) try: return int(SettingsService().get("camera.stream_ttl_seconds", 3600)) except (TypeError, ValueError): return 3600 def issue_token(self, camera: Camera, ttl: int | None = None) -> tuple: resolved_ttl = self._ttl(ttl) payload = {"camera": str(camera.id), "exp": int(time.time()) + resolved_ttl} raw = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip("=") sig = hmac.new(settings.SECRET_KEY.encode(), raw.encode(), hashlib.sha256).hexdigest() return f"{raw}.{sig}", payload["exp"] def verify_token(self, token: str): try: raw, sig = token.split(".", 1) except (ValueError, AttributeError): return None expected = hmac.new(settings.SECRET_KEY.encode(), raw.encode(), hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, sig): return None try: pad = raw + "=" * (-len(raw) % 4) payload = json.loads(base64.urlsafe_b64decode(pad).decode()) except Exception: return None if payload.get("exp", 0) < int(time.time()): return None return payload def resolve_stream(self, camera: Camera) -> dict[str, Any]: """Build a client-safe stream descriptor for an authorized camera. Credentials from ``auth_config`` / global settings are applied server-side only. When the protocol is not browser-native and a media gateway is configured, the URL is rewritten to the gateway playlist. """ service = SettingsService() if not service.get_bool("camera.enabled", True): return { "protocol": camera.resolved_protocol(), "stream_url": None, "browser_native": False, "mime_type": "", "player": "none", "error": "Live monitoring is disabled in settings.", } protocol_key = camera.protocol or detect_protocol_from_url(camera.stream_url) or camera.resolved_protocol() if protocol_key not in STREAM_PROTOCOLS: protocol_key = DEFAULT_PROTOCOL meta = get_protocol(protocol_key) playable_url = self._build_playable_url(camera, protocol_key, service) browser_native = bool(meta.get("browser_native")) # Non-browser protocols need a gateway remux (typically to HLS). if not browser_native: gateway = (service.get("camera.media_gateway_url", "") or "").rstrip("/") if gateway: template = service.get("camera.media_gateway_path_template", "{camera_id}/index.m3u8") or "{camera_id}/index.m3u8" path = template.format(camera_id=str(camera.id), protocol=protocol_key) playable_url = f"{gateway}/{path.lstrip('/')}" protocol_key = "hls" meta = get_protocol("hls") browser_native = True else: return { "protocol": protocol_key, "stream_url": None, "browser_native": False, "mime_type": meta.get("mime_type", ""), "player": "none", "transport": self._transport(protocol_key, camera, service), "error": ( f"{meta['label']} is not playable in the browser. " "Configure a media gateway URL in Live Monitoring settings." ), } player = self._player_hint(protocol_key) result: dict[str, Any] = { "protocol": protocol_key, "stream_url": playable_url, "browser_native": browser_native, "mime_type": meta.get("mime_type", ""), "player": player, "transport": self._transport(protocol_key, camera, service), } if protocol_key == "webrtc": ice = service.get("camera.webrtc.ice_servers", "") if ice: result["ice_servers"] = ice return result def _transport(self, protocol_key: str, camera: Camera, service: SettingsService) -> str | None: auth = camera.auth_config or {} if auth.get("transport"): return auth["transport"] value = service.get(f"camera.{protocol_key}.transport") return value or None def _player_hint(self, protocol_key: str) -> str: return { "hls": "hls", "dash": "dash", "webrtc": "webrtc", "mjpeg": "mjpeg", "http": "video", }.get(protocol_key, "none") def _build_playable_url(self, camera: Camera, protocol_key: str, service: SettingsService) -> str: """Return a URL safe to hand to the client (no embedded passwords).""" raw = (camera.stream_url or "").strip() auth = dict(camera.auth_config or {}) # Absolute URL already provided — strip userinfo credentials. if "://" in raw: return self._strip_userinfo(raw) # Relative / host-based config: assemble from protocol defaults + auth_config. meta = get_protocol(protocol_key) prefer_secure = service.get_bool("camera.prefer_secure", True) schemes = meta["schemes"] scheme = schemes[0] if prefer_secure: for candidate in schemes: if candidate.endswith("s") or candidate in {"https", "wss", "rtsps", "rtmps"}: scheme = candidate break host = auth.get("host") or raw if not host: return raw port = auth.get("port") if port in (None, ""): port = service.get(f"camera.{protocol_key}.default_port", meta["default_port"]) path = (auth.get("path") or service.get(f"camera.{protocol_key}.default_path", meta["default_path"]) or "").lstrip("/") netloc = f"{host}:{port}" if port not in (None, "") else host url = urlunsplit((scheme, netloc, path, "", "")) # Query-token auth can be appended without embedding secrets in userinfo. auth_method = ( auth.get("auth_method") or service.get("camera.default_auth_method", "none") or "none" ) token = auth.get("token") or auth.get("password") if auth_method == "query_token" and token: sep = "&" if "?" in url else "?" url = f"{url}{sep}{urlencode({'token': token})}" return url @staticmethod def _strip_userinfo(url: str) -> str: try: parts = urlsplit(url) except ValueError: return url if not parts.username and not parts.password: return url host = parts.hostname or "" if parts.port: host = f"{host}:{parts.port}" # Bracket IPv6 if parts.hostname and ":" in parts.hostname and not host.startswith("["): host = f"[{parts.hostname}]" if parts.port: host = f"{host}:{parts.port}" return urlunsplit((parts.scheme, host, parts.path, parts.query, parts.fragment))