import os import bcrypt from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from pydantic import BaseModel, EmailStr # Configuration - use environment variables in production SECRET_KEY = os.getenv("JWT_SECRET_KEY", "contextflow-secret-key-change-in-production-2024") ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("JWT_EXPIRE_MINUTES", "1440")) # 24 hours default # Pydantic models for request/response class Token(BaseModel): access_token: str token_type: str username: str class TokenData(BaseModel): username: Optional[str] = None class UserCreate(BaseModel): username: str email: EmailStr password: str class UserLogin(BaseModel): username: str password: str class UserResponse(BaseModel): id: int username: str email: str created_at: datetime is_active: int class Config: from_attributes = True def verify_password(plain_password: str, hashed_password: str) -> bool: """Verify a password against its hash""" return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8')) def get_password_hash(password: str) -> str: """Hash a password""" return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: """Create a JWT access token""" to_encode = data.copy() expire = datetime.utcnow() + (expires_delta or 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) -> Optional[str]: """Decode a JWT token and return the username""" try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") return username except JWTError: return None