File size: 4,972 Bytes
c794b6b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """Production security contract tests."""
import os
import sys
from unittest.mock import MagicMock
os.environ.setdefault("CEPHEUS_CLOUD", "1")
os.environ.setdefault("CEPHEUS_API_KEY", "test-key")
os.environ.pop("CEPHEUS_PRODUCTION", None)
for mod in ("google.adk", "google.adk.agents", "google.adk.runners", "google.adk.sessions"):
sys.modules.setdefault(mod, MagicMock())
BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if BACKEND_DIR not in sys.path:
sys.path.insert(0, BACKEND_DIR)
from fastapi.testclient import TestClient # noqa: E402
import main # noqa: E402
API_HEADERS = {"X-API-Key": "test-key"}
client = TestClient(main.app)
def test_cors_allows_delete():
r = client.options(
"/site/signage-placements/s1",
headers={
"Origin": "http://localhost:5173",
"Access-Control-Request-Method": "DELETE",
},
)
assert r.status_code == 200
allow = r.headers.get("access-control-allow-methods", "")
assert "DELETE" in allow
def test_face_files_not_publicly_mounted():
r = client.get("/face_database/test/person.jpg")
assert r.status_code == 404
def test_secure_face_proxy_missing_returns_404():
r = client.get("/files/face/test/person.jpg", headers=API_HEADERS)
assert r.status_code == 404
def test_ws_ticket_requires_auth_when_enabled():
os.environ["CEPHEUS_AUTH_DEV_MODE"] = "1"
try:
r = client.post("/auth/ws-ticket", headers=API_HEADERS)
assert r.status_code in (200, 401)
finally:
os.environ.pop("CEPHEUS_AUTH_DEV_MODE", None)
def test_fight_model_toggle_cloud_returns_bad_request():
"""Cloud stub engine does not support YOLO-based fight heuristic."""
r = client.post("/ai/toggle/fight", headers=API_HEADERS)
assert r.status_code == 400
detail = r.json().get("detail", "").lower()
assert "not available" in detail or "gpu" in detail
def test_ai_status_includes_capabilities():
r = client.get("/ai/status", headers=API_HEADERS)
assert r.status_code == 200
body = r.json()
assert "models" in body
assert "capabilities" in body
assert body["capabilities"]["fire"]["supported"] is False
def test_demo_login_rejected_without_demo_mode():
os.environ.pop("CEPHEUS_AUTH_DEV_MODE", None)
os.environ.pop("CEPHEUS_JWT_SECRET", None)
try:
r = client.post("/auth/login", json={"username": "anyone", "password": "anything"})
assert r.status_code == 503
finally:
os.environ["CEPHEUS_AUTH_DEV_MODE"] = "1"
def test_strict_startup_requires_jwt_secret():
import security_config
env = {
"CEPHEUS_CLOUD": "1",
"CEPHEUS_PRODUCTION": "1",
"CEPHEUS_API_KEY": "prod-key-not-default",
"CORS_ORIGINS": "https://example.com",
}
saved = {k: os.environ.get(k) for k in env}
os.environ.update(env)
os.environ.pop("CEPHEUS_AUTH_DEV_MODE", None)
os.environ.pop("CEPHEUS_JWT_SECRET", None)
try:
try:
security_config.validate_startup()
assert False, "expected SystemExit"
except SystemExit:
pass
finally:
for k, v in saved.items():
if v is None:
os.environ.pop(k, None)
else:
os.environ[k] = v
def test_readonly_api_key_blocks_writes():
os.environ["CEPHEUS_READONLY_API_KEY"] = "readonly-test-key"
try:
import importlib
import main as main_module
importlib.reload(main_module)
ro_client = TestClient(main_module.app)
r = ro_client.post(
"/alert",
headers={"X-API-Key": "readonly-test-key"},
json={"type": "test", "message": "x", "location": "y", "severity": "low"},
)
assert r.status_code == 403
g = ro_client.get("/alerts", headers={"X-API-Key": "readonly-test-key"})
assert g.status_code == 200
finally:
os.environ.pop("CEPHEUS_READONLY_API_KEY", None)
def test_query_string_auth_blocked_in_production():
os.environ["CEPHEUS_PRODUCTION"] = "1"
os.environ["CEPHEUS_JWT_SECRET"] = "prod-jwt-secret-min-32-characters-long"
os.environ["CEPHEUS_AUTH_DEV_MODE"] = "0"
try:
import importlib
import auth_service
import main as main_module
importlib.reload(auth_service)
importlib.reload(main_module)
# load_dotenv(override=True) in main.py resets env on reload — re-apply prod flags
os.environ["CEPHEUS_PRODUCTION"] = "1"
os.environ["CEPHEUS_JWT_SECRET"] = "prod-jwt-secret-min-32-characters-long"
os.environ["CEPHEUS_AUTH_DEV_MODE"] = "0"
prod_client = TestClient(main_module.app)
r = prod_client.get("/files/face/test/person.jpg?api_key=test-key", headers=API_HEADERS)
assert r.status_code == 400
finally:
os.environ.pop("CEPHEUS_PRODUCTION", None)
os.environ.pop("CEPHEUS_JWT_SECRET", None)
|