File size: 2,980 Bytes
1425afc | 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 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | import os
from datetime import datetime, timedelta, timezone
from typing import Optional, Dict, Any
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import HTTPException, status, Depends
from fastapi.security import OAuth2PasswordBearer
# =========================================================
# ENV CONFIG
# =========================================================
SECRET_KEY = os.getenv("SECRET_KEY")
if not SECRET_KEY:
raise RuntimeError("SECRET_KEY environment variable missing")
ALGORITHM = os.getenv("ALGORITHM", "HS256")
ACCESS_TOKEN_EXPIRE_MINUTES = int(
os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "1440")
)
# =========================================================
# PASSWORD HASHING
# =========================================================
# IMPORTANT:
# pbkdf2_sha256 avoids:
# - bcrypt crashes
# - bcrypt native dependency issues
# - 72-byte limits
# - passlib backend bugs
# =========================================================
pwd_context = CryptContext(
schemes=["pbkdf2_sha256"],
deprecated="auto",
)
def hash_password(password: str) -> str:
if not password:
raise ValueError("Password required")
if len(password) < 8:
raise ValueError("Password too short")
return pwd_context.hash(password)
def verify_password(
plain_password: str,
hashed_password: str,
) -> bool:
try:
return pwd_context.verify(
plain_password,
hashed_password,
)
except Exception:
return False
# =========================================================
# JWT
# =========================================================
def create_access_token(
data: Dict[str, Any],
expires_delta: Optional[timedelta] = None,
) -> str:
to_encode = data.copy()
expire = datetime.now(timezone.utc) + (
expires_delta
if expires_delta
else timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
)
to_encode.update({"exp": expire})
return jwt.encode(
to_encode,
SECRET_KEY,
algorithm=ALGORITHM,
)
def decode_token(token: str):
try:
return jwt.decode(
token,
SECRET_KEY,
algorithms=[ALGORITHM],
)
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token",
headers={"WWW-Authenticate": "Bearer"},
)
# =========================================================
# AUTH DEPENDENCY
# =========================================================
oauth2_scheme = OAuth2PasswordBearer(
tokenUrl="/api/auth/login"
)
def get_current_user(
token: str = Depends(oauth2_scheme)
):
payload = decode_token(token)
user_id = payload.get("sub")
if not user_id:
raise HTTPException(
status_code=401,
detail="Invalid authentication",
)
return payload |