File size: 2,570 Bytes
de1e3fc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | """Password hashing + JWT helpers and FastAPI auth dependencies (spec §11).
Auth = JWT access tokens (Bearer), passwords hashed with bcrypt via passlib (DECISIONS.md).
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import bcrypt
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from .config import settings
from .db import get_db
from .models import User
from .models.base import UserRole
# auto_error=False so endpoints can support optional auth (anonymous finders).
_bearer = HTTPBearer(auto_error=False)
# bcrypt operates on at most 72 bytes; truncate consistently for hash + verify.
_BCRYPT_MAX = 72
def hash_password(password: str) -> str:
return bcrypt.hashpw(password.encode("utf-8")[:_BCRYPT_MAX], bcrypt.gensalt()).decode("utf-8")
def verify_password(plain: str, hashed: str) -> bool:
try:
return bcrypt.checkpw(plain.encode("utf-8")[:_BCRYPT_MAX], hashed.encode("utf-8"))
except (ValueError, TypeError):
return False
def create_access_token(user_id: int) -> str:
expire = datetime.now(timezone.utc) + timedelta(
minutes=settings.access_token_ttl_minutes
)
payload = {"sub": str(user_id), "exp": expire}
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
def _decode_user_id(token: str) -> int | None:
try:
payload = jwt.decode(
token, settings.jwt_secret, algorithms=[settings.jwt_algorithm]
)
return int(payload["sub"])
except (jwt.PyJWTError, KeyError, ValueError):
return None
def get_current_user_optional(
creds: HTTPAuthorizationCredentials | None = Depends(_bearer),
db: Session = Depends(get_db),
) -> User | None:
if creds is None:
return None
user_id = _decode_user_id(creds.credentials)
if user_id is None:
return None
return db.get(User, user_id)
def get_current_user(
user: User | None = Depends(get_current_user_optional),
) -> User:
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers={"WWW-Authenticate": "Bearer"},
)
return user
def require_admin(user: User = Depends(get_current_user)) -> User:
if user.role != UserRole.admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, detail="Admin only"
)
return user
|