b5db15d6ab
Streamlit + VectorBT dashboard, Parquet harvester with nightly cron, Authentik header auth, SQLite strategy persistence, and Bugsink telemetry. Co-authored-by: Cursor <cursoragent@cursor.com>
45 lines
1.1 KiB
Python
45 lines
1.1 KiB
Python
"""Read Authentik / reverse-proxy identity headers in Streamlit."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Mapping
|
|
|
|
HEADER_CANDIDATES = (
|
|
"X-Forwarded-User",
|
|
"X-Authentik-Username",
|
|
"X-Authentik-Uid",
|
|
"Remote-User",
|
|
"X-Forwarded-Email",
|
|
)
|
|
|
|
|
|
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:
|
|
"""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
|
|
|
|
return os.environ.get("DEV_USER", "anonymous")
|