107 lines
4.0 KiB
Python
107 lines
4.0 KiB
Python
"""Optional local-admin authentication with signed cookies and CSRF checks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import secrets
|
|
import time
|
|
from typing import Any
|
|
|
|
from fastapi import Request, Response
|
|
|
|
from .config import Settings
|
|
from .errors import LumaOpsError
|
|
|
|
SESSION_COOKIE = "lumaops_session"
|
|
CSRF_COOKIE = "lumaops_csrf"
|
|
SESSION_MAX_AGE = 12 * 60 * 60
|
|
|
|
|
|
class AuthManager:
|
|
def __init__(self, settings: Settings) -> None:
|
|
self.settings = settings
|
|
self._key = (settings.admin_token or "auth-disabled").encode("utf-8")
|
|
|
|
def require(self, request: Request) -> str:
|
|
if not self.settings.auth_enabled:
|
|
return "local-admin"
|
|
authorization = request.headers.get("authorization", "")
|
|
if authorization.startswith("Bearer ") and self._valid_token(authorization[7:]):
|
|
return "api-token"
|
|
session = request.cookies.get(SESSION_COOKIE)
|
|
if not session or not self._verify_session(session):
|
|
raise LumaOpsError("authentication_required", "Aanmelding is vereist.", 401)
|
|
if request.method not in {"GET", "HEAD", "OPTIONS"}:
|
|
csrf_cookie = request.cookies.get(CSRF_COOKIE, "")
|
|
csrf_header = request.headers.get("x-csrf-token", "")
|
|
if not csrf_cookie or not hmac.compare_digest(csrf_cookie, csrf_header):
|
|
raise LumaOpsError("csrf_failed", "CSRF-validatie is mislukt.", 403)
|
|
return "local-admin"
|
|
|
|
def login(self, token: str, response: Response) -> dict[str, Any]:
|
|
if not self.settings.auth_enabled:
|
|
return {"authenticated": True, "auth_enabled": False}
|
|
if not self._valid_token(token):
|
|
raise LumaOpsError("invalid_credentials", "Ongeldige beheertoken.", 401)
|
|
csrf = secrets.token_urlsafe(32)
|
|
session = self._create_session()
|
|
response.set_cookie(
|
|
SESSION_COOKIE,
|
|
session,
|
|
max_age=SESSION_MAX_AGE,
|
|
httponly=True,
|
|
secure=self.settings.secure_cookies,
|
|
samesite="strict",
|
|
path="/",
|
|
)
|
|
response.set_cookie(
|
|
CSRF_COOKIE,
|
|
csrf,
|
|
max_age=SESSION_MAX_AGE,
|
|
httponly=False,
|
|
secure=self.settings.secure_cookies,
|
|
samesite="strict",
|
|
path="/",
|
|
)
|
|
return {"authenticated": True, "auth_enabled": True, "csrf_token": csrf}
|
|
|
|
@staticmethod
|
|
def logout(response: Response) -> None:
|
|
response.delete_cookie(SESSION_COOKIE, path="/")
|
|
response.delete_cookie(CSRF_COOKIE, path="/")
|
|
|
|
def _valid_token(self, candidate: str) -> bool:
|
|
configured = self.settings.admin_token
|
|
return configured is not None and hmac.compare_digest(
|
|
candidate.encode("utf-8"), configured.encode("utf-8")
|
|
)
|
|
|
|
def _create_session(self) -> str:
|
|
payload = json.dumps(
|
|
{"sub": "local-admin", "iat": int(time.time()), "nonce": secrets.token_hex(16)},
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
encoded = base64.urlsafe_b64encode(payload).rstrip(b"=")
|
|
signature = hmac.new(self._key, encoded, hashlib.sha256).digest()
|
|
return f"{encoded.decode()}.{base64.urlsafe_b64encode(signature).decode().rstrip('=')}"
|
|
|
|
def _verify_session(self, value: str) -> bool:
|
|
try:
|
|
encoded, signature = value.split(".", 1)
|
|
expected = hmac.new(self._key, encoded.encode("ascii"), hashlib.sha256).digest()
|
|
actual = base64.urlsafe_b64decode(signature + "=" * (-len(signature) % 4))
|
|
if not hmac.compare_digest(expected, actual):
|
|
return False
|
|
payload = json.loads(
|
|
base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)).decode("utf-8")
|
|
)
|
|
return (
|
|
payload.get("sub") == "local-admin"
|
|
and int(payload["iat"]) + SESSION_MAX_AGE > time.time()
|
|
)
|
|
except (ValueError, KeyError, TypeError, json.JSONDecodeError):
|
|
return False
|