147 lines
6.6 KiB
Python
147 lines
6.6 KiB
Python
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()
|