import hashlib import secrets import time import jwt from fastapi import HTTPException, Request, status from jwt import InvalidTokenError SESSION_COOKIE = "chart-session" SESSION_MAX_AGE = 60 * 60 * 24 * 30 def configured_settings(app): runtime = getattr(app.state, "runtime", None) return runtime.settings if runtime else None def configured_token(app) -> str: settings = configured_settings(app) return settings.chart_auth_token if settings else "" def configured_password(app) -> str: settings = configured_settings(app) return settings.chart_password if settings else "" def token_matches(app, presented: str) -> bool: """True when the caller may proceed. An empty CHART_AUTH_TOKEN leaves everything open, which is what local development wants — the check only engages once a token is configured. """ want = configured_token(app) if not want: return not configured_password(app) return secrets.compare_digest((presented or "").encode(), want.encode()) def password_matches(app, presented: str) -> bool: # Falling back to the token avoids locking out a deployment while # CHART_PASSWORD is being added. Once set, only the friendly password logs # a browser in; the opaque token remains valid for direct API clients. want = configured_password(app) or configured_token(app) return bool(want) and secrets.compare_digest( (presented or "").encode(), want.encode() ) def session_secret(app) -> bytes: configured = configured_token(app) or configured_password(app) return hashlib.sha256(configured.encode()).digest() if configured else b"" def create_session(app, now: int | None = None) -> str: secret = session_secret(app) if not secret: return "" issued = now if now is not None else int(time.time()) return jwt.encode( {"sub": "shared", "iat": issued, "exp": issued + SESSION_MAX_AGE}, secret, algorithm="HS256", ) def session_principal(app, presented: str) -> str | None: secret = session_secret(app) if not secret or not presented: return None try: payload = jwt.decode( presented, secret, algorithms=["HS256"], options={"require": ["sub", "iat", "exp"]}, ) except InvalidTokenError: return None principal = payload.get("sub") return principal if isinstance(principal, str) and principal else None def session_matches(app, presented: str) -> bool: return session_principal(app, presented) is not None def require_token(request: Request) -> str: presented = request.headers.get("x-chart-token") or request.query_params.get("token", "") session = request.cookies.get(SESSION_COOKIE, "") if token_matches(request.app, presented): return "api-token" if configured_token(request.app) else "anonymous" principal = session_principal(request.app, session) if principal is not None: return principal raise HTTPException( status.HTTP_401_UNAUTHORIZED, "Authentication required", headers={"WWW-Authenticate": "Session, X-Chart-Token"}, )