video_monitor/tests/test_phase3_security.py

import importlib.util
import logging
import sys
import tempfile
import types
import unittest
from pathlib import Path
from unittest import mock


ROOT = Path(__file__).resolve().parents[1]


class LifecycleTests(unittest.TestCase):
    def test_only_one_process_lock_owner_and_lock_is_recoverable(self):
        from app.services.lifecycle import ServiceLeaderLock
        with tempfile.TemporaryDirectory() as temp_dir:
            path = Path(temp_dir) / "leader.lock"
            first = ServiceLeaderLock(path)
            second = ServiceLeaderLock(path)
            self.assertTrue(first.acquire())
            self.assertFalse(second.acquire())
            first.release()
            self.assertTrue(second.acquire())
            second.release()

    def test_modules_have_no_import_time_service_start(self):
        global_utils = (ROOT / "app" / "utils" / "GlobalUtils.py").read_text(encoding="utf-8")
        inner = (ROOT / "app" / "views" / "InnerlView.py").read_text(encoding="utf-8")
        self.assertNotIn("g_gb28181SipServer.start()\n\ng_pull_stream_types", global_utils)
        self.assertNotIn("t_init_thread", inner)
        self.assertNotIn("threading.Thread(target=t_init_thread)", inner)

    def test_explicit_management_command_and_disabled_default(self):
        apps = (ROOT / "app" / "apps.py").read_text(encoding="utf-8")
        command = ROOT / "app" / "management" / "commands" / "runservices.py"
        self.assertTrue(command.is_file())
        self.assertIn('MONITOR_SERVICE_MODE", "disabled"', apps)
        self.assertIn("get_service_manager().start()", apps)


class TelemetryTests(unittest.TestCase):
    def test_endpoint_is_https_only(self):
        from app.services import telemetry
        with mock.patch.dict("os.environ", {"MONITOR_TELEMETRY_ENDPOINT": "http://example.com/h"}):
            with self.assertRaises(ValueError):
                telemetry._https_endpoint("MONITOR_TELEMETRY_ENDPOINT")
        with mock.patch.dict("os.environ", {"MONITOR_TELEMETRY_ENDPOINT": "https://example.com/h"}):
            self.assertEqual(
                telemetry._https_endpoint("MONITOR_TELEMETRY_ENDPOINT"), "https://example.com/h"
            )

    def test_payload_is_minimal_and_has_no_host_identifiers(self):
        from app.services import telemetry
        fake_settings = types.ModuleType("framework.settings")
        fake_settings.PROJECT_FLAG = "monitor"
        fake_settings.PROJECT_VERSION = "test"
        with mock.patch.dict(sys.modules, {"framework.settings": fake_settings}):
            payload = telemetry.heartbeat_payload(7)
        self.assertEqual(set(payload), {"event", "product", "version", "sequence"})
        serialized = str(payload).lower()
        for forbidden in ("ip", "mac", "hostname", "node", "cpu", "osinfo", "uname"):
            self.assertNotIn(forbidden, serialized)

    def test_default_config_explicitly_disables_outbound_features(self):
        config = (ROOT / "config.json").read_text(encoding="utf-8")
        self.assertIn('"telemetryEnabled": false', config)
        self.assertIn('"updateCheckEnabled": false', config)
        global_utils = (ROOT / "app" / "utils" / "GlobalUtils.py").read_text(encoding="utf-8")
        telemetry = (ROOT / "app" / "services" / "telemetry.py").read_text(encoding="utf-8")
        self.assertNotIn("http://www.yuturuishi.com", global_utils)
        self.assertIn("allow_redirects=False", telemetry)


class SecurityRegressionTests(unittest.TestCase):
    def _load_middleware_with_stubs(self):
        fake_http = types.ModuleType("django.http")

        class Redirect:
            def __init__(self, url):
                self.url = url

        class Json:
            def __init__(self, data, status=200):
                self.data, self.status_code = data, status

        fake_http.HttpResponseRedirect = Redirect
        fake_http.JsonResponse = Json
        fake_dep = types.ModuleType("django.utils.deprecation")
        fake_dep.MiddlewareMixin = object
        fake_security = types.ModuleType("app.security")
        fake_security.verify_internal_request = lambda request: False
        spec = importlib.util.spec_from_file_location(
            "phase3_middleware", ROOT / "app" / "middleware.py"
        )
        module = importlib.util.module_from_spec(spec)
        with mock.patch.dict(sys.modules, {
            "django.http": fake_http,
            "django.utils.deprecation": fake_dep,
            "app.security": fake_security,
        }):
            spec.loader.exec_module(module)
        return module

    def test_unauthenticated_and_safe_header_bypass_are_rejected(self):
        middleware = self._load_middleware_with_stubs()
        anonymous = types.SimpleNamespace(is_authenticated=False)
        request = types.SimpleNamespace(
            path_info="/system/config", method="GET", user=anonymous,
            headers={"Safe": "legacy-bypass"}, META={"HTTP_SAFE": "legacy-bypass"},
        )
        response = middleware.SimpleMiddleware().process_request(request)
        self.assertEqual(response.url, "/login")

    def test_path_traversal_and_command_argument_boundaries_remain(self):
        storage = (ROOT / "app" / "views" / "StorageView.py").read_text(encoding="utf-8")
        nvr = (ROOT / "app" / "views" / "NvrView.py").read_text(encoding="utf-8")
        self.assertIn("filename != os.path.basename(filename)", storage)
        self.assertIn("os.path.realpath", storage)
        self.assertNotIn('params.get("filename"', storage)
        self.assertIn("subprocess.run(command, shell=False", nvr)
        self.assertIn("re.fullmatch", nvr)

    def test_sql_model_upload_and_logging_boundaries_remain(self):
        stream = (ROOT / "app" / "views" / "StreamView.py").read_text(encoding="utf-8")
        users = (ROOT / "app" / "views" / "UserView.py").read_text(encoding="utf-8")
        models = (ROOT / "app" / "views" / "SmallModelView.py").read_text(encoding="utf-8")
        self.assertIn("g_database.select(sql, query_params)", stream)
        self.assertNotIn("nickname like '%{search_text}%'", stream)
        self.assertNotIn("username='%s'", users)
        self.assertIn("require_trusted_model(dest)", models)
        self.assertIn("os.unlink(dest)", models)

        from app.utils.Logger import SensitiveDataFilter
        record = logging.LogRecord(
            "test", logging.INFO, __file__, 1,
            "password='sensitive value' rtsp://user:pass@camera/live", (), None,
        )
        SensitiveDataFilter().filter(record)
        message = record.getMessage()
        self.assertNotIn("sensitive value", message)
        self.assertNotIn("user:pass", message)


if __name__ == "__main__":
    unittest.main()