backend/tests/test_security.py

"""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)