7f7133f535
Document issuer, outpost, and header settings for Coolify, fail closed when AUTH_REQUIRED is true, and add harvester healthcheck per Coolify conventions.
72 lines
1.9 KiB
Python
72 lines
1.9 KiB
Python
"""Read Authentik / reverse-proxy identity headers in Streamlit."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Mapping
|
|
|
|
DEFAULT_HEADER_CANDIDATES = (
|
|
"X-Forwarded-User",
|
|
"X-Authentik-Username",
|
|
"X-Authentik-Uid",
|
|
"Remote-User",
|
|
"X-Forwarded-Email",
|
|
)
|
|
|
|
|
|
def _truthy(value: str | None, default: bool = False) -> bool:
|
|
if value is None:
|
|
return default
|
|
return value.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def auth_required() -> bool:
|
|
return _truthy(os.environ.get("AUTH_REQUIRED"), default=True)
|
|
|
|
|
|
def header_candidates() -> tuple[str, ...]:
|
|
primary = os.environ.get("AUTH_USERNAME_HEADER", "").strip()
|
|
extras = [
|
|
os.environ.get("AUTH_UID_HEADER", "").strip(),
|
|
os.environ.get("AUTH_EMAIL_HEADER", "").strip(),
|
|
]
|
|
ordered: list[str] = []
|
|
for name in (primary, *extras, *DEFAULT_HEADER_CANDIDATES):
|
|
if name and name not in ordered:
|
|
ordered.append(name)
|
|
return tuple(ordered)
|
|
|
|
|
|
def _normalize(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
cleaned = value.strip()
|
|
return cleaned or None
|
|
|
|
|
|
def username_from_headers(headers: Mapping[str, str]) -> str | None:
|
|
lowered = {k.lower(): v for k, v in headers.items()}
|
|
for name in header_candidates():
|
|
value = _normalize(lowered.get(name.lower()))
|
|
if value:
|
|
return value
|
|
return None
|
|
|
|
|
|
def get_current_user() -> str | None:
|
|
"""Return the authenticated username from proxy-injected headers."""
|
|
try:
|
|
from streamlit.web.server.websocket_headers import _get_websocket_headers
|
|
|
|
headers = _get_websocket_headers() or {}
|
|
user = username_from_headers(headers)
|
|
if user:
|
|
return user
|
|
except Exception:
|
|
pass
|
|
|
|
if auth_required():
|
|
return None
|
|
|
|
dev_user = os.environ.get("DEV_USER", "").strip()
|
|
return dev_user or "anonymous"
|