7d8c226da2
Admin-Plattform für die Tenants von Onyscribe: Editor mit Live-Vorschau (echter Core in Sandbox), Versionen, KI-Agent (Cortecs), OIDC-Abgleich über die Login-Developer-API, Deploy-Engine (Mittwald, gesperrt ohne DEPLOY_LIVE, Freigabe pro Tenant), Core-Vertrag (Schema/Manifest), Core-Releases mit Kompatibilitätsprüfung, Canary-Rollout und Rollback, Zustandsabgleich. Runbook: docs/RUNBOOK_PARALLELTEST.md Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
163 lines
7.1 KiB
Python
163 lines
7.1 KiB
Python
"""OIDC-Login der Admin-Plattform über login.onyva.dev (Authorization-Code + PKCE, wie im Core).
|
||
|
||
Aktiv mit AUTH_ENABLED=true. Env: OIDC_ISSUER, OIDC_CLIENT_ID, OIDC_CLIENT_SECRET, OIDC_REDIRECT_URI,
|
||
OIDC_LOGOUT_REDIRECT_URI, APP_COOKIE_SECURE, sowie die Zugriffsregel (mind. eine):
|
||
ADMIN_ALLOWED_EMAILS=a@x.de,b@y.de und/oder ADMIN_REQUIRED_ROLE=admin (gegen das Claim `app_roles`)
|
||
Ohne Zugriffsregel wird jeder Login abgelehnt (Default-Deny) – die Admin-Plattform hält alle Secrets.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
import re
|
||
import secrets
|
||
import time
|
||
from urllib.parse import urlencode
|
||
|
||
import httpx
|
||
from fastapi import APIRouter, HTTPException, Request
|
||
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
||
|
||
from app.settings import AUTH_ENABLED, SECRET_KEY
|
||
|
||
COOKIE, STATE_COOKIE, STATE_TTL, SESSION_TTL = "oa_session", "oa_oauth", 600, 12 * 3600
|
||
router = APIRouter()
|
||
|
||
PUBLIC = re.compile(r"^/(healthz|auth/.*|static/.*|api/tenants/[^/]+/bundle|api/core/releases)$") # bundle: Token · releases: CI-Token (POST), GET prüft die Session selbst
|
||
DEV_USER = {"sub": "dev", "email": "dev@local", "name": "Dev (Auth aus)", "app_roles": ["admin"]}
|
||
|
||
|
||
def cfg() -> dict:
|
||
g = lambda k, d="": os.getenv(k, d).strip() # noqa: E731
|
||
return {
|
||
"issuer": g("OIDC_ISSUER", "https://login.onyva.dev").rstrip("/"), "client_id": g("OIDC_CLIENT_ID"),
|
||
"client_secret": g("OIDC_CLIENT_SECRET"), "redirect": g("OIDC_REDIRECT_URI", "http://localhost:8300/auth/callback"),
|
||
"logout_redirect": g("OIDC_LOGOUT_REDIRECT_URI", "http://localhost:8300/"),
|
||
"secure": g("APP_COOKIE_SECURE", "false").lower() == "true",
|
||
"emails": {e.strip().lower() for e in g("ADMIN_ALLOWED_EMAILS").split(",") if e.strip()},
|
||
"role": g("ADMIN_REQUIRED_ROLE"),
|
||
}
|
||
|
||
|
||
def _sign(data: dict) -> str:
|
||
payload = base64.urlsafe_b64encode(json.dumps(data, separators=(",", ":")).encode()).decode().rstrip("=")
|
||
return f"{payload}.{hmac.new(SECRET_KEY.encode(), payload.encode(), hashlib.sha256).hexdigest()}"
|
||
|
||
|
||
def _verify(token: str) -> dict | None:
|
||
try:
|
||
payload, _, sig = token.partition(".")
|
||
if not sig or not hmac.compare_digest(sig, hmac.new(SECRET_KEY.encode(), payload.encode(), hashlib.sha256).hexdigest()):
|
||
return None
|
||
data = json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4)))
|
||
return data if int(data.get("exp", 0)) > time.time() else None
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
|
||
|
||
def session_user(request: Request) -> dict | None:
|
||
if not AUTH_ENABLED:
|
||
return DEV_USER
|
||
tok = request.cookies.get(COOKIE)
|
||
return _verify(tok) if tok else None
|
||
|
||
|
||
def is_allowed(user: dict) -> bool:
|
||
c = cfg()
|
||
if not c["emails"] and not c["role"]:
|
||
return False
|
||
if user.get("email", "").lower() in c["emails"]:
|
||
return True
|
||
return bool(c["role"]) and c["role"] in (user.get("app_roles") or [])
|
||
|
||
|
||
async def middleware(request: Request, call_next):
|
||
user = session_user(request)
|
||
request.state.user = user or {}
|
||
if user is None and not PUBLIC.match(request.url.path):
|
||
if request.url.path.startswith(("/api/", "/preview/")):
|
||
return JSONResponse({"detail": "Nicht angemeldet"}, status_code=401)
|
||
return RedirectResponse("/auth/login", status_code=302)
|
||
return await call_next(request)
|
||
|
||
|
||
def actor_name(request: Request) -> str:
|
||
u = getattr(request.state, "user", None) or {}
|
||
return u.get("email") or u.get("name") or "unbekannt"
|
||
|
||
|
||
# ── Routen ──────────────────────────────────────────────────────────────────
|
||
|
||
def exchange_code(code: str, verifier: str) -> dict:
|
||
"""Token-Tausch beim IdP (in Tests ersetzbar)."""
|
||
c = cfg()
|
||
r = httpx.post(f"{c['issuer']}/oauth/token", timeout=20, data={
|
||
"grant_type": "authorization_code", "code": code, "redirect_uri": c["redirect"], "client_id": c["client_id"],
|
||
"client_secret": c["client_secret"], "code_verifier": verifier})
|
||
if r.status_code >= 400:
|
||
raise HTTPException(502, f"Token-Tausch fehlgeschlagen ({r.status_code})")
|
||
return r.json()
|
||
|
||
|
||
def _claims(id_token: str) -> dict:
|
||
try:
|
||
p = id_token.split(".")[1]
|
||
return json.loads(base64.urlsafe_b64decode(p + "=" * (-len(p) % 4)))
|
||
except Exception: # noqa: BLE001
|
||
return {}
|
||
|
||
|
||
@router.get("/auth/login")
|
||
def login():
|
||
c = cfg()
|
||
if not AUTH_ENABLED:
|
||
return RedirectResponse("/", 302)
|
||
if not c["client_id"] or not c["client_secret"]:
|
||
raise HTTPException(503, "OIDC nicht konfiguriert (OIDC_CLIENT_ID / OIDC_CLIENT_SECRET).")
|
||
state, verifier = secrets.token_urlsafe(24), secrets.token_urlsafe(64)
|
||
challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
|
||
q = urlencode({"response_type": "code", "client_id": c["client_id"], "redirect_uri": c["redirect"], "scope": "openid profile email",
|
||
"state": state, "code_challenge": challenge, "code_challenge_method": "S256"})
|
||
resp = RedirectResponse(f"{c['issuer']}/oauth/authorize?{q}", 302)
|
||
resp.set_cookie(STATE_COOKIE, _sign({"state": state, "cv": verifier, "exp": int(time.time()) + STATE_TTL}),
|
||
max_age=STATE_TTL, httponly=True, secure=c["secure"], samesite="lax")
|
||
return resp
|
||
|
||
|
||
@router.get("/auth/callback")
|
||
def callback(request: Request, code: str = "", state: str = "", error: str = ""):
|
||
c = cfg()
|
||
if error:
|
||
raise HTTPException(400, f"Login abgebrochen: {error}")
|
||
pending = _verify(request.cookies.get(STATE_COOKIE, "") or "")
|
||
if not pending or not hmac.compare_digest(str(pending.get("state", "")), state) or not code:
|
||
return RedirectResponse("/auth/login", 303) # abgelaufen/Reload → neu starten
|
||
tokens = exchange_code(code, pending["cv"])
|
||
cl = _claims(tokens.get("id_token", ""))
|
||
user = {"sub": cl.get("sub", ""), "email": cl.get("email", ""), "name": cl.get("name", ""), "app_roles": cl.get("app_roles", [])}
|
||
if not is_allowed(user):
|
||
return HTMLResponse(f"<h2 style='font-family:sans-serif'>Kein Zugriff</h2><p style='font-family:sans-serif'>{user['email'] or 'Dieses Konto'} "
|
||
"ist für die Onyscribe-Admin-Plattform nicht freigeschaltet.</p><p><a href='/auth/logout'>Abmelden</a></p>", status_code=403)
|
||
resp = RedirectResponse("/", 303)
|
||
resp.set_cookie(COOKIE, _sign({**user, "exp": int(time.time()) + SESSION_TTL}), max_age=SESSION_TTL, httponly=True,
|
||
secure=c["secure"], samesite="lax")
|
||
resp.delete_cookie(STATE_COOKIE)
|
||
return resp
|
||
|
||
|
||
@router.get("/auth/logout")
|
||
def logout():
|
||
c = cfg()
|
||
url = f"{c['issuer']}/logout?{urlencode({'post_logout_redirect_uri': c['logout_redirect']})}" if AUTH_ENABLED else "/"
|
||
resp = RedirectResponse(url, 302)
|
||
resp.delete_cookie(COOKIE)
|
||
return resp
|
||
|
||
|
||
@router.get("/auth/me")
|
||
def me(request: Request):
|
||
return {"user": getattr(request.state, "user", {}), "auth_enabled": AUTH_ENABLED}
|