"""Auth — JWT tokens, password hashing, admin guard.""" import os from datetime import datetime, timedelta, timezone import bcrypt import jwt from fastapi import Depends, HTTPException, Request, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy.orm import Session from database import get_db from models import User JWT_SECRET = os.environ.get("VELA_JWT_SECRET", "change-me-in-production") JWT_ALGORITHM = "HS256" JWT_EXPIRE_HOURS = 24 bearer_scheme = HTTPBearer(auto_error=False) def hash_password(password: str) -> str: return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode() def verify_password(password: str, hashed: str) -> bool: return bcrypt.checkpw(password.encode(), hashed.encode()) def create_access_token(user_id: int) -> str: payload = { "sub": str(user_id), "exp": datetime.now(timezone.utc) + timedelta(hours=JWT_EXPIRE_HOURS), "iat": datetime.now(timezone.utc), } return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM) def get_current_user( request: Request, creds: HTTPAuthorizationCredentials | None = Depends(bearer_scheme), db: Session = Depends(get_db), ) -> User: """Try JWT header first, then fall back to cookie.""" token = None if creds: token = creds.credentials elif "vela_token" in request.cookies: token = request.cookies["vela_token"] if not token: raise HTTPException(status_code=401, detail="Not authenticated") try: payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM]) user_id = int(payload["sub"]) except (jwt.ExpiredSignatureError, jwt.InvalidTokenError, KeyError, ValueError): raise HTTPException(status_code=401, detail="Invalid or expired token") user = db.query(User).filter(User.id == user_id, User.active == True).first() if not user: raise HTTPException(status_code=401, detail="User not found") return user def require_admin(user: User = Depends(get_current_user)) -> User: """Currently all users are admins — placeholder for future roles.""" return user