| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253 |
- import asyncio
- import httpx
- from core_shared.config import ServiceSettings
- from core_shared.security import (
- add_internal_service_auth,
- build_internal_service_headers,
- mask_sensitive_mapping,
- )
- from fastapi import FastAPI
- def test_mask_sensitive_mapping_masks_nested_secrets() -> None:
- payload = {
- "api_key": "agp_super_secret_value",
- "nested": {"authorization": "Bearer token-value"},
- "safe": "visible",
- }
- masked = mask_sensitive_mapping(payload)
- assert masked["api_key"] != payload["api_key"]
- assert masked["nested"]["authorization"] != payload["nested"]["authorization"]
- assert masked["safe"] == "visible"
- def test_internal_service_auth_middleware_requires_token() -> None:
- asyncio.run(_run_internal_service_auth_smoke())
- async def _run_internal_service_auth_smoke() -> None:
- settings = ServiceSettings(
- service_name="test-service",
- internal_service_auth_required=True,
- internal_service_token="secret-token")
- app = FastAPI()
- add_internal_service_auth(app, settings)
- @app.get("/private")
- async def private() -> dict[str, str]:
- return {"status": "ok"}
- transport = httpx.ASGITransport(app=app)
- async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
- denied_response = await client.get("/private")
- allowed_response = await client.get(
- "/private",
- headers=build_internal_service_headers(settings, source_service="caller"))
- health_response = await client.get("/private/health")
- assert denied_response.status_code == 401
- assert allowed_response.status_code == 200
- assert health_response.status_code == 404
|
