Spaces:
Paused
Paused
File size: 3,023 Bytes
bc45e54 cd0918a bc45e54 cd0918a bc45e54 cd0918a bc45e54 | 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 | """Simple email + password JWT authentication for ClassLens."""
from __future__ import annotations
from datetime import datetime, timedelta
from typing import Optional
import bcrypt
from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from .config import get_settings
from .database import get_user_by_email, get_user_by_id, create_user, update_last_login
security = HTTPBearer(auto_error=False)
def hash_password(password: str) -> str:
return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
def verify_password(plain: str, hashed: str) -> bool:
return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8"))
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
settings = get_settings()
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=settings.jwt_expire_minutes))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)
async def get_current_user(
credentials: Optional[HTTPAuthorizationCredentials] = Depends(security),
) -> dict:
"""FastAPI dependency: extract and validate JWT, return user dict."""
if credentials is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
settings = get_settings()
try:
payload = jwt.decode(
credentials.credentials, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm]
)
user_id: int = payload.get("sub")
if user_id is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
except JWTError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
user = await get_user_by_id(int(user_id))
if user is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
return user
async def register_user(email: str, password: str) -> dict:
"""Register a new user. Returns user dict."""
existing = await get_user_by_email(email)
if existing:
raise HTTPException(status_code=400, detail="Email already registered")
hashed = hash_password(password)
user_id = await create_user(email, hashed)
user = await get_user_by_id(user_id)
return user
async def login_user(email: str, password: str) -> dict:
"""Validate credentials and return user dict + JWT token."""
user = await get_user_by_email(email)
if not user or not verify_password(password, user["password_hash"]):
raise HTTPException(status_code=401, detail="Invalid email or password")
await update_last_login(user["id"])
token = create_access_token({"sub": str(user["id"])})
return {"token": token, "user": {"id": user["id"], "email": user["email"], "display_name": user["display_name"]}}
|