220 lines
8.6 KiB
Python
220 lines
8.6 KiB
Python
"""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))
|