7c741062a9
Backend: - src/routers/auth.py NEU: POST /api/auth/magic-link + POST /api/auth/verify - src/auth.py: verify_password/hash_password raus, generate_magic_token rein - src/main.py: alter Login-Endpoint + Brute-Force-Logik raus, neuer auth-Router eingebunden - src/config.py: ALLOWED_EMAIL + PORTAL_MAGIC_LINK_* hinzu - src/models.py: LoginRequest raus, MagicLinkRequest etc. rein - src/email_utils/templates.py: portal_magic_link_email Template Frontend: - src/static/index.html: Email-Eingabe statt Passwort, Token-Verify-Logik fuer ?token= aus URL Datenbank-Migration (migrations/2026-05-09_portal_magic_link.py): - portal_magic_links + portal_magic_link_attempts neu - portal_login_attempts gedroppt - portal_admins.email Spalte hinzu, password_hash geleert Whitelist info@aegis-sight.de, Rate-Limit 5/15 Min, Anti-Enumeration generische Antwort.
66 Zeilen
1.9 KiB
Python
66 Zeilen
1.9 KiB
Python
"""Magic-Link-Authentifizierung für das Verwaltungsportal.
|
|
|
|
JWT für Session, Magic-Link an info@aegis-sight.de zur Anmeldung.
|
|
Passwort-Login wurde mit Migration 2026-05-09 entfernt.
|
|
"""
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
from jose import jwt, JWTError
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from config import JWT_SECRET, JWT_ALGORITHM, JWT_EXPIRE_HOURS
|
|
|
|
security = HTTPBearer()
|
|
|
|
JWT_ISSUER = "aegissight-portal"
|
|
JWT_AUDIENCE = "aegissight-portal"
|
|
|
|
|
|
def generate_magic_token() -> str:
|
|
"""Erzeugt einen URL-sicheren Token (43 Zeichen) für den Magic-Link."""
|
|
return secrets.token_urlsafe(32)
|
|
|
|
|
|
def create_token(admin_id: int, email: str, username: str = "") -> str:
|
|
"""JWT-Session-Token nach erfolgreichem Magic-Link-Verify."""
|
|
now = datetime.now(timezone.utc)
|
|
expire = now + timedelta(hours=JWT_EXPIRE_HOURS)
|
|
payload = {
|
|
"sub": str(admin_id),
|
|
"email": email,
|
|
"username": username or email.split("@")[0],
|
|
"role": "portal_admin",
|
|
"iss": JWT_ISSUER,
|
|
"aud": JWT_AUDIENCE,
|
|
"iat": now,
|
|
"exp": expire,
|
|
}
|
|
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
|
|
|
|
|
|
def decode_token(token: str) -> dict:
|
|
try:
|
|
return jwt.decode(
|
|
token,
|
|
JWT_SECRET,
|
|
algorithms=[JWT_ALGORITHM],
|
|
issuer=JWT_ISSUER,
|
|
audience=JWT_AUDIENCE,
|
|
)
|
|
except JWTError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Token ungültig oder abgelaufen",
|
|
)
|
|
|
|
|
|
async def get_current_admin(
|
|
credentials: HTTPAuthorizationCredentials = Depends(security),
|
|
) -> dict:
|
|
payload = decode_token(credentials.credentials)
|
|
return {
|
|
"id": int(payload["sub"]),
|
|
"email": payload.get("email", ""),
|
|
"username": payload.get("username", ""),
|
|
}
|