Spaces:
Sleeping
Sleeping
File size: 2,207 Bytes
d623240 | 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 | from __future__ import annotations
from datetime import datetime, timedelta, timezone
import hashlib
from typing import Any
import bcrypt
import jwt
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel
from app.core.config import Settings
from app.core.exceptions import AuthenticationError
bearer_scheme = HTTPBearer(auto_error=False)
class TokenPayload(BaseModel):
sub: str
role: str = "user"
exp: int
def hash_password(password: str) -> str:
password_bytes = _password_to_bytes(password)
return bcrypt.hashpw(password_bytes, bcrypt.gensalt()).decode("utf-8")
def verify_password(plain_password: str, hashed_password: str) -> bool:
password_bytes = _password_to_bytes(plain_password)
return bcrypt.checkpw(password_bytes, hashed_password.encode("utf-8"))
def _password_to_bytes(password: str) -> bytes:
raw_bytes = password.encode("utf-8")
if len(raw_bytes) <= 72:
return raw_bytes
return hashlib.sha256(raw_bytes).hexdigest().encode("utf-8")
def create_access_token(
*,
subject: str,
role: str,
settings: Settings,
expires_delta: timedelta | None = None,
) -> str:
expire_at = datetime.now(timezone.utc) + (
expires_delta or timedelta(minutes=settings.access_token_expire_minutes)
)
payload: dict[str, Any] = {
"sub": subject,
"role": role,
"exp": expire_at,
}
return jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)
def decode_access_token(token: str, settings: Settings) -> TokenPayload:
try:
payload = jwt.decode(token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm])
return TokenPayload(**payload)
except jwt.ExpiredSignatureError as exc:
raise AuthenticationError("Access token expired.") from exc
except jwt.PyJWTError as exc:
raise AuthenticationError("Invalid access token.") from exc
def extract_bearer_token(credentials: HTTPAuthorizationCredentials | None) -> str:
if credentials is None or not credentials.credentials:
raise AuthenticationError("Missing bearer token.")
return credentials.credentials
|