import os from types import SimpleNamespace from fastapi import Depends, HTTPException, Request, status from jose import JWTError, jwt CENTRAL_AUTH_SECRET = os.getenv("CENTRAL_AUTH_SECRET", "") JWT_ALGORITHM = "HS256" APP_SLUG = os.getenv("APP_SLUG", "kds") async def get_current_user(request: Request): """ Verify the stack hnf_session cookie. Returns a SimpleNamespace so archive kds.py can use current_user.kitchen_id, current_user.is_admin, etc. without modification. kitchen_id is pinned to 1. """ token = request.cookies.get("hnf_session") if not token: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated") try: payload = jwt.decode(token, CENTRAL_AUTH_SECRET, algorithms=[JWT_ALGORITHM]) except JWTError: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid session") apps = payload.get("apps", []) if APP_SLUG not in apps: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="No permission for this app") prefix = f"{APP_SLUG}:" raw_caps = payload.get("caps", []) caps = [c[len(prefix):] for c in raw_caps if isinstance(c, str) and c.startswith(prefix)] is_admin = payload.get("is_admin", False) return SimpleNamespace( id=0, email=payload.get("sub", ""), username=payload.get("sub", ""), name=payload.get("name", ""), display_name=payload.get("name", ""), is_admin=is_admin, is_active=True, kitchen_id=1, caps=caps, role="admin" if is_admin else "user", ) def has_cap(user, cap: str) -> bool: return user.is_admin or cap in user.caps def require_cap(cap: str): async def checker(user=Depends(get_current_user)): if not has_cap(user, cap): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing capability: {cap}", ) return user return checker