from __future__ import annotations import secrets from typing import Annotated from fastapi import Depends, Header, HTTPException, Request from fastapi.responses import PlainTextResponse from .dns import DnsProviderError from .security import credentials_from_basic_auth from .services.ddns import BadIpError, DdnsService, UpdateIpParams from .settings import DdnsSettings from .store import DdnsStore def get_settings(request: Request) -> DdnsSettings: return request.app.state.settings def get_store(request: Request) -> DdnsStore: return request.app.state.store def get_service(request: Request) -> DdnsService: return request.app.state.service SettingsDep = Annotated[DdnsSettings, Depends(get_settings)] StoreDep = Annotated[DdnsStore, Depends(get_store)] ServiceDep = Annotated[DdnsService, Depends(get_service)] def authenticate_admin( request: Request, authorization: Annotated[str | None, Header()] = None, ) -> None: settings = get_settings(request) if not settings.admin_password: raise HTTPException(status_code=500, detail="DDNS_ADMIN_PASSWORD is required") supplied = _basic_auth_password(authorization) if supplied is None or not secrets.compare_digest(supplied, settings.admin_password): raise HTTPException( status_code=401, detail="admin authentication required", headers={"WWW-Authenticate": 'Basic realm="karldns"'}, ) def _basic_auth_password(authorization: str | None) -> str | None: if authorization is None or not authorization.lower().startswith("basic "): return None _, password = credentials_from_basic_auth(authorization) return password def apply_update_response( service: DdnsService, request: Request, hostname: str, ip_params: UpdateIpParams, ) -> str | PlainTextResponse: try: return service.apply_update(request, hostname, ip_params).result.dyndns_response except BadIpError: return PlainTextResponse("badip", status_code=400) except DnsProviderError: return PlainTextResponse("911 dns publish failed", status_code=500)