Blame for app.py

"""Public Hugging Face ZeroGPU Gateway for PropNode Admin.

- Keeps the real codebase private in GitHub (`Gini-Eddie/propnode-admin`).
- Displays a public decoy Gradio interface at `/` to regular Hugging Face visitors.
- Unlocks the real PropNode Admin portal only via `/gate/<GATEWAY_KEY>` or authorized origins.

ZeroGPU only registers @spaces.GPU functions when the app starts through
`demo.launch()`. The gateway is therefore attached as ASGI middleware on Gradio's
own server (via `app_kwargs`), instead of running a separate Uvicorn server.
"""
import hashlib
import hmac
import logging
import os
import shutil
import subprocess
import sys
from http.cookies import SimpleCookie
from pathlib import Path

import huggingface_hub

# Compatibility shim if huggingface_hub >= 1.0 is present in the base container
if not hasattr(huggingface_hub, "HfFolder"):
    class _HfFolder:
        @classmethod
        def get_token(cls):
            return os.environ.get("HF_TOKEN")

        @classmethod
        def save_token(cls, token: str):
            pass

    huggingface_hub.HfFolder = _HfFolder

import gradio as gr
import spaces
import torch
from starlette.middleware import Middleware
from starlette.responses import RedirectResponse

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("propnode.gateway")

# ---------------------------------------------------------------------------
# 1. ZeroGPU Decoy Interface (shown to public Hugging Face visitors)
# ---------------------------------------------------------------------------
zero = torch.Tensor([0]).cuda()


@spaces.GPU
def greet(n: float) -> str:
    return f"Node telemetry ping: {zero.item() + float(n or 0):.1f} OK"


demo = gr.Interface(
    fn=greet,
    inputs=gr.Number(label="Diagnostic Signal", value=1),
    outputs=gr.Text(label="Gateway Response"),
    title="PropNode Telemetry Gateway",
    description="Internal compute node status check.",
)

# ---------------------------------------------------------------------------
# 2. Private Repository Loader
# ---------------------------------------------------------------------------
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "").strip()
GITHUB_REPO = os.environ.get("GITHUB_REPO", "Gini-Eddie/propnode-admin").strip()
GATEWAY_KEY = os.environ.get("GATEWAY_KEY", "").strip()
RUNTIME_DIR = Path("/tmp/propnode_admin_runtime")
GATE_COOKIE = "__pn_gate"


def _expected_key() -> str:
    return GATEWAY_KEY or os.environ.get("ADMIN_SECRET_KEY", "").strip()


def _gate_token() -> str:
    secret = _expected_key() or "fallback-unconfigured"
    return hashlib.sha256(f"propnode-gate:{secret}".encode()).hexdigest()


def _load_private_admin_app():
    missing = [
        name
        for name in ("GITHUB_TOKEN", "DATABASE_URL", "ADMIN_SECRET_KEY")
        if not os.environ.get(name, "").strip()
    ]
    if missing:
        logger.warning(
            "Gateway running in decoy-only mode. Configure Space Secrets to enable backend: %s",
            ", ".join(missing),
        )
        return None

    if RUNTIME_DIR.exists():
        shutil.rmtree(RUNTIME_DIR)

    repo_url = f"https://x-access-token:{GITHUB_TOKEN}@github.com/{GITHUB_REPO}.git"
    logger.info("Fetching private repository %s into runtime directory...", GITHUB_REPO)
    result = subprocess.run(
        ["git", "clone", "--depth", "1", repo_url, str(RUNTIME_DIR)],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        # stderr may contain the token-bearing URL; do not log it verbatim.
        logger.error("Failed to clone private repository (code %s).", result.returncode)
        return None

    sys.path.insert(0, str(RUNTIME_DIR))
    try:
        from backend.app.main import app as private_app
        logger.info("Private PropNode Admin backend mounted successfully.")
        return private_app
    except Exception as exc:
        logger.error("Failed to import private backend: %s", exc)
        return None


admin_app = _load_private_admin_app()

# ---------------------------------------------------------------------------
# 3. Gateway Middleware (runs in front of Gradio's own server)
# ---------------------------------------------------------------------------
GRADIO_PREFIXES = (
    "/gradio_api",
    "/queue/",
    "/config",
    "/info",
    "/theme.css",
    "/custom.css",
    "/file=",
    "/heartbeat/",
    "/manifest.json",
)


def _is_gradio_route(path: str) -> bool:
    if path.startswith(GRADIO_PREFIXES):
        return True
    if path.startswith("/assets/"):
        # Route `/assets/logo.png` and `/assets/fonts/*` to PropNode Admin;
        # route Gradio's own hashed JS/CSS bundle files in `/assets/` to Gradio.
        if path in {"/assets/logo.png", "/assets/icon-logo.png"} or path.startswith("/assets/fonts/"):
            return False
        return True
    return False


def _headers(scope: dict) -> dict:
    return {k.decode("latin1").lower(): v.decode("latin1") for k, v in scope.get("headers", [])}


def _is_authorized(scope: dict) -> bool:
    if admin_app is None or scope.get("type") != "http":
        return False

    path: str = scope.get("path", "")
    if _is_gradio_route(path):
        return False

    headers = _headers(scope)

    # Never expose the admin app inside the Hugging Face Space iframe
    if headers.get("sec-fetch-dest") == "iframe":
        return False

    # 1. Browser gateway cookie set via /gate/<GATEWAY_KEY> or active admin session cookie
    raw_cookie = headers.get("cookie", "")
    if raw_cookie:
        cookie = SimpleCookie()
        try:
            cookie.load(raw_cookie)
            if GATE_COOKIE in cookie and hmac.compare_digest(cookie[GATE_COOKIE].value, _gate_token()):
                return True
            if path.startswith("/api/") and "propnode_admin_session" in cookie:
                return True
        except Exception:
            pass

    # 2. Explicit X-Gateway-Key header (from a reverse proxy)
    expected = _expected_key()
    header_key = headers.get("x-gateway-key", "")
    if expected and header_key and hmac.compare_digest(header_key, expected):
        return True

    # 3. API call from an explicitly allowed external domain (e.g. https://admin.propnode.app)
    allowed_origins = {
        o.strip().rstrip("/")
        for o in os.environ.get("ADMIN_ALLOWED_ORIGINS", "").split(",")
        if o.strip() and not o.strip().endswith(".hf.space")
    }
    if path.startswith("/api/") and allowed_origins:
        origin = headers.get("origin", "").rstrip("/")
        referer = headers.get("referer", "")
        forwarded_host = headers.get("x-forwarded-host", "").strip()
        if origin and origin in allowed_origins:
            return True
        if any(referer.startswith(o + "/") or referer == o for o in allowed_origins):
            return True
        if forwarded_host and f"https://{forwarded_host}" in allowed_origins:
            return True

    return False


class GatewayMiddleware:
    """Pure ASGI middleware: routes authorized traffic to PropNode Admin, the rest to Gradio."""

    def __init__(self, app):
        self.app = app  # Gradio's ASGI app

    async def __call__(self, scope, receive, send):
        if scope.get("type") == "http" and scope.get("path", "").startswith("/gate/"):
            key = scope["path"][len("/gate/"):].strip("/")
            expected = _expected_key()
            if admin_app is None or not expected or not hmac.compare_digest(key, expected):
                await RedirectResponse(url="/", status_code=302)(scope, receive, send)
                return
            response = RedirectResponse(url="/login.html", status_code=302)
            response.set_cookie(
                GATE_COOKIE,
                _gate_token(),
                httponly=True,
                secure=True,
                samesite="lax",
                path="/",
                max_age=86400 * 30,
            )
            await response(scope, receive, send)
            return

        if _is_authorized(scope):
            await admin_app(scope, receive, send)
            return

        await self.app(scope, receive, send)


# ---------------------------------------------------------------------------
# 4. Launch through Gradio so ZeroGPU detects the @spaces.GPU function
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    demo.launch(
        server_name=os.environ.get("GRADIO_SERVER_NAME", "0.0.0.0"),
        server_port=int(os.environ.get("GRADIO_SERVER_PORT", 7860)),
        ssr_mode=False,
        app_kwargs={"middleware": [Middleware(GatewayMiddleware)]},
    )

# Deployed: 2026-11-08 21:24:00 UTC