Spaces:
Running on Zero
Running on Zero
File size: 2,995 Bytes
5f7afef | 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 | """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
|