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