tests/test_security.py

import asyncio
import os
import unittest
import uuid
from unittest.mock import patch

os.environ["DATABASE_URL"] = "sqlite+aiosqlite:///./test_voice_agent.db"
os.environ["WARMUP_STT"] = "false"
os.environ["WARMUP_RAG"] = "false"
os.environ["GEMINI_API_KEY"] = ""

from fastapi import HTTPException
from fastapi.testclient import TestClient
from starlette.requests import Request
from twilio.request_validator import RequestValidator

from backend.core.security import (
    create_twilio_stream_token,
    rate_limiter,
    validate_production_config,
    validate_twilio_webhook,
    verify_twilio_stream_token,
)
from backend.main import app


class SecurityTests(unittest.TestCase):
    def test_admin_endpoints_require_configured_key(self):
        admin_key = "a" * 32
        with patch.dict(os.environ, {"ADMIN_API_KEY": admin_key}):
            with TestClient(app) as client:
                self.assertEqual(client.get("/api/admin/verify").status_code, 401)
                self.assertEqual(
                    client.get(
                        "/api/admin/verify",
                        headers={"X-API-Key": admin_key},
                    ).status_code,
                    200,
                )
                self.assertEqual(client.get("/api/leads").status_code, 401)

    def test_production_configuration_fails_closed(self):
        with patch.dict(
            os.environ,
            {
                "ENVIRONMENT": "production",
                "ADMIN_API_KEY": "short",
                "GEMINI_API_KEY": "",
                "DATABASE_URL": "sqlite:///local.db",
                "CORS_ORIGINS": "http://localhost:5173",
            },
        ):
            with self.assertRaises(RuntimeError):
                validate_production_config()

    def test_valid_production_configuration_is_accepted(self):
        environment = {
            "ENVIRONMENT": "production",
            "ADMIN_API_KEY": "a" * 32,
            "GEMINI_API_KEY": "configured",
            "DATABASE_URL": "postgresql+asyncpg://user:pass@db/app",
            "CORS_ORIGINS": "https://voice.example.com",
            "APP_WORKERS": "1",
            "ENABLE_PHONE_CALLS": "false",
        }
        with patch.dict(os.environ, environment):
            validate_production_config()

    def test_twilio_stream_tokens_are_signed_and_expire(self):
        with patch.dict(os.environ, {"TWILIO_STREAM_SECRET": "s" * 32}):
            token = create_twilio_stream_token()
            self.assertTrue(verify_twilio_stream_token(token))
            self.assertFalse(verify_twilio_stream_token(token + "changed"))
            self.assertFalse(verify_twilio_stream_token("invalid"))

    def test_twilio_webhook_signature_is_verified(self):
        auth_token = "twilio-test-auth-token"
        url = "https://voice.example.com/api/calls/incoming"
        form = {"CallSid": "CA123", "From": "+910000000000"}
        signature = RequestValidator(auth_token).compute_signature(url, form)
        scope = {
            "type": "http",
            "method": "POST",
            "scheme": "https",
            "path": "/api/calls/incoming",
            "raw_path": b"/api/calls/incoming",
            "query_string": b"",
            "headers": [(b"x-twilio-signature", signature.encode())],
            "server": ("voice.example.com", 443),
            "client": ("127.0.0.1", 1234),
        }
        request = Request(scope)
        environment = {
            "ENVIRONMENT": "production",
            "PUBLIC_BASE_URL": "https://voice.example.com",
            "TWILIO_AUTH_TOKEN": auth_token,
        }
        with patch.dict(os.environ, environment):
            asyncio.run(validate_twilio_webhook(request, form))
            scope["headers"] = [(b"x-twilio-signature", b"invalid")]
            with self.assertRaises(HTTPException):
                asyncio.run(validate_twilio_webhook(Request(scope), form))

    def test_rate_limiter_rejects_excess_requests(self):
        key = f"test:{uuid.uuid4()}"

        async def exercise():
            await rate_limiter.check(key, 2)
            await rate_limiter.check(key, 2)
            with self.assertRaises(HTTPException) as caught:
                await rate_limiter.check(key, 2)
            self.assertEqual(caught.exception.status_code, 429)

        asyncio.run(exercise())

    def test_readiness_and_security_headers(self):
        with TestClient(app) as client:
            response = client.get("/ready")
            self.assertEqual(response.status_code, 503)
            self.assertEqual(response.json()["status"], "not_ready")
            health = client.get("/health")
            self.assertEqual(health.headers["x-content-type-options"], "nosniff")
            self.assertEqual(health.headers["x-frame-options"], "DENY")
            self.assertTrue(health.headers["x-request-id"])


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