Spaces:
Running on Zero
Running on Zero
| """Authentication and JWT Token Utilities for CodeStar.""" | |
| from __future__ import annotations | |
| import os | |
| import hmac | |
| import hashlib | |
| import secrets | |
| from datetime import datetime, timezone, timedelta | |
| from typing import Optional, Dict, Any | |
| from fastapi import Header, HTTPException, status | |
| import jwt | |
| SECRET_KEY = os.environ.get("CODESTAR_JWT_SECRET", "codestar-super-secret-key-change-in-production-2026") | |
| ALGORITHM = "HS256" | |
| ACCESS_TOKEN_EXPIRE_DAYS = 30 | |
| def hash_password(password: str) -> tuple[str, str]: | |
| """Hashes a password with a random salt using PBKDF2-HMAC-SHA256.""" | |
| salt = secrets.token_hex(16) | |
| pwd_hash = hashlib.pbkdf2_hmac( | |
| "sha256", password.encode("utf-8"), salt.encode("utf-8"), 100_000 | |
| ).hex() | |
| return pwd_hash, salt | |
| def verify_password(plain_password: str, stored_hash: str, salt: str) -> bool: | |
| """Verifies a plain password against the stored hash and salt.""" | |
| calc_hash = hashlib.pbkdf2_hmac( | |
| "sha256", plain_password.encode("utf-8"), salt.encode("utf-8"), 100_000 | |
| ).hex() | |
| return hmac.compare_digest(calc_hash, stored_hash) | |
| def create_access_token(user_id: int, username: str, expires_delta: Optional[timedelta] = None) -> str: | |
| """Creates a signed JWT access token.""" | |
| expire = datetime.now(timezone.utc) + ( | |
| expires_delta if expires_delta else timedelta(days=ACCESS_TOKEN_EXPIRE_DAYS) | |
| ) | |
| payload = { | |
| "sub": str(user_id), | |
| "username": username, | |
| "exp": expire, | |
| } | |
| return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM) | |
| def decode_access_token(token: str) -> Optional[Dict[str, Any]]: | |
| """Decodes and validates a JWT token.""" | |
| try: | |
| payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) | |
| return payload | |
| except (jwt.PyJWTError, Exception): | |
| return None | |
| def get_current_user(authorization: Optional[str] = Header(None)) -> Optional[Dict[str, Any]]: | |
| """Extracts the logged-in user from Authorization header if present, otherwise returns None.""" | |
| if not authorization: | |
| return None | |
| parts = authorization.split() | |
| if len(parts) != 2 or parts[0].lower() != "bearer": | |
| return None | |
| token = parts[1] | |
| payload = decode_access_token(token) | |
| if not payload or "sub" not in payload: | |
| return None | |
| try: | |
| user_id = int(payload["sub"]) | |
| username = payload.get("username", "") | |
| return {"id": user_id, "username": username} | |
| except (ValueError, TypeError): | |
| return None | |
| def require_current_user(authorization: Optional[str] = Header(None)) -> Dict[str, Any]: | |
| """Requires a valid logged-in user, raising 401 Unauthorized if missing or invalid.""" | |
| user = get_current_user(authorization) | |
| if not user: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Authentication required. Please login first.", | |
| headers={"WWW-Authenticate": "Bearer"}, | |
| ) | |
| return user | |