ScarriffleandClaude Opus 5
31baf4b18c
Zusaetzlicher Anmeldeweg per OIDC neben dem bestehenden Passwort-Login. Ohne gesetztes OIDC_PROVIDERS aendert sich nichts: /api/auth/oidc/providers meldet enabled:false und der Login-Screen bleibt unveraendert. Backend - oidc_config.py: Provider-Registry rein aus ENV. Ein zweiter Provider ist reine Konfiguration (OIDC_PROVIDERS + OIDC_<KEY>_* Block), keine Codeaenderung. - oidc_client.py: Discovery, JWKS, Token-Tausch, ID-Token-Pruefung ueber authlib.jose mit fest gepinnter Signatur-Allowlist (nur asymmetrisch, damit alg:none und Algorithmus-Confusion ausgeschlossen sind). - oidc_identity.py: Zuordnung ueber (provider_key, sub). Ein E-Mail-Treffer auf ein bestehendes Konto wird standardmaessig abgelehnt statt verknuepft — sonst erbt jeder, der im IdP eine beliebige E-Mail setzen kann, das passende Konto. Verknuepfen ist eine explizite, authentifizierte Aktion. Auto-Provisioning nur per ENV-Flag, standardmaessig aus, optional mit Domain-Allowlist. - routers/oidc_router.py: /providers, /start, /callback, /complete, /exchange, /link, /identities. PKCE S256; Flow-State als kurzlebiges signiertes JWT in einem HttpOnly-Cookie mit jti==state statt eines prozesslokalen Dicts, damit der Flow Neustarts und mehrere Worker uebersteht. Das Calendarr-JWT wird ueber ein Einmal-Cookie uebergeben, nie ueber URL oder Fragment. - /exchange als Kontrakt fuer die AppAuth-Clients: die App fuehrt den Flow als Public Client selbst aus und tauscht das ID-Token; kein Secret im Request. - models.py: neue Tabelle oidc_identities, users.auth_source; Migration in main.py im dortigen try/except-Stil. Nebenbefund, mitbehoben: auth.verify_password liess ValueError aus bcrypt.checkpw durch. Ein nicht wohlgeformter password_hash wurde damit zu HTTP 500 auf /api/auth/login und bei jedem CalDAV-Basic-Auth. SSO-Nutzer bekommen einen echten bcrypt-Hash ueber ein zufaelliges, nirgends gespeichertes Geheimnis; fuer CalDAV nutzen sie App-Passwoerter wie 2FA-Nutzer. Web - SSO-Block auf dem Login-Screen, aus /providers befuellt; Buttons sind Links, weil /start eine Top-Level-Navigation braucht. - consumeSsoHandoff() laeuft vor den Setup-/Token-Pruefungen in boot() und raeumt die Query per replaceState ab, bevor es awaited. - 401-Auto-Logout um die SSO-Pfade erweitert, sonst dreht ein fehlgeschlagener Handoff eine Reload-Schleife. - i18n-Keys in de + en (Fallback-Kette deckt die uebrigen acht Sprachen). - Service Worker laesst /api/auth/oidc/ unangetastet durch. 37 neue Tests (Gesamtsuite 60 gruen): Signaturpruefung mit echten RSA-Tokens, alg:none, falscher iss/aud/nonce, unbekannte client_id, Provisioning-Policies, Username-Kollision, Einmaligkeit des Handoff-Cookies, unerreichbarer Provider. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
77 lines
2.6 KiB
Python
77 lines
2.6 KiB
Python
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
from jose import JWTError, jwt
|
|
import bcrypt
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from sqlalchemy.orm import Session
|
|
import os
|
|
|
|
import models
|
|
from database import get_db
|
|
|
|
SECRET_KEY = os.environ.get(
|
|
"SECRET_KEY", "insecure-default-key-change-in-production-use-env-var"
|
|
)
|
|
ALGORITHM = "HS256"
|
|
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7 days
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/token")
|
|
|
|
|
|
def verify_password(plain: str, hashed: str) -> bool:
|
|
# bcrypt.checkpw raises ValueError when `hashed` is not a well-formed hash
|
|
# (empty string, a sentinel like "!", a truncated column). Uncaught, that
|
|
# became an HTTP 500 on /api/auth/login and on every CalDAV Basic-Auth
|
|
# attempt in dav_router. A malformed hash simply means "no password".
|
|
try:
|
|
return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8"))
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
def get_password_hash(password: str) -> str:
|
|
return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
|
|
|
|
|
|
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
|
|
to_encode = data.copy()
|
|
expire = datetime.utcnow() + (
|
|
expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
)
|
|
to_encode["exp"] = expire
|
|
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
|
|
|
|
|
def get_current_user(
|
|
token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)
|
|
) -> models.User:
|
|
exc = HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Could not validate credentials",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
try:
|
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
|
# Other short-lived tokens are signed with the same key (e.g. the OIDC
|
|
# flow token); only a real access token may authenticate a request.
|
|
if payload.get("typ") not in (None, "access"):
|
|
raise exc
|
|
username: str = payload.get("sub")
|
|
if not username:
|
|
raise exc
|
|
except JWTError:
|
|
raise exc
|
|
user = db.query(models.User).filter(models.User.username == username).first()
|
|
if not user:
|
|
raise exc
|
|
return user
|
|
|
|
|
|
def get_current_admin(
|
|
current_user: models.User = Depends(get_current_user),
|
|
) -> models.User:
|
|
if not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Admin access required")
|
|
return current_user
|