"""Validated, credential-safe configuration loading for Silver Pose V1.""" import hashlib import json import os import re from dataclasses import dataclass from pathlib import Path from typing import Any, Dict from urllib.parse import quote class ConfigError(ValueError): """Raised when a configuration file cannot safely start V1.""" # Hikvision RTSP path: channel "101" = channel 1 main stream, "102" = sub stream. _HIK_PATH = "/Streaming/Channels/{0}" def build_rtsp_url( host: str, port: int, username: str, password: str, channel: str = "101", path_template: str = _HIK_PATH, ) -> str: """Build an RTSP URL, percent-encoding credentials per RFC 3986. Encoding the username and password means special characters such as ``@``, ``:`` and ``/`` in a password no longer break URL parsing. The returned URL holds credentials and must never be logged or persisted. """ user = quote(str(username), safe="") secret = quote(str(password), safe="") path = path_template.format(quote(str(channel), safe="")) return "rtsp://{0}:{1}@{2}:{3}{4}".format(user, secret, host, int(port), path) @dataclass(frozen=True) class EventConfig: keypoint_confidence_threshold: float suspect_window_seconds: float confirm_window_seconds: float recovery_window_seconds: float cooldown_seconds: float @dataclass(frozen=True) class AppConfig: source_id: str source_url: str model_path: Path model_sha256: str confidence_threshold: float event: EventConfig event_dir: Path source_mode: str = "stream" @property def runtime_config_version(self) -> str: """Return a stable, non-secret identifier for the active event settings.""" payload = { "source_id": self.source_id, "model_sha256": self.model_sha256, "confidence_threshold": self.confidence_threshold, "event": { "keypoint_confidence_threshold": self.event.keypoint_confidence_threshold, "suspect_window_seconds": self.event.suspect_window_seconds, "confirm_window_seconds": self.event.confirm_window_seconds, "recovery_window_seconds": self.event.recovery_window_seconds, "cooldown_seconds": self.event.cooldown_seconds, }, } canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")) return "cfg-" + hashlib.sha256(canonical.encode("utf-8")).hexdigest() _ENVIRONMENT_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") _SHA256 = re.compile(r"^[0-9a-fA-F]{64}$") def _mapping(value: Any, field_name: str) -> Dict[str, Any]: if not isinstance(value, dict): raise ConfigError("{0} must be an object".format(field_name)) return value def _text(value: Any, field_name: str) -> str: if not isinstance(value, str) or not value.strip(): raise ConfigError("{0} must be a non-empty string".format(field_name)) return value.strip() def _number(value: Any, field_name: str, minimum: float, maximum: float) -> float: try: result = float(value) except (TypeError, ValueError): raise ConfigError("{0} must be numeric".format(field_name)) if not minimum <= result <= maximum: raise ConfigError( "{0} must be between {1} and {2}".format(field_name, minimum, maximum) ) return result def _resolve_path(config_path: Path, value: Any, field_name: str) -> Path: raw_path = Path(_text(value, field_name)) if raw_path.is_absolute(): return raw_path return (config_path.parent / raw_path).resolve() def write_local_camera_source( config_path: Path, source_id: str, host: str, port: int, channel: str, username: str, password: str, mode: str = "stream", ) -> Path: """Persist a structured camera source into an untracked local config file. Only ever call this on ``config.local.json`` (git-ignored). Credentials are written in plaintext by design of the chosen policy; the public example must never receive them. """ path = Path(config_path) data = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {} data["source"] = { "id": str(source_id), "host": str(host), "port": int(port), "channel": str(channel), "username": str(username), "password": str(password), "mode": mode, } path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return path def load_config(path: Path) -> AppConfig: """Load one public/local configuration pair without persisting credentials. The config file can name an environment variable but must not embed a source address. The resolved address remains in memory only. """ config_path = Path(path).resolve() try: raw = json.loads(config_path.read_text(encoding="utf-8")) except OSError as exc: raise ConfigError("cannot read config: {0}".format(exc)) except json.JSONDecodeError as exc: raise ConfigError("invalid JSON config: {0}".format(exc)) root = _mapping(raw, "config") source = _mapping(root.get("source"), "source") if "url" in source or "rtsp_url" in source: raise ConfigError("source must define structured host fields or rtsp_url_env, not a full URL") if "host" in source: host = _text(source.get("host"), "source.host") port = int(_number(source.get("port", 554), "source.port", 1.0, 65535.0)) channel = str(source.get("channel", "101")).strip() if not channel: raise ConfigError("source.channel must be a non-empty value") username = _text(source.get("username"), "source.username") password = source.get("password") if not isinstance(password, str) or password == "": raise ConfigError("source.password must be a non-empty string") source_url = build_rtsp_url(host, port, username, password, channel) else: environment_name = _text(source.get("rtsp_url_env"), "source.rtsp_url_env") if not _ENVIRONMENT_NAME.match(environment_name): raise ConfigError("source.rtsp_url_env must be an environment variable name") source_url = os.environ.get(environment_name) if not source_url: raise ConfigError("missing RTSP environment variable: {0}".format(environment_name)) source_mode = source.get("mode", "stream") if source_mode not in ("replay", "stream"): raise ConfigError("source.mode must be 'replay' or 'stream'") model = _mapping(root.get("model"), "model") model_sha256 = _text(model.get("sha256"), "model.sha256").lower() if not _SHA256.match(model_sha256): raise ConfigError("model.sha256 must be a 64-character SHA-256 value") event_raw = _mapping(root.get("event"), "event") event = EventConfig( keypoint_confidence_threshold=_number( event_raw.get("keypoint_confidence_threshold"), "event.keypoint_confidence_threshold", 0.0, 1.0, ), suspect_window_seconds=_number( event_raw.get("suspect_window_seconds"), "event.suspect_window_seconds", 0.0, 30.0, ), confirm_window_seconds=_number( event_raw.get("confirm_window_seconds"), "event.confirm_window_seconds", 1.0, 3.0, ), recovery_window_seconds=_number( event_raw.get("recovery_window_seconds"), "event.recovery_window_seconds", 0.0, 300.0, ), cooldown_seconds=_number( event_raw.get("cooldown_seconds"), "event.cooldown_seconds", 0.0, 3600.0, ), ) artifacts = _mapping(root.get("artifacts"), "artifacts") return AppConfig( source_id=_text(source.get("id"), "source.id"), source_url=source_url, model_path=_resolve_path(config_path, model.get("path"), "model.path"), model_sha256=model_sha256, confidence_threshold=_number( model.get("confidence_threshold"), "model.confidence_threshold", 0.0, 1.0 ), event=event, event_dir=_resolve_path(config_path, artifacts.get("event_dir"), "artifacts.event_dir"), source_mode=source_mode, )