Verificação em outras linguagens

# SPDX-License-Identifier: MIT
# Copyright (c) 2026 uloggd
"""Verifies uloggd Arcade identity tokens and data event tokens.

A copy-ready port of verifyIdentityToken and verifyEventToken from
@uloggd/arcade-sdk/server: the same rules, in the same order, with the same
refusal codes and the same JWKS cache. Python 3.10 or newer, and only the
`cryptography` package (requirements.txt). No JWT library: the format is
fixed, and a general-purpose library would accept more than uloggd ever
signs.

    verifier = ArcadeVerifier("YOUR_APP_ID")
    player = verifier.verify_identity_token(token, nonce=nonce)
    event = verifier.verify_event_token(event_token)

Both raise ArcadeTokenError, whose .code is one of TOKEN_ERROR_CODES. One
verifier per process: it caches uloggd's keys, and it is safe to share
between threads.
"""

from __future__ import annotations

import base64
import http.client
import json
import math
import re
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Callable, Optional

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric.utils import (
    encode_dss_signature,
)

DEFAULT_JWKS_URL = "https://uloggd.com/api/v1/arcade/jwks"
IDENTITY_TOKEN_TYPE = "uloggd-arcade-id+jwt"
EVENT_TOKEN_TYPE = "uloggd-arcade-event+jwt"
DATA_EVENTS = (
    "grant_revoked",
    "grant_forgotten",
    "nickname_changed",
    "account_deleted",
)
TOKEN_ERROR_CODES = (
    "malformed",
    "unsupported_header",
    "unknown_key",
    "bad_signature",
    "wrong_type",
    "wrong_issuer",
    "wrong_audience",
    "expired",
    "not_yet_valid",
    "too_old",
    "nonce_mismatch",
    "mock_refused",
    "test_refused",
    "jwks_unavailable",
)

_MOCK_ISSUER = "mock"
_MOCK_KID_PREFIX = "mock-"
_IDENTITY_MAX_AGE = 300
_EVENT_MAX_AGE = 7 * 24 * 60 * 60
_DEFAULT_SKEW = 60
_MAX_SKEW = 300
_MAX_TOKEN_LENGTH = 8192
_MAX_JWKS_BYTES = 64 * 1024
_MAX_JWKS_KEYS = 20
# However long the JWKS answer says it may be cached.
_JWKS_MAX_AGE = 300.0
# An unknown kid refetches the JWKS at most this often.
_JWKS_REFETCH = 60.0
# After a failed fetch the JWKS is not asked again for this long: a forged
# token reaches the fetch (its header and iss are unsigned), so without a
# pause each one would cost a request while the JWKS is down.
_JWKS_RETRY = 10.0
_FETCH_TIMEOUT = 5.0

# Checked before any decoder runs: base64.b64decode drops characters outside
# the alphabet instead of refusing them.
_BASE64URL = re.compile(r"[A-Za-z0-9_-]*")
_COORDINATE = re.compile(r"[A-Za-z0-9_-]{43}")
_LOOPBACK_IPV4 = re.compile(r"127(?:\.(?:0|[1-9][0-9]{0,2})){3}")
# JavaScript's \s, spelled out: Python's own \s is a different set.
_SPACE = (
    "[\t\n\v\f\r \u00a0\u1680\u2000-\u200a\u2028\u2029\u202f\u205f\u3000"
    "\ufeff]"
)
# The header is lowercased in ASCII first, so no Unicode case folding
# applies, and \Z because $ also matches before a final newline.
_NO_STORE = re.compile(
    "(?:^|,)" + _SPACE + "*(?:no-store|no-cache)" + _SPACE + r"*(?:,|\Z)"
)
_MAX_AGE = re.compile(
    "(?:^|,)" + _SPACE + "*max-age" + _SPACE + "*=" + _SPACE + '*"?([0-9]+)"?'
)
_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)


class ArcadeTokenError(Exception):
    """A refused token. `code` is one of TOKEN_ERROR_CODES."""

    def __init__(self, code: str, message: Optional[str] = None) -> None:
        super().__init__(message or code)
        self.code = code


@dataclass(frozen=True)
class VerifiedIdentity:
    # The player's id for your app alone.
    subject: str
    nickname: str
    scopes: tuple[str, ...]
    instance_id: str
    # "live" or "test".
    mode: str
    # The token's jti: deduplicate by it (and the nonce), never by the
    # token's text.
    id: str
    issued_at: datetime
    expires_at: datetime


@dataclass(frozen=True)
class VerifiedEvent:
    subject: str
    event: str
    # Present only with nickname_changed.
    nickname: Optional[str]
    id: str
    issued_at: datetime


class _KeySet:
    def __init__(self, keys, expires_at: float, attempted_at: float) -> None:
        self.keys = keys
        self.expires_at = expires_at
        self.attempted_at = attempted_at


class _RefuseRedirects(urllib.request.HTTPRedirectHandler):
    # A redirect is a failure: the answer stays a 3xx, never followed.
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def _fail(code: str, message: Optional[str] = None):
    raise ArcadeTokenError(code, message)


def _too_long(token: str) -> bool:
    # The limit counts UTF-16 code units, as JavaScript's length does, so an
    # astral character counts twice. Code points never outnumber them, so a
    # token already over in code points is refused without encoding it.
    return (
        len(token) > _MAX_TOKEN_LENGTH
        or len(token.encode("utf-16-le", "surrogatepass")) // 2 > _MAX_TOKEN_LENGTH
    )


def _decode_base64url(text: str) -> bytes:
    if _BASE64URL.fullmatch(text) is None or len(text) % 4 == 1:
        _fail("malformed", "not base64url")
    return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))


def _refuse_constant(name: str):
    raise ValueError(f"{name} is not JSON")


def _loads(text: str):
    # As JavaScript's JSON.parse: NaN and Infinity are not JSON, and every
    # number is a double, however many digits it has.
    return json.loads(text, parse_int=float, parse_constant=_refuse_constant)


def _decode_json(segment: str) -> dict:
    raw = _decode_base64url(segment)
    try:
        # utf-8-sig drops one leading byte order mark, as TextDecoder does.
        value = _loads(raw.decode("utf-8-sig"))
    except (UnicodeDecodeError, ValueError, RecursionError):
        _fail("malformed", "not JSON")
    if not isinstance(value, dict):
        _fail("malformed", "not a JSON object")
    return value


def _whole(value) -> Optional[float]:
    """A JSON number with a whole value, or None. Booleans are not numbers."""
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return None
    try:
        number = float(value)
    except OverflowError:
        return None
    if not math.isfinite(number) or not number.is_integer():
        return None
    return number


def _utc(seconds: float) -> datetime:
    try:
        return _EPOCH + timedelta(seconds=seconds)
    except OverflowError:
        # Python's datetime ends with the year 9999; uloggd never signs
        # past it (an identity token lives 300 s).
        _fail("malformed", "a time beyond the year 9999")


def _is_loopback(host: str) -> bool:
    if host in ("localhost", "::1") or host.endswith(".localhost"):
        return True
    return _LOOPBACK_IPV4.fullmatch(host) is not None and all(
        int(part) <= 255 for part in host.split(".")
    )


def _trusted_origin(url: str, what: str) -> str:
    """https://uloggd.com, or a loopback origin for tests, as an origin."""
    if not isinstance(url, str) or any(
        ord(char) <= 0x20 or char in "\\\x7f" for char in url
    ):
        raise ValueError(f"{what} must be on https://uloggd.com")
    try:
        parts = urllib.parse.urlsplit(url)
        port = parts.port
    except ValueError:
        raise ValueError(f"{what} must be on https://uloggd.com") from None
    scheme = parts.scheme.lower()
    host = parts.hostname or ""
    if parts.username is not None or parts.password is not None:
        raise ValueError(f"{what} must not carry a user name or password")
    default_port = {"http": 80, "https": 443}.get(scheme)
    if scheme == "https" and host == "uloggd.com" and port in (None, 443):
        return "https://uloggd.com"
    if default_port is not None and _is_loopback(host):
        shown = "[::1]" if host == "::1" else host
        suffix = "" if port in (None, default_port) else f":{port}"
        return f"{scheme}://{shown}{suffix}"
    raise ValueError(f"{what} must be on https://uloggd.com")


def _max_age(cache_control: str) -> float:
    if not cache_control:
        return _JWKS_MAX_AGE
    lowered = cache_control.translate(
        {code: code + 32 for code in range(ord("A"), ord("Z") + 1)}
    )
    if _NO_STORE.search(lowered):
        return 0.0
    match = _MAX_AGE.search(lowered)
    return min(float(match.group(1)), _JWKS_MAX_AGE) if match else _JWKS_MAX_AGE


def _point(raw: bytes):
    try:
        return ec.EllipticCurvePublicKey.from_encoded_point(
            ec.SECP256R1(), raw
        )
    except ValueError:
        return None


def _import_jwk(jwk: dict):
    x, y = jwk.get("x"), jwk.get("y")
    if (
        jwk.get("kty") != "EC"
        or jwk.get("crv") != "P-256"
        or not isinstance(x, str)
        or not isinstance(y, str)
        or _COORDINATE.fullmatch(x) is None
        or _COORDINATE.fullmatch(y) is None
        or ("alg" in jwk and jwk["alg"] != "ES256")
        or ("use" in jwk and jwk["use"] != "sig")
    ):
        return None
    # Only the public coordinates are read, whatever else the entry holds.
    return _point(b"\x04" + _decode_base64url(x) + _decode_base64url(y))


def _mock_key(kid: str):
    # The mock names its public key in the kid: "mock-" and the base64url
    # of the uncompressed point. A kid that is not base64url is malformed,
    # as in Node.
    raw = (
        _decode_base64url(kid[len(_MOCK_KID_PREFIX):])
        if kid.startswith(_MOCK_KID_PREFIX)
        else None
    )
    if raw is None or len(raw) != 65 or raw[0] != 4:
        _fail("unknown_key", "not a mock key")
    key = _point(raw)
    if key is None:
        _fail("unknown_key", "not a mock key")
    return key


class ArcadeVerifier:
    """Verifies uloggd's tokens for one app. Make one, and share it.

    jwks_url and issuer accept only https://uloggd.com, or a loopback
    origin for tests; leave them out. allow_mock accepts tokens from the
    SDK's mock (iss "mock"): local development only. allow_test_mode=False
    refuses tokens from the owner's test mode with test_refused. A bad
    argument raises ValueError here, rather than on each token.
    """

    def __init__(
        self,
        app_id: str,
        *,
        jwks_url: str = DEFAULT_JWKS_URL,
        issuer: Optional[str] = None,
        allow_mock: bool = False,
        allow_test_mode: bool = True,
        clock_skew_seconds: float = _DEFAULT_SKEW,
        clock: Callable[[], float] = time.time,
    ) -> None:
        if not isinstance(app_id, str) or app_id == "":
            raise ValueError("app_id is required")
        if (
            isinstance(clock_skew_seconds, bool)
            or not isinstance(clock_skew_seconds, (int, float))
            or not 0 <= clock_skew_seconds <= _MAX_SKEW
        ):
            raise ValueError(
                f"clock_skew_seconds must be between 0 and {_MAX_SKEW}"
            )
        jwks_origin = _trusted_origin(jwks_url, "jwks_url")
        self._app_id = app_id
        self._jwks_url = jwks_url
        self._issuer = (
            jwks_origin if issuer is None else _trusted_origin(issuer, "issuer")
        )
        self._allow_mock = allow_mock is True
        self._allow_test_mode = allow_test_mode is not False
        self._skew = float(clock_skew_seconds)
        self._clock = clock
        self._opener = urllib.request.build_opener(_RefuseRedirects)
        self._state = threading.Lock()
        self._fetching = threading.Lock()
        self._set: Optional[_KeySet] = None
        self._failed_at: Optional[float] = None
        # Counts finished fetches, so a thread that waited for another's
        # fetch takes its outcome instead of fetching again.
        self._fetches = 0
        self._last_set: Optional[_KeySet] = None
        self._last_error: Optional[str] = None

    def verify_identity_token(self, token, *, nonce) -> VerifiedIdentity:
        """Verifies an identity token from getIdentityToken.

        Accept each nonce once, bind it to the session that asked for it,
        and deduplicate by the returned id.
        """
        claims = self._verify(
            token, IDENTITY_TOKEN_TYPE, self._skew, _IDENTITY_MAX_AGE, True
        )
        scope = claims.get("scope")
        if (
            not isinstance(claims.get("nickname"), str)
            or not isinstance(scope, str)
            or "identify" not in scope.split(" ")
            or not isinstance(claims.get("iid"), str)
            or claims.get("mode") not in ("live", "test")
        ):
            _fail(
                "malformed",
                "nickname, scope with identify, iid and mode are required",
            )
        claimed = claims.get("nonce")
        if (
            not isinstance(nonce, str)
            or nonce == ""
            or not isinstance(claimed, str)
            or claimed != nonce
        ):
            _fail("nonce_mismatch")
        if claims["mode"] == "test" and not self._allow_test_mode:
            _fail("test_refused", "a token from the owner's test mode")
        return VerifiedIdentity(
            subject=claims["sub"],
            nickname=claims["nickname"],
            scopes=tuple(part for part in scope.split(" ") if part),
            instance_id=claims["iid"],
            mode=claims["mode"],
            id=claims["jti"],
            issued_at=_utc(claims["iat"]),
            expires_at=_utc(claims["exp"]),
        )

    def verify_event_token(self, token) -> VerifiedEvent:
        """Verifies a data event token posted as {"event_token": "<jwt>"}.

        Deduplicate by the returned id; act on the event within 15 days.
        """
        claims = self._verify(
            token, EVENT_TOKEN_TYPE, _DEFAULT_SKEW, _EVENT_MAX_AGE, False
        )
        event = claims.get("event")
        if not isinstance(event, str) or event not in DATA_EVENTS:
            _fail("malformed", "unknown event")
        renamed = event == "nickname_changed"
        if (
            not isinstance(claims.get("nickname"), str)
            if renamed
            else "nickname" in claims
        ):
            _fail("malformed", "nickname comes only with nickname_changed")
        return VerifiedEvent(
            subject=claims["sub"],
            event=event,
            nickname=claims["nickname"] if renamed else None,
            id=claims["jti"],
            issued_at=_utc(claims["iat"]),
        )

    def _verify(self, token, typ, skew, max_age, allow_mock) -> dict:
        if not isinstance(token, str) or _too_long(token):
            _fail("malformed", "not a token")
        parts = token.split(".")
        if len(parts) != 3:
            _fail("malformed", "a JWT has three parts")
        header = _decode_json(parts[0])
        if (
            set(header) != {"alg", "kid", "typ"}
            or header["alg"] != "ES256"
            or not isinstance(header["kid"], str)
        ):
            _fail(
                "unsupported_header",
                "the header must be exactly {alg: ES256, typ, kid}",
            )
        if not isinstance(header["typ"], str) or header["typ"] != typ:
            _fail("wrong_type", f"typ must be {typ}")
        claims = _decode_json(parts[1])
        signature = _decode_base64url(parts[2])
        if len(signature) != 64:
            _fail("bad_signature", "the signature must be raw r||s (64 bytes)")

        iss = claims.get("iss")
        if isinstance(iss, str) and iss == _MOCK_ISSUER:
            if not (allow_mock and self._allow_mock):
                _fail("mock_refused", "a token from the SDK's mock")
            key = _mock_key(header["kid"])
        else:
            if not isinstance(iss, str) or iss != self._issuer:
                _fail("wrong_issuer", f"iss must be {self._issuer}")
            key = self._jwks_key(header["kid"])
        der = encode_dss_signature(
            int.from_bytes(signature[:32], "big"),
            int.from_bytes(signature[32:], "big"),
        )
        try:
            key.verify(
                der,
                f"{parts[0]}.{parts[1]}".encode("ascii"),
                ec.ECDSA(hashes.SHA256()),
            )
        except InvalidSignature:
            _fail("bad_signature")

        aud = claims.get("aud")
        if not isinstance(aud, str) or aud != self._app_id:
            _fail("wrong_audience", "aud is not this app")
        iat, exp = _whole(claims.get("iat")), _whole(claims.get("exp"))
        if (
            iat is None
            or exp is None
            or exp <= iat
            or not isinstance(claims.get("sub"), str)
            or not isinstance(claims.get("jti"), str)
        ):
            _fail("malformed", "iat, exp, sub and jti are required")
        now = self._clock()
        if iat > now + skew:
            _fail("not_yet_valid")
        if now >= exp + skew:
            _fail("expired")
        if now - iat > max_age + skew:
            _fail("too_old")
        return {**claims, "iat": iat, "exp": exp}

    def _jwks_key(self, kid: str):
        with self._state:
            current = self._set
            now = self._clock()
            expired = current is None or now >= current.expires_at
            if expired:
                # An expired set is never used, even when the refetch
                # fails: a dropped key must stop verifying.
                self._set = None
                if (
                    self._failed_at is not None
                    and now - self._failed_at < _JWKS_RETRY
                ):
                    _fail("jwks_unavailable", "the JWKS failed moments ago")
        if expired:
            current = self._refresh()
        key = current.keys.get(kid)
        if key is None:
            with self._state:
                now = self._clock()
                # A new kid after a rotation: refetch, but at most once a
                # minute, so made-up kids cannot become a flood of requests.
                refetch = now - current.attempted_at >= _JWKS_REFETCH
                if refetch:
                    current.attempted_at = now
            if refetch:
                current = self._refresh()
                key = current.keys.get(kid)
        if key is None:
            _fail("unknown_key", f"no key {kid} in the JWKS")
        return key

    def _refresh(self) -> _KeySet:
        """One fetch at a time; threads that wait share its outcome."""
        with self._state:
            seen = self._fetches
        with self._fetching:
            with self._state:
                if self._fetches != seen:
                    if self._last_error is not None:
                        _fail("jwks_unavailable", self._last_error)
                    return self._last_set
            try:
                fetched = self._fetch()
            except ArcadeTokenError as error:
                with self._state:
                    self._failed_at = self._clock()
                    self._fetches += 1
                    self._last_error = str(error)
                raise
            with self._state:
                self._failed_at = None
                self._set = fetched
                self._fetches += 1
                self._last_set = fetched
                self._last_error = None
            return fetched

    def _fetch(self) -> _KeySet:
        attempted_at = self._clock()
        request = urllib.request.Request(
            self._jwks_url, headers={"Accept": "application/json"}
        )
        try:
            with self._opener.open(request, timeout=_FETCH_TIMEOUT) as answer:
                status = answer.status
                cache_control = ", ".join(
                    answer.headers.get_all("Cache-Control") or []
                )
                body = answer.read(_MAX_JWKS_BYTES + 1)
        except urllib.error.HTTPError as error:
            error.close()
            _fail("jwks_unavailable", f"the JWKS answered {error.code}")
        except (OSError, ValueError, http.client.HTTPException):
            _fail("jwks_unavailable", "the JWKS could not be fetched")
        if status != 200 or len(body) > _MAX_JWKS_BYTES:
            _fail("jwks_unavailable", f"the JWKS answered {status}")
        try:
            parsed = _loads(body.decode("utf-8-sig", errors="replace"))
        except (ValueError, RecursionError):
            _fail("jwks_unavailable", "the JWKS is not JSON")
        if not isinstance(parsed, dict) or not isinstance(
            parsed.get("keys"), list
        ):
            _fail("jwks_unavailable", "the JWKS has no keys array")
        keys = {}
        for jwk in parsed["keys"][:_MAX_JWKS_KEYS]:
            if (
                not isinstance(jwk, dict)
                or not isinstance(jwk.get("kid"), str)
                or jwk["kid"] in keys
            ):
                continue
            key = _import_jwk(jwk)
            if key is not None:
                keys[jwk["kid"]] = key
        return _KeySet(
            keys, attempted_at + _max_age(cache_control), attempted_at
        )

@app.post("/session")
def session():
    nonce = take_nonce(request)
    body = request.get_json(silent=True)
    token = body.get("token") if isinstance(body, dict) else None
    try:
        player = verifier.verify_identity_token(token, nonce=nonce)
    except ArcadeTokenError as error:
        return jsonify(error=error.code), 401
    # Deduplicate by the token's id, never by its text.
    if not remember_once(player.id):
        return "", 409
    return jsonify(session=open_session(player.subject, player.nickname))

@app.post("/uloggd-events")
def uloggd_events():
    body = request.get_json(silent=True)
    token = body.get("event_token") if isinstance(body, dict) else None
    try:
        event = verifier.verify_event_token(token)
    except ArcadeTokenError:
        return "", 400
    # The same event can arrive twice: the job deduplicates by event.id.
    enqueue(event)
    return "", 204