feat(v1): add secure configuration baseline
This commit is contained in:
@@ -0,0 +1,5 @@
|
||||
"""Silver Pose Python V1 package.
|
||||
|
||||
V1 keeps configuration, video acquisition, pose inference, tracking, evidence,
|
||||
and temporal event decisions in separate modules.
|
||||
"""
|
||||
@@ -0,0 +1,21 @@
|
||||
{
|
||||
"source": {
|
||||
"id": "lobby-camera-01",
|
||||
"rtsp_url_env": "SILVER_POSE_RTSP_URL"
|
||||
},
|
||||
"model": {
|
||||
"path": "models/best.pt",
|
||||
"sha256": "0000000000000000000000000000000000000000000000000000000000000000",
|
||||
"confidence_threshold": 0.25
|
||||
},
|
||||
"event": {
|
||||
"keypoint_confidence_threshold": 0.4,
|
||||
"suspect_window_seconds": 0.5,
|
||||
"confirm_window_seconds": 1.8,
|
||||
"recovery_window_seconds": 2.0,
|
||||
"cooldown_seconds": 10.0
|
||||
},
|
||||
"artifacts": {
|
||||
"event_dir": "../artifacts/events"
|
||||
}
|
||||
}
|
||||
+146
@@ -0,0 +1,146 @@
|
||||
"""Validated, credential-safe configuration loading for Silver Pose V1."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict
|
||||
|
||||
|
||||
class ConfigError(ValueError):
|
||||
"""Raised when a configuration file cannot safely start V1."""
|
||||
|
||||
|
||||
@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
|
||||
|
||||
|
||||
_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 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 rtsp_url_env, not an embedded address")
|
||||
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))
|
||||
|
||||
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"),
|
||||
)
|
||||
@@ -0,0 +1,5 @@
|
||||
numpy==1.24.4
|
||||
opencv-python==4.6.0.66
|
||||
ultralytics==8.3.205
|
||||
PyQt5==5.15.9
|
||||
pytest==8.3.5
|
||||
@@ -0,0 +1,56 @@
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from v1.config import ConfigError, load_config
|
||||
|
||||
|
||||
def _write_config(path, source):
|
||||
path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"source": source,
|
||||
"model": {
|
||||
"path": "models/best.pt",
|
||||
"sha256": "a" * 64,
|
||||
"confidence_threshold": 0.25,
|
||||
},
|
||||
"event": {
|
||||
"keypoint_confidence_threshold": 0.4,
|
||||
"suspect_window_seconds": 0.5,
|
||||
"confirm_window_seconds": 1.8,
|
||||
"recovery_window_seconds": 2.0,
|
||||
"cooldown_seconds": 10.0,
|
||||
},
|
||||
"artifacts": {"event_dir": "artifacts/events"},
|
||||
},
|
||||
ensure_ascii=False,
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
|
||||
def test_load_config_resolves_source_from_environment_variable(tmp_path, monkeypatch):
|
||||
config_file = tmp_path / "config.json"
|
||||
_write_config(
|
||||
config_file,
|
||||
{"id": "lobby-camera-01", "rtsp_url_env": "SILVER_POSE_RTSP_URL"},
|
||||
)
|
||||
monkeypatch.setenv("SILVER_POSE_RTSP_URL", "rtsp://demo.invalid/live")
|
||||
|
||||
config = load_config(config_file)
|
||||
|
||||
assert config.source_id == "lobby-camera-01"
|
||||
assert config.source_url == "rtsp://demo.invalid/live"
|
||||
assert config.event.confirm_window_seconds == 1.8
|
||||
|
||||
|
||||
def test_load_config_rejects_embedded_source_address(tmp_path):
|
||||
config_file = tmp_path / "config.json"
|
||||
_write_config(
|
||||
config_file,
|
||||
{"id": "lobby-camera-01", "url": "rtsp://demo.invalid/live"},
|
||||
)
|
||||
|
||||
with pytest.raises(ConfigError, match="rtsp_url_env"):
|
||||
load_config(config_file)
|
||||
Reference in New Issue
Block a user