diff --git a/docs/api.md b/docs/api.md index f3fd75b..91ba667 100644 --- a/docs/api.md +++ b/docs/api.md @@ -32,8 +32,10 @@ } ``` -- `rtsp_url_env` 必填；应用从同名环境变量读取真实 URL。 +- 来源二选一：`rtsp_url_env`（从同名环境变量读取真实 URL）**或**结构化字段 `host`/`port`(默认 554)/`channel`(默认 `101`，`102` 为子码流)/`username`/`password`；两者都会在内存中解析出 RTSP URL。结构化凭证只允许出现在**未跟踪的** `config.local.json`；公开示例 `config.example.json` 只能用 `rtsp_url_env`，禁止任何 `host`/`username`/`password`/完整 URL（有测试守卫）。 +- 结构化模式下应用用 `build_rtsp_url` 拼接，并对用户名/密码做 RFC 3986 百分号编码，因此密码含 `@`、`:`、`/` 也不会破坏 URL 解析。 - `source.mode` 显式声明来源类型，取值 `stream`（默认，实时 RTSP，允许有界重连）或 `replay`（本地录像，EOF 不重放）；由配置决定，不再按 URL 前缀猜测。 +- 凭证（`host`/`username`/`password`/解析出的 URL）绝不进入 `runtime_config_version`、事件 JSONL、日志或截图文件名。 - `model.confidence_threshold` 是模型检测置信度，可在设置草稿中调整，并在下次开始监控时经 `PoseAdapter.set_confidence_threshold` 真正生效。 - 数值是待现场录像校准的默认值；每个值必须真正进入事件逻辑：`keypoint_confidence_threshold` 决定姿态质量门槛；`suspect_window_seconds` 限制快速下移到水平姿态的最大间隔；`confirm_window_seconds` 是水平倒地候选需持续的确认时间；`recovery_window_seconds` 是恢复姿态需持续的时间；`cooldown_seconds` 是确认事件后允许开始恢复判断前的最短等待时间。示例中的全零 SHA-256 只占位配置形状，T-103 必须以受控模型的真实哈希替换并验证后才能启动推理。 - 缺少环境变量、模型不存在或哈希不符时，应用显示配置错误，不启动监控。 diff --git a/v1/camera.py b/v1/camera.py new file mode 100644 index 0000000..f4b307c --- /dev/null +++ b/v1/camera.py @@ -0,0 +1,57 @@ +"""RTSP URL building and a bounded connection probe for camera setup. + +Kept Qt-free so URL construction (including credential percent-encoding) and the +connection test are unit-testable without a display or a real camera. Credentials +are only ever held in memory here; callers must not log or persist the built URL. +""" + +from dataclasses import dataclass +from typing import Optional + +import numpy as np + +from v1.config import build_rtsp_url # re-exported for the settings UI +from v1.video_source import SourceMode, SourceStatus, VideoSource + +__all__ = ["build_rtsp_url", "ProbeResult", "probe_stream"] + + +@dataclass(frozen=True) +class ProbeResult: + ok: bool + message: str + frame: Optional[np.ndarray] + + +def probe_stream( + url: str, + attempts: int = 15, + capture_factory=None, +) -> ProbeResult: + """Try to open the stream and grab one frame; never raise on a bad source. + + Returns a decoded frame on success so the caller can show a preview. The + message is safe to display: it never contains the URL or credentials. + """ + + source = VideoSource( + url, + SourceMode.STREAM, + retry_initial_seconds=0.1, + retry_max_seconds=0.2, + capture_factory=capture_factory, + ) + last_error = None + try: + for _ in range(max(1, attempts)): + packet = source.read() + if packet.image is not None: + height, width = packet.image.shape[:2] + return ProbeResult(True, "连接成功 {0}x{1}".format(width, height), packet.image) + if packet.error: + last_error = packet.error + if packet.status is SourceStatus.ERROR: + return ProbeResult(False, "连接失败：{0}".format(last_error or "无法打开来源"), None) + return ProbeResult(False, "连接超时，未获取到画面", None) + finally: + source.close() diff --git a/v1/config.py b/v1/config.py index 0dcbae2..e627241 100644 --- a/v1/config.py +++ b/v1/config.py @@ -7,12 +7,38 @@ 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 @@ -106,13 +132,26 @@ def load_config(path: Path) -> AppConfig: 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)) + 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"): diff --git a/v1/tests/test_camera.py b/v1/tests/test_camera.py new file mode 100644 index 0000000..0f9f278 --- /dev/null +++ b/v1/tests/test_camera.py @@ -0,0 +1,49 @@ +import numpy as np + +from v1.camera import build_rtsp_url, probe_stream + + +class _FakeCapture: + def __init__(self, frames): + self._frames = list(frames) if frames is not None else None + + def isOpened(self): + return self._frames is not None + + def read(self): + if self._frames: + return True, self._frames.pop(0) + return False, None + + def release(self): + pass + + +def test_build_rtsp_url_percent_encodes_special_characters_in_password(): + # Example TEST-NET address and a synthetic password (never a real credential). + url = build_rtsp_url("192.0.2.10", 554, "admin", "p@ss/w:d", "102") + + assert url == "rtsp://admin:p%40ss%2Fw%3Ad@192.0.2.10:554/Streaming/Channels/102" + + +def test_build_rtsp_url_defaults_to_main_channel(): + url = build_rtsp_url("192.0.2.10", 554, "admin", "secret") + + assert url.endswith("/Streaming/Channels/101") + + +def test_probe_stream_returns_a_frame_on_success(): + frame = np.zeros((480, 640, 3), dtype=np.uint8) + + result = probe_stream("rtsp://x", capture_factory=lambda _u: _FakeCapture([frame])) + + assert result.ok is True + assert "640x480" in result.message + assert result.frame is not None + + +def test_probe_stream_reports_failure_without_raising(): + result = probe_stream("rtsp://x", attempts=3, capture_factory=lambda _u: _FakeCapture(None)) + + assert result.ok is False + assert result.frame is None diff --git a/v1/tests/test_config.py b/v1/tests/test_config.py index 6ce568f..c3f448b 100644 --- a/v1/tests/test_config.py +++ b/v1/tests/test_config.py @@ -56,6 +56,68 @@ def test_load_config_rejects_embedded_source_address(tmp_path): load_config(config_file) +def test_structured_source_builds_percent_encoded_rtsp_url(tmp_path): + config_file = tmp_path / "config.json" + _write_config( + config_file, + { + "id": "hik-ipcamera-101", + "host": "192.0.2.10", + "port": 554, + "channel": "102", + "username": "admin", + "password": "p@ss/w:d", + }, + ) + + config = load_config(config_file) + + assert config.source_url == "rtsp://admin:p%40ss%2Fw%3Ad@192.0.2.10:554/Streaming/Channels/102" + assert config.source_id == "hik-ipcamera-101" + assert config.source_mode == "stream" + + +def test_structured_source_requires_password(tmp_path): + config_file = tmp_path / "config.json" + _write_config( + config_file, + {"id": "hik-ipcamera-101", "host": "192.0.2.10", "username": "admin"}, + ) + + with pytest.raises(ConfigError, match="password"): + load_config(config_file) + + +def test_structured_credentials_are_excluded_from_config_version(tmp_path): + first_file = tmp_path / "a.json" + second_file = tmp_path / "b.json" + _write_config( + first_file, + {"id": "hik-101", "host": "192.0.2.10", "username": "admin", "password": "secret-a"}, + ) + _write_config( + second_file, + {"id": "hik-101", "host": "192.0.2.20", "username": "operator", "password": "secret-b"}, + ) + + first = load_config(first_file) + second = load_config(second_file) + + assert first.source_url != second.source_url + assert first.runtime_config_version == second.runtime_config_version + + +def test_public_example_config_has_no_embedded_credentials(): + from pathlib import Path + + example = json.loads(Path("v1/config.example.json").read_text(encoding="utf-8")) + source = example["source"] + + for forbidden in ("host", "username", "password", "url", "rtsp_url"): + assert forbidden not in source + assert "rtsp_url_env" in source + + def test_source_mode_defaults_to_stream(tmp_path, monkeypatch): config_file = tmp_path / "config.json" _write_config(