Blame for gamify_gate.py

"""
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")])