curo1305andClaude Sonnet 4.6
c38c6b1c01
- revoke_all_refresh_tokens: add skip_token_hash optional param (exclude
current session while revoking others)
- change_password, enable_totp, disable_totp: call revoke with skip hash
derived from refresh cookie; return sessions_revoked in response and
write to audit log metadata_
- 3 new tests: test_{change_password,enable_totp,disable_totp}_revokes_other_sessions
— all PASSED; 373 total passing, 0 regressions
- Frontend toasts: SettingsAccountTab + TotpEnrollment show
"Other sessions have been terminated." when sessions_revoked > 0
- Companion fixes: rate_limiting get_client_ip refactor, deps/auth.py
request.state.current_user, locustfile refresh-token task removal
- Version bump: 0.1.0 → 0.1.1
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
"""
|
|
FastAPI authentication dependency chain for DocuVault.
|
|
|
|
Provides two reusable FastAPI dependencies:
|
|
- get_current_user: validates the Bearer JWT and returns the User ORM object
|
|
- get_current_admin: requires user.role == 'admin' (T-02-07)
|
|
|
|
Usage in route handlers:
|
|
from deps.auth import get_current_user, get_current_admin
|
|
from db.models import User
|
|
|
|
@router.get("/me")
|
|
async def get_me(current_user: User = Depends(get_current_user)):
|
|
return {"id": str(current_user.id), "email": current_user.email}
|
|
|
|
@router.get("/admin/users")
|
|
async def list_users(
|
|
_admin: User = Depends(get_current_admin),
|
|
session: AsyncSession = Depends(get_db),
|
|
):
|
|
...
|
|
"""
|
|
import uuid
|
|
|
|
from fastapi import Depends, HTTPException, Request, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from db.models import User
|
|
from deps.db import get_db
|
|
from services import auth as auth_service
|
|
|
|
# HTTPBearer parses the Authorization: Bearer <token> header.
|
|
# auto_error=True (default) raises 403 if no Authorization header is present.
|
|
security = HTTPBearer()
|
|
|
|
|
|
async def get_current_user(
|
|
request: Request,
|
|
credentials: HTTPAuthorizationCredentials = Depends(security),
|
|
session: AsyncSession = Depends(get_db),
|
|
) -> User:
|
|
"""Validate the Bearer JWT and return the active User ORM object.
|
|
|
|
Raises HTTP 401 if:
|
|
- Token is missing, expired, or tampered (handled by HTTPBearer + decode)
|
|
- User does not exist in the database
|
|
- User account is deactivated (is_active=False)
|
|
"""
|
|
try:
|
|
payload = auth_service.decode_access_token(credentials.credentials)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or expired token",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
) from exc
|
|
|
|
try:
|
|
user_uuid = uuid.UUID(payload["sub"])
|
|
except (KeyError, ValueError) as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid token subject",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
) from exc
|
|
|
|
user = await session.get(User, user_uuid)
|
|
if user is None or not user.is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="User not found or deactivated",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
# Set on request.state so the per-account rate-limiter key_func (_account_key)
|
|
# can read it. The dependency resolves before @account_limiter.limit() calls
|
|
# key_func, so this must live here — not in the handler body (too late).
|
|
request.state.current_user = user
|
|
return user
|
|
|
|
|
|
async def get_current_admin(
|
|
user: User = Depends(get_current_user),
|
|
) -> User:
|
|
"""Require admin role; raises HTTP 403 otherwise (T-02-07).
|
|
|
|
Admin impersonation is architecturally excluded (ADMIN-07, T-02-08):
|
|
no code path sets a JWT sub to a different user's id. This dependency
|
|
only checks that the authenticated user's role is 'admin'.
|
|
"""
|
|
if user.role != "admin":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Admin access required",
|
|
)
|
|
return user
|
|
|
|
|
|
async def get_regular_user(
|
|
user: User = Depends(get_current_user),
|
|
) -> User:
|
|
"""Reject admin accounts on all /api/documents/* endpoints (D-16, SC4).
|
|
|
|
Admin accounts cannot access document content (CLAUDE.md + SEC-04).
|
|
Returns 403 (not 404) — the admin knows document endpoints exist.
|
|
Regular users are passed through unchanged.
|
|
"""
|
|
if user.role == "admin":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Admin accounts cannot access document content",
|
|
)
|
|
return user
|