Spaces:
Running on Zero
Running on Zero
| """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: | |
| def get_token(cls): | |
| return os.environ.get("HF_TOKEN") | |
| 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() | |
| 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 | |