app.py

#!/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()