74 lines
2.4 KiB
Python
74 lines
2.4 KiB
Python
|
"""JWT authentication utilities for BadNote."""
|
||
|
|
|
||
|
|
from datetime import datetime, timedelta, timezone
|
||
|
|
from uuid import uuid4
|
||
|
|
|
||
|
|
from fastapi import Depends, HTTPException, status
|
||
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
||
|
|
from jose import JWTError, jwt
|
||
|
|
from passlib.context import CryptContext
|
||
|
|
|
||
|
|
from .config import settings
|
||
|
|
|
||
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||
|
|
bearer_scheme = HTTPBearer()
|
||
|
|
|
||
|
|
|
||
|
|
def hash_password(password: str) -> str:
|
||
|
|
"""Hash a plaintext password with bcrypt."""
|
||
|
|
return pwd_context.hash(password)
|
||
|
|
|
||
|
|
|
||
|
|
def verify_password(password: str, password_hash: str) -> bool:
|
||
|
|
"""Verify a password against its hash."""
|
||
|
|
return pwd_context.verify(password, password_hash)
|
||
|
|
|
||
|
|
|
||
|
|
# A precomputed hash used to spend roughly the same time verifying a password
|
||
|
|
# for a non-existent user as for an existing one, so login response timing does
|
||
|
|
# not leak whether a username exists.
|
||
|
|
_DUMMY_HASH = pwd_context.hash("badnote-dummy-password")
|
||
|
|
|
||
|
|
|
||
|
|
def dummy_verify() -> None:
|
||
|
|
"""Run a throwaway bcrypt verification to equalise login timing."""
|
||
|
|
pwd_context.verify("badnote-dummy-password", _DUMMY_HASH)
|
||
|
|
|
||
|
|
|
||
|
|
def create_access_token(user_id: str) -> str:
|
||
|
|
"""Create a JWT access token for the given user_id."""
|
||
|
|
expire = datetime.now(timezone.utc) + timedelta(hours=settings.jwt_expiry_hours)
|
||
|
|
payload = {
|
||
|
|
"sub": user_id,
|
||
|
|
"exp": expire,
|
||
|
|
"iat": datetime.now(timezone.utc),
|
||
|
|
"jti": str(uuid4()),
|
||
|
|
}
|
||
|
|
return jwt.encode(payload, settings.jwt_secret, algorithm="HS256")
|
||
|
|
|
||
|
|
|
||
|
|
def decode_access_token(token: str) -> dict:
|
||
|
|
"""Decode and validate a JWT token. Returns the payload dict."""
|
||
|
|
try:
|
||
|
|
payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
|
||
|
|
return payload
|
||
|
|
except JWTError as exc:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
|
|
detail="Invalid or expired token",
|
||
|
|
) from exc
|
||
|
|
|
||
|
|
|
||
|
|
async def get_current_user(
|
||
|
|
credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
|
||
|
|
) -> str:
|
||
|
|
"""FastAPI dependency: extract user_id from Bearer token."""
|
||
|
|
payload = decode_access_token(credentials.credentials)
|
||
|
|
user_id: str | None = payload.get("sub")
|
||
|
|
if user_id is None:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
|
|
detail="Token missing subject",
|
||
|
|
)
|
||
|
|
return user_id
|