Archived
81 lines
2.6 KiB
Python
81 lines
2.6 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import secrets
|
||
|
|
from typing import Annotated
|
||
|
|
|
||
|
|
from fastapi import Depends, Header, HTTPException, Query, Request
|
||
|
|
from fastapi.responses import PlainTextResponse
|
||
|
|
|
||
|
|
from .ddns_dns import DnsProviderError
|
||
|
|
from .ddns_models import DdnsSettings
|
||
|
|
from .ddns_security import credentials_from_basic_auth
|
||
|
|
from .ddns_service import BadIpError, DdnsService
|
||
|
|
from .ddns_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_update(
|
||
|
|
authorization: Annotated[str | None, Header()] = None,
|
||
|
|
username: str | None = Query(default=None),
|
||
|
|
password: str | None = Query(default=None),
|
||
|
|
pass_: str | None = Query(default=None, alias="pass"),
|
||
|
|
passwd: str | None = Query(default=None),
|
||
|
|
) -> tuple[str | None, str | None]:
|
||
|
|
supplied_user = username
|
||
|
|
supplied_token = password or pass_ or passwd
|
||
|
|
if authorization and authorization.lower().startswith("basic "):
|
||
|
|
supplied_user, supplied_token = credentials_from_basic_auth(authorization)
|
||
|
|
return supplied_user, supplied_token
|
||
|
|
|
||
|
|
|
||
|
|
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 = None
|
||
|
|
if authorization and authorization.lower().startswith("basic "):
|
||
|
|
_, supplied = credentials_from_basic_auth(authorization)
|
||
|
|
|
||
|
|
if not supplied 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 apply_update_response(
|
||
|
|
service: DdnsService,
|
||
|
|
request: Request,
|
||
|
|
hostname: str,
|
||
|
|
myip: str | None,
|
||
|
|
ipaddr: str | None,
|
||
|
|
myipv6: str | None,
|
||
|
|
ip6addr: str | None,
|
||
|
|
) -> str | PlainTextResponse:
|
||
|
|
try:
|
||
|
|
return service.apply_update(request, hostname, myip, ipaddr, myipv6, ip6addr).result.dyndns_response
|
||
|
|
except BadIpError:
|
||
|
|
return PlainTextResponse("badip", status_code=400)
|
||
|
|
except DnsProviderError:
|
||
|
|
return PlainTextResponse("911 dns publish failed", status_code=500)
|