karldns.de/karldns/security.py

from __future__ import annotations

import base64
import hashlib
import hmac
import secrets
import time
from datetime import UTC, datetime
from ipaddress import ip_address, ip_network

from fastapi import HTTPException, Request

from .settings import DdnsSettings

DASHBOARD_UNLOCK_COOKIE = "ddns_dashboard_unlock"
DASHBOARD_UNLOCK_DEFAULT_SECONDS = 24 * 60 * 60
DASHBOARD_UNLOCK_STAY_SIGNED_IN_SECONDS = 30 * 24 * 60 * 60


def credentials_from_basic_auth(header: str) -> tuple[str | None, str | None]:
    try:
        raw = base64.b64decode(header.split(" ", 1)[1]).decode("utf-8")
    except Exception:
        return None, None
    if ":" not in raw:
        return None, None
    username, password = raw.split(":", 1)
    return username, password


def hash_secret(token: str) -> str:
    salt = secrets.token_hex(16)
    digest = hashlib.pbkdf2_hmac("sha256", token.encode("utf-8"), salt.encode("ascii"), 200_000)
    return f"pbkdf2_sha256${salt}${digest.hex()}"


def verify_secret(token: str, stored: str) -> bool:
    try:
        algorithm, salt, expected = stored.split("$", 2)
    except ValueError:
        return False
    if algorithm != "pbkdf2_sha256":
        return False
    digest = hashlib.pbkdf2_hmac("sha256", token.encode("utf-8"), salt.encode("ascii"), 200_000)
    return hmac.compare_digest(digest.hex(), expected)


def hash_lookup_token(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def sign_dashboard_unlock_cookie(
    hostname: str,
    management_slug: str,
    password_hash: str,
    signing_secret: str,
    max_age_seconds: int,
    now: int | None = None,
) -> str:
    timestamp = int(time.time() if now is None else now)
    bounded_max_age = _bounded_dashboard_unlock_seconds(max_age_seconds)
    signature = _dashboard_unlock_signature(
        hostname,
        management_slug,
        password_hash,
        signing_secret,
        timestamp,
        bounded_max_age,
    )
    return f"v3:{timestamp}:{bounded_max_age}:{signature}"


def verify_dashboard_unlock_cookie(
    hostname: str,
    management_slug: str,
    password_hash: str | None,
    signing_secret: str,
    cookie_value: str | None,
) -> bool:
    if not password_hash or not signing_secret or not cookie_value:
        return False
    try:
        version, raw_timestamp, raw_max_age, supplied = cookie_value.split(":", 3)
        timestamp = int(raw_timestamp)
        max_age_seconds = _bounded_dashboard_unlock_seconds(int(raw_max_age))
    except ValueError:
        return False
    if version != "v3":
        return False
    now = int(time.time())
    if timestamp > now + 60 or now - timestamp > max_age_seconds:
        return False
    expected = _dashboard_unlock_signature(
        hostname,
        management_slug,
        password_hash,
        signing_secret,
        timestamp,
        max_age_seconds,
    )
    return hmac.compare_digest(supplied, expected)


def _dashboard_unlock_signature(
    hostname: str,
    management_slug: str,
    password_hash: str,
    signing_secret: str,
    timestamp: int,
    max_age_seconds: int,
) -> str:
    payload = f"{hostname}\0{management_slug}\0{password_hash}\0{timestamp}\0{max_age_seconds}".encode()
    return hmac.new(signing_secret.encode(), payload, hashlib.sha256).hexdigest()


def dashboard_cookie_signing_secret(settings: DdnsSettings) -> str:
    return settings.dashboard_cookie_secret or settings.admin_password


def dashboard_csrf_token(management_slug: str, account: dict[str, str | int | None]) -> str:
    hostname = str(account["hostname"])
    password_hash = str(account.get("dashboard_password_hash") or "")
    day = datetime.now(UTC).strftime("%Y-%m-%d")
    key = f"{management_slug}\0{password_hash}".encode()
    return hmac.new(key, f"dashboard:{hostname}:{day}".encode(), hashlib.sha256).hexdigest()


def require_dashboard_csrf(management_slug: str, account: dict[str, str | int | None], supplied: str) -> None:
    if not supplied or not hmac.compare_digest(supplied, dashboard_csrf_token(management_slug, account)):
        raise HTTPException(status_code=403, detail="invalid csrf token")


def _bounded_dashboard_unlock_seconds(max_age_seconds: int) -> int:
    return max(60, min(max_age_seconds, DASHBOARD_UNLOCK_STAY_SIGNED_IN_SECONDS))


def admin_csrf_token(settings: DdnsSettings) -> str:
    day = datetime.now(UTC).strftime("%Y-%m-%d")
    return hmac.new(settings.admin_password.encode("utf-8"), f"admin:{day}".encode(), hashlib.sha256).hexdigest()


def require_admin_csrf(settings: DdnsSettings, supplied: str) -> None:
    if not supplied or not hmac.compare_digest(supplied, admin_csrf_token(settings)):
        raise HTTPException(status_code=403, detail="invalid csrf token")


def client_ip(request: Request, settings: DdnsSettings | None = None) -> str:
    peer = request.client.host if request.client else "unknown"
    if settings and _trusted_proxy(peer, settings.trusted_proxy_ips):
        for header_name in ("cf-connecting-ip", "x-forwarded-for", "x-real-ip"):
            candidate = request.headers.get(header_name, "")
            if candidate:
                return candidate.split(",", 1)[0].strip()
    return peer


def is_rate_limited_path(path: str) -> bool:
    return path == "/" or path.startswith(("/m/", "/api/v1/hostnames", "/api/v1/router/"))


def rate_limit_bucket(path: str) -> str:
    for prefix in ("/m/", "/api/v1/hostnames", "/api/v1/router/"):
        if path.startswith(prefix):
            return prefix.rstrip("/")
    return path


def _trusted_proxy(peer: str, trusted: set[str]) -> bool:
    if not trusted:
        return False
    try:
        peer_ip = ip_address(peer)
    except ValueError:
        return False
    for candidate in trusted:
        try:
            if "/" in candidate and peer_ip in ip_network(candidate, strict=False):
                return True
            if "/" not in candidate and peer_ip == ip_address(candidate):
                return True
        except ValueError:
            continue
    return False