Upload_Files_BackUp / gamify_gate.py
AdnaPlays's picture
Update gamify_gate.py
b989503 verified
Raw History Blame Contribute Delete
7.59 kB
"""
gamify_gate.py β€” access gate shared by the three Hugging Face Spaces
(Video Processor, Thumbnail Editor, Upload Center).
admin.gamify.pt/home opens each tool in an IFRAME with https://<space>.hf.space/?__theme=dark&gate=<token>
The token comes from make_gate_token() on the admin side:
base64url(JSON {"u": user, "exp": unix_time, "src": "home"}) + "." + base64url(HMAC-SHA256(secret, payload))
Every request is checked on the server (signature + "exp"). The token is accepted from:
1. the URL (?gate=...) β€” the first page load
2. the page address the browser sends along β€” Gradio's own requests from that page (Referer)
3. a cookie set on the first page load β€” SameSite=None; Secure; Partitioned, so it also
works inside the admin.gamify.pt iframe
Files that are the same for everyone (Gradio's JS/CSS/fonts) are served without a token.
When the token expires, the tool shows "open it again from admin.gamify.pt" and API calls get 401.
Environment (Space settings β†’ Variables and secrets)
GAMIFY_GATE_SECRET (Secret) same value as on admin.gamify.pt β€” required
GAMIFY_LOGIN_URL (Variable) default https://admin.gamify.pt/home
Usage: from gamify_gate import GamifyGate; app.add_middleware(GamifyGate)
"""
import base64
import hashlib
import hmac
import html
import json
import logging
import os
import time
from urllib.parse import parse_qsl, urlsplit
COOKIE_NAME = "gamify_gate"
TOKEN_PARAMS = ("gate", "token")
REQUIRED_SRC = "home"
OPEN_PATHS = ("/healthz", "/favicon.ico", "/manifest.json")
OPEN_PREFIXES = ("/assets/", "/static/", "/gradio_api/static/") # Gradio's own JS/CSS/fonts (public anyway)
def _b64e(raw: bytes) -> str:
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
def _b64d(text: str) -> bytes:
return base64.urlsafe_b64decode((text + "=" * (-len(text) % 4)).encode())
def verify_token(token, secret, now=None):
"""Returns the token's data {"u", "exp", "src"} if it is genuine and not expired, else None."""
if not token or not secret or token.count(".") != 1:
return None
payload, sig = token.split(".")
expected = _b64e(hmac.new(secret.encode(), payload.encode(), hashlib.sha256).digest())
if not hmac.compare_digest(expected, sig):
return None
try:
data = json.loads(_b64d(payload))
except (ValueError, TypeError):
return None
exp = data.get("exp") if isinstance(data, dict) else None
if not isinstance(exp, (int, float)) or exp <= (now if now is not None else time.time()):
return None
if REQUIRED_SRC and data.get("src") != REQUIRED_SRC:
return None
return data
def _token_from_query(query_string):
for k, v in parse_qsl(query_string or "", keep_blank_values=True):
if k in TOKEN_PARAMS:
return v
return None
def _header(scope, name):
for k, v in scope.get("headers") or []:
if k.lower() == name:
return v.decode("latin-1")
return ""
def _cookie_value(scope, name=COOKIE_NAME):
for key, value in scope.get("headers") or []:
if key.lower() != b"cookie":
continue
for part in value.decode("latin-1").split(";"):
k, _, v = part.strip().partition("=")
if k == name:
return v
return None
EXPIRED_PAGE = """<!doctype html><html><head><meta charset="utf-8"><meta name="viewport" content="width=device-width">
<title>Session expired</title></head>
<body style="margin:0;height:100vh;display:flex;align-items:center;justify-content:center;background:#0D0E12;
color:#ECEDF2;font-family:system-ui,sans-serif;text-align:center;padding:20px">
<div><div style="font-size:18px;font-weight:700;margin-bottom:8px">Session expired</div>
<div style="color:#8C8FA0;margin-bottom:16px">Open this tool again from the GAMIFY admin panel.</div>
<a href="{url}" target="_top" style="background:linear-gradient(90deg,#1ED6E0,#5C7CFF);color:#06131A;
font-weight:800;padding:10px 18px;border-radius:10px;text-decoration:none">Go to admin.gamify.pt</a></div></body></html>"""
class GamifyGate:
"""Pure ASGI middleware (doesn't buffer uploads or streams)."""
def __init__(self, app, secret=None, login_url=None):
self.app = app
self.secret = (secret if secret is not None else os.environ.get("GAMIFY_GATE_SECRET", "")).strip()
self.login_url = login_url or os.environ.get("GAMIFY_LOGIN_URL", "https://admin.gamify.pt/home")
if not self.secret:
logging.error("GAMIFY GATE: GAMIFY_GATE_SECRET is not set β€” nobody can open this tool")
def _valid(self, token):
return verify_token(token, self.secret)
async def _respond(self, send, status, body=b"", headers=()):
await send({"type": "http.response.start", "status": status, "headers": list(headers)})
await send({"type": "http.response.body", "body": body})
async def __call__(self, scope, receive, send):
kind = scope.get("type")
path = scope.get("path", "")
if kind not in ("http", "websocket") or path in OPEN_PATHS or path.startswith(OPEN_PREFIXES):
return await self.app(scope, receive, send)
# 1) token in this request's URL (first load of the tool inside the iframe)
url_token = _token_from_query(scope.get("query_string", b"").decode("latin-1"))
data = self._valid(url_token)
if data and kind == "http":
max_age = max(int(data["exp"] - time.time()), 0)
cookie = (f"{COOKIE_NAME}={url_token}; Max-Age={max_age}; Path=/; HttpOnly; "
f"SameSite=None; Secure; Partitioned").encode()
async def send_with_cookie(message):
if message.get("type") == "http.response.start":
message = dict(message)
message["headers"] = list(message.get("headers") or []) + [(b"set-cookie", cookie)]
await send(message)
logging.info(f"GAMIFY GATE: access for {data.get('u')!r} until "
f"{time.strftime('%Y-%m-%d %H:%M UTC', time.gmtime(data['exp']))}")
return await self.app(scope, receive, send_with_cookie)
if data:
return await self.app(scope, receive, send)
# 2) Gradio's own requests from that page carry the page address (with ?gate=...)
referer = _header(scope, b"referer")
if referer and self._valid(_token_from_query(urlsplit(referer).query)):
return await self.app(scope, receive, send)
# 3) the cookie set on the first load
if self._valid(_cookie_value(scope)):
return await self.app(scope, receive, send)
# no valid token: never logged in, tampered, or expired
if kind == "websocket":
await receive()
return await send({"type": "websocket.close", "code": 1008})
wants_page = scope.get("method") == "GET" and "text/html" in _header(scope, b"accept")
if wants_page:
body = EXPIRED_PAGE.replace("{url}", html.escape(self.login_url, quote=True)).encode()
return await self._respond(send, 401, body, [(b"content-type", b"text/html; charset=utf-8"),
(b"cache-control", b"no-store")])
return await self._respond(
send, 401,
b'{"error": "Session expired or missing - open this tool again from admin.gamify.pt"}',
[(b"content-type", b"application/json")])