Public Access
74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
"""Auth dependencies for the MealPlanner backend."""
|
|
|
|
import secrets
|
|
|
|
from fastapi import HTTPException, Request, status
|
|
from itsdangerous import TimestampSigner
|
|
|
|
from app.config import settings
|
|
from app.database import get_db
|
|
from app.models import FamilyProfile
|
|
|
|
bearer_header = "Authorization"
|
|
|
|
SESSION_COOKIE = "mp_session"
|
|
SESSION_MAX_AGE = 60 * 60 * 24 * 30 # 30 days
|
|
|
|
|
|
def require_admin(request: Request) -> str:
|
|
"""Enforce the shared bearer token and return its configured principal."""
|
|
expected = settings.ADMIN_TOKEN
|
|
if not expected:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="Admin auth not configured",
|
|
)
|
|
auth = request.headers.get(bearer_header, "")
|
|
if not auth.startswith("Bearer ") or not secrets.compare_digest(auth[7:], expected):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid admin token",
|
|
)
|
|
return (settings.ADMIN_EMAIL or "").strip() or "admin"
|
|
|
|
|
|
def _signer() -> TimestampSigner:
|
|
return TimestampSigner(settings.SECRET_KEY)
|
|
|
|
|
|
def issue_session(family_profile_id: str) -> str:
|
|
"""Sign the family_profile_id and return the cookie value."""
|
|
return _signer().sign(family_profile_id.encode()).decode()
|
|
|
|
|
|
def require_session(request: Request) -> str:
|
|
"""Require a signed session cookie, with explicit LAN auto-auth opt-in."""
|
|
raw = request.cookies.get(SESSION_COOKIE)
|
|
if raw:
|
|
try:
|
|
return (
|
|
_signer().unsign(raw.encode(), max_age=SESSION_MAX_AGE).decode()
|
|
)
|
|
except Exception:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or expired session",
|
|
)
|
|
|
|
if not settings.TRUSTED_NETWORK_AUTO_AUTH:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Session required",
|
|
)
|
|
|
|
db_gen = get_db()
|
|
db = next(db_gen)
|
|
try:
|
|
profile = db.query(FamilyProfile).first()
|
|
if profile:
|
|
return str(profile.id)
|
|
finally:
|
|
db_gen.close()
|
|
|
|
return "bootstrap"
|