"""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/` 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/ 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