M8: add operational authentication mode
This commit is contained in:
+12
-3
@@ -8,6 +8,7 @@ from sqlalchemy.orm import Session
|
||||
from app.core.config import get_settings
|
||||
from app.core.db import SessionLocal
|
||||
from app.core.security import SessionPayload, read_session_token
|
||||
from app.models.user import User
|
||||
from app.schemas import CurrentUser, Role
|
||||
|
||||
settings = get_settings()
|
||||
@@ -22,13 +23,21 @@ def get_db() -> Generator[Session, None, None]:
|
||||
db.close()
|
||||
|
||||
|
||||
def get_current_user(request: Request) -> CurrentUser:
|
||||
def get_current_user(request: Request, db: Session = Depends(get_db)) -> CurrentUser:
|
||||
token = request.cookies.get(settings.session_cookie_name)
|
||||
payload: SessionPayload | None = read_session_token(token) if token else None
|
||||
if payload is None or payload.role not in _VALID_ROLES:
|
||||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
|
||||
role: Role = payload.role # type: ignore[assignment]
|
||||
return CurrentUser(public_ref=payload.public_ref, display_name=payload.display_name, role=role)
|
||||
user = db.get(User, payload.user_id)
|
||||
if (
|
||||
user is None
|
||||
or not user.active
|
||||
or user.public_ref != payload.public_ref
|
||||
or user.role not in _VALID_ROLES
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
|
||||
role: Role = user.role # type: ignore[assignment]
|
||||
return CurrentUser(public_ref=user.public_ref, display_name=user.display_name, role=role)
|
||||
|
||||
|
||||
def require_operations_manager(
|
||||
|
||||
Reference in New Issue
Block a user