File size: 2,494 Bytes
0602b26 40d9233 263e392 b94e45f 5474ace db883b3 d29db6e 0d0f0f9 40d9233 d29db6e 40d9233 892c28e 0602b26 da1a902 0602b26 da1a902 728f32a 0602b26 d29db6e 0602b26 b2005c6 0602b26 db883b3 b2005c6 da1a902 b2005c6 71665ff f987504 b2005c6 f987504 7c98fe6 0602b26 b94e45f 7c98fe6 51c6312 b94e45f 7c98fe6 b94e45f 7c98fe6 b94e45f 7c98fe6 0602b26 263e392 7c98fe6 | 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 | """FastAPI dependencies for the explicit runtime service boundary."""
import secrets
from fastapi import Depends, HTTPException, Request
from loguru import logger
from free_claude_code.application.errors import UnknownProviderError
from free_claude_code.application.ports import ProviderPort, RequestRuntimeLease
from free_claude_code.config.provider_catalog import PROVIDER_CATALOG
from free_claude_code.config.settings import Settings
from .ports import ApiServices
def get_services(request: Request) -> ApiServices:
"""Return the complete services supplied when the app was constructed."""
return request.app.state.services
def get_settings(services: ApiServices = Depends(get_services)) -> Settings:
"""Return the current request-runtime settings snapshot."""
return services.requests.current_settings()
def resolve_provider(
provider_type: str,
*,
lease: RequestRuntimeLease,
) -> ProviderPort:
"""Resolve a provider through one retained generation."""
should_log_init = not lease.is_provider_cached(provider_type)
try:
provider = lease.resolve_provider(provider_type)
except UnknownProviderError:
logger.error(
"Unknown provider_type: '{}'. Supported: {}",
provider_type,
", ".join(f"'{key}'" for key in PROVIDER_CATALOG),
)
raise
if should_log_init:
logger.info("Provider initialized: {}", provider_type)
return provider
def require_proxy_auth(
request: Request,
settings: Settings = Depends(get_settings),
) -> None:
"""Require the configured proxy token as HTTP bearer authorization."""
anthropic_auth_token = settings.anthropic_auth_token.strip()
if not anthropic_auth_token:
return
authorization = request.headers.get("authorization")
if not authorization:
raise HTTPException(
status_code=401,
detail="Missing proxy authentication token",
)
parts = authorization.strip().split(maxsplit=1)
if len(parts) != 2 or parts[0].casefold() != "bearer":
raise HTTPException(
status_code=401,
detail="Invalid proxy authentication token",
)
token = parts[1].strip()
if not token or not secrets.compare_digest(
token.encode("utf-8"),
anthropic_auth_token.encode("utf-8"),
):
raise HTTPException(
status_code=401,
detail="Invalid proxy authentication token",
)
|