107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""Management authentication: login/logout, session cookies (spec section 8)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
|
|
from sqlalchemy import select
|
|
|
|
from app import audit
|
|
from app.config import get_settings
|
|
from app.db import get_db
|
|
from app.deps import SESSION_COOKIE, CurrentUser
|
|
from app.models import ManagementSession, ManagementUser
|
|
from app.schemas import LoginRequest, PasswordChangeRequest, SessionInfo
|
|
from app.security import (
|
|
generate_session_id,
|
|
hash_password,
|
|
session_expiry,
|
|
verify_password,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
|
|
settings = get_settings()
|
|
|
|
|
|
@router.post("/login", response_model=SessionInfo)
|
|
def login(body: LoginRequest, request: Request, response: Response, db=Depends(get_db)):
|
|
user = db.scalar(
|
|
select(ManagementUser).where(ManagementUser.username == body.username)
|
|
)
|
|
ok = user is not None and user.enabled and verify_password(
|
|
user.password_hash, body.password
|
|
)
|
|
if not ok:
|
|
audit.record(
|
|
db,
|
|
"LOGIN_FAILED",
|
|
request=request,
|
|
user=body.username,
|
|
result="failure",
|
|
)
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid credentials")
|
|
|
|
session = ManagementSession(
|
|
id=generate_session_id(),
|
|
user_id=user.id,
|
|
expires_at=session_expiry(settings.session_ttl_seconds),
|
|
)
|
|
db.add(session)
|
|
audit.record(db, "LOGIN", request=request, user=user.username)
|
|
db.flush()
|
|
|
|
response.set_cookie(
|
|
SESSION_COOKIE,
|
|
str(session.id),
|
|
httponly=True,
|
|
samesite="strict",
|
|
secure=settings.is_production,
|
|
max_age=settings.session_ttl_seconds,
|
|
)
|
|
return SessionInfo(
|
|
username=user.username,
|
|
role=user.role,
|
|
must_change_password=user.must_change_password,
|
|
)
|
|
|
|
|
|
@router.post("/logout")
|
|
def logout(request: Request, response: Response, user: CurrentUser, db=Depends(get_db)):
|
|
token = request.cookies.get(SESSION_COOKIE)
|
|
if token:
|
|
mgmt_session = db.get(ManagementSession, token)
|
|
if mgmt_session is not None:
|
|
mgmt_session.revoked = True
|
|
audit.record(db, "LOGOUT", request=request, user=user.username)
|
|
response.delete_cookie(SESSION_COOKIE)
|
|
return {"status": "logged_out"}
|
|
|
|
|
|
@router.get("/session", response_model=SessionInfo)
|
|
def whoami(user: CurrentUser) -> SessionInfo:
|
|
return SessionInfo(
|
|
username=user.username,
|
|
role=user.role,
|
|
must_change_password=user.must_change_password,
|
|
)
|
|
|
|
|
|
@router.post("/password")
|
|
def change_password(
|
|
body: PasswordChangeRequest,
|
|
request: Request,
|
|
response: Response,
|
|
user: CurrentUser,
|
|
db=Depends(get_db),
|
|
):
|
|
if not verify_password(user.password_hash, body.current_password):
|
|
raise HTTPException(status.HTTP_403_FORBIDDEN, "current password incorrect")
|
|
user.password_hash = hash_password(body.new_password)
|
|
user.must_change_password = False
|
|
# Revoke all other sessions as a precaution after credential change.
|
|
for mgmt_session in db.scalars(
|
|
select(ManagementSession).where(ManagementSession.user_id == user.id)
|
|
):
|
|
mgmt_session.revoked = True
|
|
audit.record(db, "PASSWORD_CHANGED", request=request, user=user.username)
|
|
response.delete_cookie(SESSION_COOKIE)
|
|
return {"status": "password_changed"} |