195 lines
6.4 KiB
Python
195 lines
6.4 KiB
Python
"""Access + auth HTTP API (magic-link login, session cookie)."""
|
|
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
|
|
from sqlmodel import Session
|
|
|
|
from app.models import User
|
|
|
|
from . import mail, service
|
|
from .deps import (
|
|
SESSION_COOKIE,
|
|
SESSION_MAX_AGE,
|
|
get_current_user,
|
|
get_db,
|
|
require_admin,
|
|
)
|
|
from .schemas import (
|
|
AccessDecisionRead,
|
|
AccessRequestCreate,
|
|
AccessRequestRead,
|
|
AdminBootstrapRequest,
|
|
MagicConsumeRequest,
|
|
MagicRequestRead,
|
|
MagicRequestRequest,
|
|
MeRead,
|
|
MessageRead,
|
|
)
|
|
|
|
access_router = APIRouter(prefix="/api/access", tags=["access"])
|
|
auth_router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
|
|
|
|
def _read_request(r) -> AccessRequestRead:
|
|
return AccessRequestRead(
|
|
id=r.id,
|
|
name=r.name,
|
|
email=r.email,
|
|
reason=r.reason,
|
|
status=r.status,
|
|
created_at=r.created_at,
|
|
decided_at=r.decided_at,
|
|
decided_by=r.decided_by,
|
|
)
|
|
|
|
|
|
def _read_me(user: User) -> MeRead:
|
|
return MeRead(
|
|
id=user.id,
|
|
email=user.email,
|
|
name=user.name,
|
|
role=user.role,
|
|
tenant_id=user.tenant_id,
|
|
)
|
|
|
|
|
|
def _set_session_cookie(response: Response, session_id: str) -> None:
|
|
# localhost is a secure context, so Secure is safe for local dev too.
|
|
response.set_cookie(
|
|
SESSION_COOKIE,
|
|
session_id,
|
|
max_age=SESSION_MAX_AGE,
|
|
path="/",
|
|
httponly=True,
|
|
secure=True,
|
|
samesite="lax",
|
|
)
|
|
|
|
|
|
@access_router.post("/requests", response_model=AccessRequestRead, status_code=201)
|
|
def create_request(data: AccessRequestCreate, session: Session = Depends(get_db)):
|
|
try:
|
|
request = service.create_access_request(
|
|
session, data.name, data.email, data.reason
|
|
)
|
|
except service.RateLimitedError as exc:
|
|
raise HTTPException(status_code=429, detail=str(exc)) from exc
|
|
mail.send_access_notification(str(request.id), request.name, request.email)
|
|
return _read_request(request)
|
|
|
|
|
|
@access_router.get("/requests", response_model=list[AccessRequestRead])
|
|
def list_requests(
|
|
status: str | None = Query(default=None),
|
|
session: Session = Depends(get_db),
|
|
admin: User = Depends(require_admin),
|
|
):
|
|
return [_read_request(r) for r in service.list_access_requests(session, status)]
|
|
|
|
|
|
@access_router.post("/requests/{request_id}/approve", response_model=AccessDecisionRead)
|
|
def approve_request(
|
|
request_id: UUID,
|
|
session: Session = Depends(get_db),
|
|
admin: User = Depends(require_admin),
|
|
):
|
|
try:
|
|
request, _user, token = service.approve_access_request(
|
|
session, request_id, admin.email
|
|
)
|
|
except service.NotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
except service.ConflictError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
# The token travels by mail only — never in an API response.
|
|
email_sent = mail.send_magic_link(request.email, mail.magic_link(token))
|
|
return AccessDecisionRead(request=_read_request(request), email_sent=email_sent)
|
|
|
|
|
|
@access_router.post("/requests/{request_id}/reject", response_model=AccessRequestRead)
|
|
def reject_request(
|
|
request_id: UUID,
|
|
session: Session = Depends(get_db),
|
|
admin: User = Depends(require_admin),
|
|
):
|
|
try:
|
|
request = service.reject_access_request(session, request_id, admin.email)
|
|
except service.NotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
except service.ConflictError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
return _read_request(request)
|
|
|
|
|
|
@auth_router.post("/magic/request", response_model=MagicRequestRead)
|
|
def request_magic(
|
|
data: MagicRequestRequest,
|
|
session: Session = Depends(get_db),
|
|
):
|
|
"""Email-first login: allowed emails get a magic link mailed.
|
|
|
|
403 means "not allowed" — the frontend forwards those to /access.
|
|
The token travels by mail only, never in the response.
|
|
"""
|
|
try:
|
|
user, token = service.request_magic_link(session, data.email)
|
|
except service.ForbiddenError as exc:
|
|
raise HTTPException(status_code=403, detail=str(exc)) from exc
|
|
except service.RateLimitedError as exc:
|
|
raise HTTPException(status_code=429, detail=str(exc)) from exc
|
|
email_sent = mail.send_magic_link(user.email, mail.magic_link(token))
|
|
return MagicRequestRead(message="login link sent", email_sent=email_sent)
|
|
|
|
|
|
@auth_router.post("/magic/consume", response_model=MeRead)
|
|
def consume_magic(
|
|
data: MagicConsumeRequest,
|
|
response: Response,
|
|
session: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
user_session = service.consume_magic_token(session, data.token)
|
|
except (
|
|
service.InvalidTokenError,
|
|
service.ExpiredTokenError,
|
|
service.UsedTokenError,
|
|
) as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
user = service.get_user(session, user_session.user_id)
|
|
if user is None: # pragma: no cover - consume already validated the user
|
|
raise HTTPException(status_code=400, detail="unknown token")
|
|
_set_session_cookie(response, user_session.id)
|
|
return _read_me(user)
|
|
|
|
|
|
@auth_router.get("/me", response_model=MeRead)
|
|
def read_me(user: User = Depends(get_current_user)):
|
|
return _read_me(user)
|
|
|
|
|
|
@auth_router.post("/logout", response_model=MessageRead)
|
|
def logout(
|
|
request: Request,
|
|
response: Response,
|
|
session: Session = Depends(get_db),
|
|
):
|
|
# Revoke the server-side row when the cookie maps to a live session;
|
|
# always clear the client cookie.
|
|
session_id = request.cookies.get(SESSION_COOKIE)
|
|
if session_id:
|
|
service.revoke_session(session, session_id)
|
|
response.delete_cookie(SESSION_COOKIE, path="/")
|
|
return MessageRead(message="logged out")
|
|
|
|
|
|
@auth_router.post("/admin-bootstrap", response_model=MessageRead, status_code=201)
|
|
def admin_bootstrap(data: AdminBootstrapRequest, session: Session = Depends(get_db)):
|
|
try:
|
|
_user, token = service.bootstrap_admin(session, data.email)
|
|
except service.ConflictError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
except service.ForbiddenError as exc:
|
|
raise HTTPException(status_code=403, detail=str(exc)) from exc
|
|
mail.send_magic_link(service.normalize_email(data.email), mail.magic_link(token))
|
|
return MessageRead(message="admin created, magic link sent")
|