B0rbor4d
8305352497
- ldap_auth Service: User bindet als user@domain (kein Servicekonto noetig), Gruppenpruefung ueber Matching-Rule-In-Chain (verschachtelte Gruppen) - Login-Route: lokaler User zuerst, dann LDAP-Fallback; LDAP-User werden lokal gespiegelt (is_ldap, is_admin, Profildaten aus AD) - Neue Config: ldap_domain, ldap_admin_group, ldap_timeout_seconds - ldap3 Dependency, 7 Unit-Tests (ldap3 gemockt), ruff+mypy sauber - Gruppen-DN verified am DC: CN=Domänen-Admins,CN=Users,DC=insight,DC=local
104 lines
3.4 KiB
Python
104 lines
3.4 KiB
Python
"""Auth routes: login (local user or LDAP/AD)."""
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from starlette.concurrency import run_in_threadpool
|
|
|
|
from app.api.deps import client_ip
|
|
from app.core.config import get_settings
|
|
from app.core.database import get_db
|
|
from app.core.exceptions import UnauthorizedError
|
|
from app.core.security import create_access_token, verify_password
|
|
from app.models.user import User
|
|
from app.schemas.auth import LoginRequest, TokenResponse
|
|
from app.services.audit import AuditService
|
|
from app.services.ldap_auth import LdapAuthError
|
|
from app.services.ldap_auth import authenticate as ldap_authenticate
|
|
|
|
settings = get_settings()
|
|
router = APIRouter()
|
|
|
|
|
|
def _issue_token(user: User) -> TokenResponse:
|
|
token = create_access_token(
|
|
subject=user.username,
|
|
extra_claims={"is_admin": user.is_admin},
|
|
)
|
|
return TokenResponse(
|
|
access_token=token,
|
|
expires_in=settings.jwt_access_token_expire_minutes * 60,
|
|
)
|
|
|
|
|
|
@router.post("/login", response_model=TokenResponse)
|
|
async def login(
|
|
payload: LoginRequest,
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> TokenResponse:
|
|
audit = AuditService(db)
|
|
result = await db.execute(select(User).where(User.username == payload.username))
|
|
user = result.scalar_one_or_none()
|
|
|
|
if user and user.password_hash and verify_password(payload.password, user.password_hash):
|
|
user.last_login_at = datetime.now(UTC)
|
|
await audit.log(
|
|
username=user.username,
|
|
action="auth.login",
|
|
ip_address=client_ip(request),
|
|
)
|
|
return _issue_token(user)
|
|
|
|
# Local user not found or password mismatch: fall back to LDAP/AD when
|
|
# enabled. Only members of the configured admin group get access.
|
|
if settings.ldap_enabled:
|
|
try:
|
|
ldap_user = await run_in_threadpool(
|
|
ldap_authenticate, payload.username, payload.password, settings
|
|
)
|
|
except LdapAuthError as exc:
|
|
await audit.log(
|
|
username=payload.username,
|
|
action="auth.login",
|
|
result="failure",
|
|
ip_address=client_ip(request),
|
|
)
|
|
raise UnauthorizedError(exc.detail) from exc
|
|
|
|
result = await db.execute(select(User).where(User.username == ldap_user.username))
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
user = User(
|
|
username=ldap_user.username,
|
|
full_name=ldap_user.full_name,
|
|
email=ldap_user.email,
|
|
is_ldap=True,
|
|
is_admin=ldap_user.is_admin,
|
|
)
|
|
db.add(user)
|
|
else:
|
|
user.is_ldap = True
|
|
user.is_admin = ldap_user.is_admin
|
|
if ldap_user.full_name:
|
|
user.full_name = ldap_user.full_name
|
|
if ldap_user.email:
|
|
user.email = ldap_user.email
|
|
user.last_login_at = datetime.now(UTC)
|
|
await audit.log(
|
|
username=user.username,
|
|
action="auth.login",
|
|
ip_address=client_ip(request),
|
|
)
|
|
return _issue_token(user)
|
|
|
|
await audit.log(
|
|
username=payload.username,
|
|
action="auth.login",
|
|
result="failure",
|
|
ip_address=client_ip(request),
|
|
)
|
|
raise UnauthorizedError("Benutzername oder Passwort falsch")
|