File size: 4,857 Bytes
1d244c9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | #!/usr/bin/env python3
"""Password gate for the private Phase I AAI dashboard."""
from __future__ import annotations
import base64
import hmac
import os
import time
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
USERNAME = os.environ.get("DASHBOARD_USERNAME", "richard")
PASSWORD = os.environ.get("DASHBOARD_PASSWORD", "")
HF_READ_TOKEN = os.environ.get("HF_PRIVATE_READ_TOKEN", "")
PRIVATE_DASHBOARD_URL = os.environ.get(
"PRIVATE_DASHBOARD_URL",
"https://huggingface.co/spaces/andreaparker/phase1-aai-source-coverage-dashboard/raw/main/index.html",
)
CACHE_TTL_SECONDS = int(os.environ.get("CACHE_TTL_SECONDS", "300"))
_cached_html: bytes | None = None
_cached_at = 0.0
def require_env() -> None:
missing = [
name
for name, value in {
"DASHBOARD_PASSWORD": PASSWORD,
"HF_PRIVATE_READ_TOKEN": HF_READ_TOKEN,
}.items()
if not value
]
if missing:
raise RuntimeError(f"Missing required environment variable(s): {', '.join(missing)}")
def fetch_dashboard() -> bytes:
global _cached_html, _cached_at
now = time.time()
if _cached_html and now - _cached_at < CACHE_TTL_SECONDS:
return _cached_html
req = urllib.request.Request(
PRIVATE_DASHBOARD_URL,
headers={"Authorization": f"Bearer {HF_READ_TOKEN}"},
)
with urllib.request.urlopen(req, timeout=45) as response:
html = response.read()
_cached_html = html
_cached_at = now
return html
def parse_basic_auth(header: str) -> tuple[str, str] | None:
prefix = "Basic "
if not header.startswith(prefix):
return None
try:
decoded = base64.b64decode(header[len(prefix) :], validate=True).decode("utf-8")
except Exception:
return None
username, sep, password = decoded.partition(":")
if not sep:
return None
return username, password
class Handler(BaseHTTPRequestHandler):
server_version = "Phase1DashboardGate/1.0"
def log_message(self, fmt: str, *args: object) -> None:
print(f"{self.address_string()} - {fmt % args}")
def is_authorized(self) -> bool:
parsed = parse_basic_auth(self.headers.get("Authorization", ""))
if not parsed:
return False
username, password = parsed
return hmac.compare_digest(username, USERNAME) and hmac.compare_digest(password, PASSWORD)
def send_login(self) -> None:
body = b"Authentication required.\n"
self.send_response(401)
self.send_header("WWW-Authenticate", 'Basic realm="Phase I AAI Dashboard"')
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self) -> None:
if self.path == "/healthz":
body = b"ok\n"
self.send_response(200)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
return
if self.path not in {"/", "/index.html"}:
self.send_response(302)
self.send_header("Location", "/")
self.end_headers()
return
if not self.is_authorized():
self.send_login()
return
try:
html = fetch_dashboard()
except urllib.error.HTTPError as exc:
body = f"Could not fetch private dashboard: HTTP {exc.code}\n".encode("utf-8")
self.send_response(502)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
return
except Exception as exc:
body = f"Could not fetch private dashboard: {type(exc).__name__}\n".encode("utf-8")
self.send_response(502)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
return
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Cache-Control", "private, max-age=60")
self.send_header("Content-Length", str(len(html)))
self.end_headers()
self.wfile.write(html)
def main() -> None:
require_env()
port = int(os.environ.get("PORT", "7860"))
server = ThreadingHTTPServer(("0.0.0.0", port), Handler)
print(f"Serving password-gated dashboard on port {port}")
server.serve_forever()
if __name__ == "__main__":
main()
|