Add get_current_user_from_token to auth.py; fix missing import in 3 api files

?token= query-param endpoints (PO/invoice print previews, recipe image exports,
backup download) call get_current_user_from_token which was not ported from the
original archive auth module. Added the function and fixed the missing import in
backup.py, ingredients.py, and invoices.py.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
jtricerolph 2026-07-12 12:34:01 +00:00
parent 907243c2cd
commit ecd63fc65f
4 changed files with 32 additions and 3 deletions

View file

@ -16,7 +16,7 @@ from database import get_db
from models.user import User
from models.settings import KitchenSettings
from models.backup import BackupHistory
from auth import get_current_user, require_cap
from auth import get_current_user, require_cap, get_current_user_from_token
from services.backup_service import BackupService
router = APIRouter()

View file

@ -22,7 +22,7 @@ from models.line_item import LineItem
from models.invoice import Invoice
from models.supplier import Supplier
from models.recipe import Recipe, RecipeIngredient, RecipeSubRecipe
from auth import get_current_user, require_cap
from auth import get_current_user, require_cap, get_current_user_from_token
logger = logging.getLogger(__name__)

View file

@ -19,7 +19,7 @@ from models.invoice import Invoice, InvoiceStatus
from models.line_item import LineItem
from models.product_definition import ProductDefinition
from models.settings import KitchenSettings
from auth import get_current_user, require_cap
from auth import get_current_user, require_cap, get_current_user_from_token
from ocr.extractor import process_invoice_image
from ocr.azure_extractor import parse_pack_size
from services.duplicate_detector import DuplicateDetector

View file

@ -73,6 +73,35 @@ def require_cap(cap: str):
return checker
async def get_current_user_from_token(token: str, db: AsyncSession = None):
"""Decode a JWT passed as a query-param (e.g. ?token=...) and return a user or None."""
try:
payload = jwt.decode(token, CENTRAL_AUTH_SECRET, algorithms=[JWT_ALGORITHM])
except JWTError:
return None
apps = payload.get("apps", [])
if APP_SLUG not in apps:
return None
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)]
return SimpleNamespace(
id=0,
email=payload.get("sub", ""),
username=payload.get("sub", ""),
name=payload.get("name", ""),
display_name=payload.get("name", ""),
is_admin=payload.get("is_admin", False),
is_active=True,
kitchen_id=1,
caps=caps,
role="admin" if payload.get("is_admin", False) else "user",
)
async def get_admin_user(user=Depends(get_current_user)):
if not user.is_admin:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")