Spaces:
Sleeping
Sleeping
| """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, full_name: str = "", school: str = "" | |
| ) -> dict: | |
| """Register a new teacher. 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, display_name=full_name, full_name=full_name, school=school | |
| ) | |
| user = await get_user_by_id(user_id) | |
| return user | |
| async def get_current_admin(user: dict = Depends(get_current_user)) -> dict: | |
| """FastAPI dependency: require the authenticated user to be an admin.""" | |
| if not user.get("is_admin"): | |
| raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") | |
| 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": public_user(user)} | |
| def public_user(user: dict) -> dict: | |
| """User fields safe to return to the client (no password hash).""" | |
| return { | |
| "id": user["id"], | |
| "email": user["email"], | |
| "display_name": user.get("display_name", ""), | |
| "full_name": user.get("full_name", ""), | |
| "school": user.get("school", ""), | |
| "is_admin": bool(user.get("is_admin")), | |
| } | |