diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..9418255 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,71 @@ +# Changelog + +All notable changes to this project will be documented in this file. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), +and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +## [1.0.0] — 2026-07-15 + +First stable release. The SDK is now aligned with the TypeScript SDK's feature +set (relay analytics, capture-v2 signals, the self-report status endpoint) and +its public surface is considered stable under semantic versioning. + +### Added + +- **Self-report status endpoint.** `handle_request()` answers the platform probe + at `GET /.well-known/supertab/status`. On a valid backend-signed challenge + (an ES256 JWT scoped to the site origin with `purpose: status-probe`) it + returns a `HandlerAction.RESPOND` result with the live SDK config — + `runtime`, `component: {"kind": "python-sdk", version}`, `enforcement`, and + `eventReporting`. Without a valid challenge it returns a minimal + `{"supertab": true}` `404`. The probe short-circuits ahead of token + verification, bot detection, and analytics — no event is emitted. +- **`HandlerAction.RESPOND`** and the `RespondHandlerResult` type — a fully + formed response the caller serves verbatim without contacting origin. +- **Relay analytics** (off by default; `analytics_enabled=True`). One event per + request to `/ingest/events`, authenticated with the merchant `api_key`; the + backend derives merchant identity, so no merchant identifier is sent. Emits + `schema_version: 2` capture-v2 spoof-detection signals. Fire-and-forget and + fail-open — analytics can never block, slow, or alter request handling. +- **`analytics_base_url` config option, plus `SupertabConnect.set_analytics_base_url()` / + `get_analytics_base_url()`.** Points the analytics ingest relay at a specific + host, independent of `set_base_url` (which stays the base for token + acquisition / JWKS / verification). Mirrors the existing `base_url` pattern. +- **`HandleRequestContext`** for passing CDN-supplied per-request signals + (`source_cdn`, `client_ip`, `request_id`, `request_country`, `request_asn`, + `tls_fingerprint`, `cdn_signals`) onto the analytics event. +- **`User-Agent` header** (`supertab-connect-sdk-python/`) on all + outbound calls to the Connect backend. + +### Changed + +- **Analytics defaults to the dedicated ingest service + (`https://ingest-connect.supertab.co`)** rather than the API host. Only + affects deployments with `analytics_enabled=True`; the `/ingest/events` path + and payload are unchanged — traffic just moves to the standalone service. + Non-prod / local setups should call `set_analytics_base_url()` to avoid + emitting to prod. +- **`EnforcementMode` values renamed** to match the backend and the TS SDK: + `SOFT` → `OBSERVE`, `STRICT` → `ENFORCE` (enum members and string values). + The default was already observe-only — a rename, not a behavior change. + +### Fixed + +- **Throttled JWKS refresh on key-rotation misses.** When a token's `kid` is absent + from the cached key set, the SDK refreshes the platform JWKS at most once per minute + per base URL (single-flight, per-base-URL) instead of clearing the whole cache and + refetching on every miss. Because `kid` is read from an unverified JWT header, the + previous behavior let an unauthenticated caller — e.g. via the public + `/.well-known/supertab/status` endpoint — submit tokens with rotating unknown `kid`s + to bypass the 48h cache and force a backend fetch per request (and evict cached keys + used by license verification). Genuine rotations still recover on the first miss; + signature verification was never bypassed. Applies to both the status-probe and + license-verification paths. +- **Analytics emits are drained on close.** The HTTP analytics transport now owns + its own client and in-flight emit tasks; `await client.aclose()` (or exiting an + `async with` block) flushes outstanding emits within a bounded timeout before + closing the client. Previously a fire-and-forget emit scheduled just before + `aclose()` could run afterwards, lazily recreate a fresh HTTP client that was + never closed (a leak), and race process shutdown. Emission remains fire-and-forget + and fail-open on the request path. diff --git a/README.md b/README.md index 2163465..17ad768 100644 --- a/README.md +++ b/README.md @@ -85,8 +85,146 @@ asyncio.run(main()) ``` For request-level enforcement, use `SupertabConnect.handle_request()` with an -`httpx.Request`. See the `examples` directory for complete merchant and customer -examples. +`httpx.Request`. It extracts the license token from the `Authorization` header, +verifies it, optionally emits a relay analytics event, and applies bot detection +and enforcement mode when no token is present. It returns either +`{"action": HandlerAction.ALLOW, ...}`, +`{"action": HandlerAction.BLOCK, "status": ..., "body": ..., "headers": ...}`, or +`{"action": HandlerAction.RESPOND, "status": ..., "body": ..., "headers": ...}` +(see [Self-report status endpoint](#self-report-status-endpoint) below). + +`handle_request()` accepts an optional second argument, a `HandleRequestContext`, +which carries per-request signals supplied by an upstream CDN/proxy +(`source_cdn`, `client_ip`, `request_id`, `request_country`, `request_asn`, +`tls_fingerprint`, and `cdn_signals`). These are recorded on the analytics event +when present; for direct SDK use the context can be omitted. + +`cdn_signals` is a `CdnRequestSignals` object carrying the richer +spoof-detection signals that cannot be read from the portable request — TLS +fingerprinting fields, the verified-bot category, the negotiated protocol, and +so on. These are platform-specific (for example, Cloudflare exposes them on +`request.cf`), so the SDK takes them from the caller rather than extracting them +itself. Everything left unset stays `null` on the event. + +See the `examples` directory for complete merchant and customer examples. + +## Self-report status endpoint + +`handle_request()` also answers the platform's self-report probe at +`GET /.well-known/supertab/status`, which powers the portal's live-health view. +When the request carries a valid backend-signed challenge +(`Authorization: Bearer `, an ES256 JWT scoped to the site origin with +`purpose: status-probe`), the SDK returns a `RESPOND` result reporting its live +config: + +```json +{ + "runtime": "cloudflare", + "component": { "kind": "python-sdk", "version": "1.0.0" }, + "enforcement": "observe", + "eventReporting": false +} +``` + +`runtime` comes from `HandleRequestContext.source_cdn` (or `null` for direct +invocation). Without a valid challenge the SDK returns a minimal +`{"supertab": true}` with a `404` status, disclosing nothing about the +deployment. The probe short-circuits ahead of token verification, bot detection, +and analytics — no event is emitted. Both responses set `Cache-Control: +no-store`. + +A `RESPOND` result must be served to the caller **verbatim** (status, body, and +headers) without forwarding to origin; it is distinguished from `ALLOW` / `BLOCK` +by `result["action"] == HandlerAction.RESPOND`. + +## Analytics + +The SDK can emit one analytics event per request to the Supertab Connect +**relay** endpoint at `/ingest/events`, served by the dedicated ingest service +(`https://ingest-connect.supertab.co`) — separate from the API host used for +token acquisition / JWKS / verification. This is **off by default** — enable it +by passing `analytics_enabled=True`: + +```python +from supertab_connect import SupertabConnect, SupertabConnectConfig + +client = SupertabConnect( + SupertabConnectConfig( + api_key="stc_live_your_api_key", + analytics_enabled=True, + ) +) +``` + +**No extra credentials are required.** Analytics requests are authenticated with +your configured merchant `api_key` using `Authorization: Bearer `. The +backend derives merchant identity from the API key, so the SDK sends **no +merchant identifier** in the analytics payload. + +Each `AnalyticsEvent` captures the request id, source CDN, a normalized client +IP, the request path (with percent-encoding preserved), method, and selected +headers — plus, when an upstream CDN exposes them via `HandleRequestContext`, the +request country, ASN, TLS fingerprint, and HTTP Message Signature headers — along +with the verification/enforcement decision for the request. + +Events emit at **`schema_version: 2`** ("capture v2"), which adds raw +spoof-detection signals for query-time classification in the warehouse (the SDK +never classifies — it emits raw signals only): + +- **Portable header signals**, read directly from the request: `sec_fetch_*`, + the `sec_ch_ua*` client hints, `accept`, `host`, `has_cookies`, and + `header_names` — the lowercased, deduped, sorted set of request-header names + with edge-injected headers (`cf-*`, `fastly-*`, `cloudfront-*`, + `x-forwarded-*`, `x-real-ip`, the synthesized `Host`, …) stripped so it + reflects only what the client sent. +- **Query-string derived signals**: `query_length`, `query_param_count`, and + `query_suspicious` (a coarse exploit-marker heuristic). The raw query string + is **never** stored. +- **CDN plumbing** supplied via `HandleRequestContext.cdn_signals`: + `accept_encoding`, `http_protocol`, `tls_version`, `tls_cipher`, + `tls_client_hello_length`, `tls_client_extensions_sha1`, `as_organization`, + `client_tcp_rtt`, `cdn_verified_bot_category`, `request_priority`, and + `tls_fingerprint_ja4`. + +`accept`, `sec_ch_ua`, and `as_organization` are truncated to 512 characters. +Every capture-v2 field is fail-open: anything unavailable is emitted as `null`. + +**Fail-open:** analytics emission is fire-and-forget and can never block, slow, +or alter request handling. If emission fails, the error is swallowed and the +request proceeds exactly as it would with analytics disabled. Analytics is sent +only to the relay at `/ingest/events`, independent of billing event recording. + +Because emission is fire-and-forget, close the client when you are done so +in-flight events are not lost: `await client.aclose()` (or an `async with +SupertabConnect(...) as client:` block) drains outstanding emits — bounded by a +short timeout — and releases the underlying HTTP client. The drain never blocks +request handling; it only applies at teardown. + +Point analytics at another environment with the `analytics_base_url` config +option (or `SupertabConnect.set_analytics_base_url(...)`) — precedence is +per-instance `analytics_base_url` > `set_analytics_base_url()` > the ingest +default. This is independent of `supertab_base_url` / `set_base_url(...)`, which +control the API host for token acquisition / JWKS / verification; changing that +no longer moves analytics traffic. + +For advanced use, the `AnalyticsTransport` protocol lets you inject a custom +transport (for example, an in-memory recorder in tests) via the internal +`analytics_transport` config field; `AnalyticsEvent` and `HandleRequestContext` +are exported from the package root. + +### Native Fastly logging (not applicable to the Python SDK) + +The TypeScript SDK can deliver analytics through a **native Fastly Compute +logging endpoint** (`FastlyLogTransport` / the `logEndpoint` option on +`fastlyHandleRequests`) instead of the HTTP relay, letting Fastly ship events +off-path to S3. That path is intentionally **not ported here**: Python does not +run on Fastly Compute (the `fastly:logger` built-in has no Python equivalent), +and — consistent with this SDK's design — the Python SDK does not embed CDN edge +handlers, receiving CDN-derived signals through `HandleRequestContext` instead. + +If you need to deliver analytics somewhere other than the relay (for example, to +a log shipper that forwards to S3/Tinybird), implement the `AnalyticsTransport` +protocol and pass it via the `analytics_transport` config field. ## Error Handling diff --git a/examples/merchant_handle_request.py b/examples/merchant_handle_request.py index f3708de..e9e3c66 100644 --- a/examples/merchant_handle_request.py +++ b/examples/merchant_handle_request.py @@ -16,7 +16,7 @@ async def main() -> None: client = SupertabConnect( SupertabConnectConfig( api_key="your_api_key", - enforcement=EnforcementMode.STRICT, + enforcement=EnforcementMode.ENFORCE, debug=True, ) ) @@ -42,6 +42,16 @@ async def main() -> None: print(result["headers"]["WWW-Authenticate"]) return + # A RESPOND result (e.g. the self-report status probe at /.well-known/supertab/status) + # must be served to the caller verbatim — status, body, and headers — and never forwarded + # to origin. Treating it as ALLOW would leak the probe through to the application. + if result["action"] is HandlerAction.RESPOND: + print("RESPOND request") + print(result["status"]) # type: ignore + print(result["headers"]) # type: ignore + print(result["body"]) # type: ignore + return + print("ALLOW request") if "headers" in result: print(result["headers"]) diff --git a/examples/merchant_verify_and_record_event.py b/examples/merchant_verify_and_record_event.py index 2d727a2..7dc91c4 100644 --- a/examples/merchant_verify_and_record_event.py +++ b/examples/merchant_verify_and_record_event.py @@ -14,7 +14,7 @@ async def main() -> None: client = SupertabConnect( SupertabConnectConfig( api_key="your_api_key", - enforcement=EnforcementMode.SOFT, + enforcement=EnforcementMode.OBSERVE, debug=True, ) ) diff --git a/pyproject.toml b/pyproject.toml index 6f40ea6..9a8e0cc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "supertab-connect-sdk" -version = "0.1.2" +version = "1.0.0" authors = [ { name = "Supertab", email = "hello@supertab.co" }, ] @@ -12,7 +12,7 @@ description = "Supertab Connect SDK" readme = "README.md" requires-python = ">=3.12" classifiers = [ - "Development Status :: 4 - Beta", + "Development Status :: 5 - Production/Stable", "Intended Audience :: Developers", "Operating System :: OS Independent", "Programming Language :: Python :: 3", diff --git a/supertab_connect/__init__.py b/supertab_connect/__init__.py index 6c12aca..5b7c670 100644 --- a/supertab_connect/__init__.py +++ b/supertab_connect/__init__.py @@ -1,5 +1,10 @@ """Supertab Connect SDK.""" +from supertab_connect.analytics.types import ( + AnalyticsEvent, + AnalyticsTransport, + CdnRequestSignals, +) from supertab_connect.customer.token import obtain_license_token from supertab_connect.exceptions import SupertabConnectError from supertab_connect.merchant.bots import default_bot_detector @@ -7,6 +12,7 @@ from supertab_connect.merchant.license import verify_license_token from supertab_connect.types import ( EnforcementMode, + HandleRequestContext, HandlerAction, HandlerResult, RSLVerificationResult, @@ -15,7 +21,11 @@ ) __all__ = [ + "AnalyticsEvent", + "AnalyticsTransport", + "CdnRequestSignals", "EnforcementMode", + "HandleRequestContext", "HandlerAction", "HandlerResult", "RSLVerificationResult", diff --git a/supertab_connect/_version.py b/supertab_connect/_version.py index cde63cb..7399e87 100644 --- a/supertab_connect/_version.py +++ b/supertab_connect/_version.py @@ -8,10 +8,13 @@ @lru_cache(maxsize=1) -def _get_sdk_user_agent() -> str: +def _get_sdk_version() -> str: try: - package_version = version(_PACKAGE_NAME) + return version(_PACKAGE_NAME) except PackageNotFoundError: - package_version = "unknown" + return "unknown" + - return f"{_SDK_NAME}/{package_version}" +@lru_cache(maxsize=1) +def _get_sdk_user_agent() -> str: + return f"{_SDK_NAME}/{_get_sdk_version()}" diff --git a/supertab_connect/analytics/__init__.py b/supertab_connect/analytics/__init__.py new file mode 100644 index 0000000..9ae09ad --- /dev/null +++ b/supertab_connect/analytics/__init__.py @@ -0,0 +1,41 @@ +"""Relay analytics for Supertab Connect (mirrors the TS SDK `analytics/` module).""" + +from supertab_connect.analytics.build_analytics_event import ( + BuildAnalyticsEventContext, + build_analytics_event, +) +from supertab_connect.analytics.ip import normalize_client_ip +from supertab_connect.analytics.transport import ( + ANALYTICS_EVENTS_PATH, + HttpAnalyticsTransport, + NoopAnalyticsTransport, +) +from supertab_connect.analytics.types import ( + SCHEMA_VERSION, + TOKEN_OUTCOME_BY_REASON, + AnalyticsEvent, + AnalyticsTransport, + CdnRequestSignals, + Decision, + FinalAction, + SourceCdn, + TokenOutcome, +) + +__all__ = [ + "ANALYTICS_EVENTS_PATH", + "SCHEMA_VERSION", + "TOKEN_OUTCOME_BY_REASON", + "AnalyticsEvent", + "AnalyticsTransport", + "BuildAnalyticsEventContext", + "CdnRequestSignals", + "Decision", + "FinalAction", + "HttpAnalyticsTransport", + "NoopAnalyticsTransport", + "SourceCdn", + "TokenOutcome", + "build_analytics_event", + "normalize_client_ip", +] diff --git a/supertab_connect/analytics/build_analytics_event.py b/supertab_connect/analytics/build_analytics_event.py new file mode 100644 index 0000000..36da063 --- /dev/null +++ b/supertab_connect/analytics/build_analytics_event.py @@ -0,0 +1,187 @@ +"""Build a relay AnalyticsEvent from a request + decision (mirrors TS `buildAnalyticsEvent.ts`).""" + +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from urllib.parse import unquote + +from httpx import Request + +from supertab_connect.analytics.ip import normalize_client_ip +from supertab_connect.analytics.types import ( + SCHEMA_VERSION, + AnalyticsEvent, + CdnRequestSignals, + Decision, + EnforcementWire, + SourceCdn, +) +from supertab_connect.types import EnforcementMode + +# Defensive cap on client-controlled free-form strings, applied at the edge (mirrored by the relay). +MAX_FIELD_LENGTH = 512 + +# Edge-injected headers are CDN artifacts, not client signals — strip them so ``header_names`` +# reflects only what the client actually sent. Covers all three CDNs: Cloudflare (``cf-*``), +# Fastly (``fastly-*``), CloudFront (``cloudfront-*``), the shared ``x-forwarded-*`` / ``x-real-ip``, +# and the SDK's own routing header ``x-original-request-url``. +_EDGE_HEADER_PREFIXES = ("cf-", "fastly-", "cloudfront-", "x-forwarded-") +# ``host`` is included here because httpx synthesizes a Host header on Request construction; the JS +# fetch ``Request`` hides it as a forbidden header, so the TS SDK never emits it in ``header_names``. +# Stripping it keeps the cross-SDK header-name set consistent (host is captured in its own field). +# ``cdn-loop``/``x-varnish``/``via``/``surrogate-key``/``surrogate-control`` are portable proxy/CDN +# service-chain artifacts (esp. Fastly hops) — not client-sent, so they would pollute ``header_names``. +# Deployment-specific injected headers (e.g. x-geoip-*, x-ua-device) must be stripped at the edge +# instead; a portable SDK can't enumerate them. Mirrors the TS SDK's ``EDGE_HEADER_NAMES``. +_EDGE_HEADER_NAMES = frozenset( + { + "x-real-ip", + "x-original-request-url", + "host", + "cdn-loop", + "x-varnish", + "via", + "surrogate-key", + "surrogate-control", + } +) + +# Mechanical exploit markers for the query-string heuristic, matched case-insensitively against the +# raw and URL-decoded query. A coarse signal only — real classification stays query-time in the +# warehouse. +_SUSPICIOUS_QUERY_MARKERS = ( + "../", + "..\\", + "union select", + " str: + """Format as ``YYYY-MM-DDTHH:MM:SS.mmmZ`` to match the TS `Date.toISOString()` wire form.""" + return value.astimezone(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") + + +def _safe_pathname(request: Request) -> str: + """Return the request path with percent-encoding preserved. + + ``request.url.path`` percent-*decodes* (``/a%2Fb`` → ``/a/b``), which loses encoded path + semantics. We read ``raw_path`` (``path[?query]`` bytes), drop the query, and decode without + URL-decoding — matching the TS SDK's ``new URL(request.url).pathname``. + """ + path_bytes = request.url.raw_path.split(b"?", 1)[0] + return path_bytes.decode("utf-8", "replace") + + +def _enforcement_to_wire(mode: EnforcementMode) -> EnforcementWire: + # EnforcementMode values are already the wire strings ("observe"/"enforce"/"disabled"). + return mode.value # type: ignore[return-value] + + +def _truncate(value: str | None, max_length: int = MAX_FIELD_LENGTH) -> str | None: + if value is None: + return None + return value[:max_length] if len(value) > max_length else value + + +def _is_edge_header(name: str) -> bool: + if name in _EDGE_HEADER_NAMES: + return True + return any(name.startswith(prefix) for prefix in _EDGE_HEADER_PREFIXES) + + +def _collect_header_names(request: Request) -> list[str]: + names = {name.lower() for name in request.headers.keys()} + return sorted(name for name in names if not _is_edge_header(name)) + + +def _query_signals(request: Request) -> tuple[int, int, bool]: + # request.url.query is the raw, percent-encoded query bytes (no leading "?"), matching the + # TS SDK's ``url.search.slice(1)``. The raw query itself is never stored on the event. + raw = request.url.query.decode("utf-8", "replace") + params = [p for p in raw.split("&") if p] if raw else [] + + haystack = raw.lower() + "\n" + unquote(raw).lower() + suspicious = any(marker in haystack for marker in _SUSPICIOUS_QUERY_MARKERS) + + return len(raw), len(params), suspicious + + +def build_analytics_event( + request: Request, + decision: Decision, + context: BuildAnalyticsEventContext, +) -> AnalyticsEvent: + headers = request.headers + timestamp = context.timestamp if context.timestamp is not None else datetime.now(timezone.utc) + request_id = context.request_id if context.request_id is not None else str(uuid.uuid4()) + query_length, query_param_count, query_suspicious = _query_signals(request) + cdn = context.cdn_signals if context.cdn_signals is not None else CdnRequestSignals() + + return AnalyticsEvent( + timestamp=_iso_utc(timestamp), + request_id=request_id, + schema_version=SCHEMA_VERSION, + source_cdn=context.source_cdn, + user_agent=headers.get("user-agent", ""), + client_ip=normalize_client_ip(context.client_ip), + path=_safe_pathname(request), + method=request.method, + referer=headers.get("referer", ""), + accept_language=headers.get("accept-language", ""), + request_country=context.request_country, + request_asn=context.request_asn, + tls_fingerprint=context.tls_fingerprint, + has_token=decision.has_token, + token_outcome=decision.token_outcome, + final_action=decision.final_action, + enforcement_mode=_enforcement_to_wire(decision.enforcement_mode), + signature_agent=headers.get("signature-agent"), + signature_input=headers.get("signature-input"), + signature=headers.get("signature"), + # --- Capture v2: portable header signals --- + sec_fetch_mode=headers.get("sec-fetch-mode"), + sec_fetch_site=headers.get("sec-fetch-site"), + sec_fetch_dest=headers.get("sec-fetch-dest"), + sec_fetch_user=headers.get("sec-fetch-user"), + sec_ch_ua=_truncate(headers.get("sec-ch-ua")), + sec_ch_ua_mobile=headers.get("sec-ch-ua-mobile"), + sec_ch_ua_platform=headers.get("sec-ch-ua-platform"), + accept=_truncate(headers.get("accept")), + # httpx synthesizes the Host header from the URL, so this is effectively the parsed host. + host=headers.get("host") or request.url.host or None, + has_cookies="cookie" in headers, + header_names=_collect_header_names(request), + # Query-string derived signals (raw query never stored). + query_length=query_length, + query_param_count=query_param_count, + query_suspicious=query_suspicious, + # --- Capture v2: CDN plumbing (passthrough from the handler context) --- + accept_encoding=cdn.accept_encoding, + http_protocol=cdn.http_protocol, + tls_version=cdn.tls_version, + tls_cipher=cdn.tls_cipher, + tls_client_hello_length=cdn.tls_client_hello_length, + tls_client_extensions_sha1=cdn.tls_client_extensions_sha1, + as_organization=_truncate(cdn.as_organization), + client_tcp_rtt=cdn.client_tcp_rtt, + cdn_verified_bot_category=cdn.cdn_verified_bot_category, + request_priority=cdn.request_priority, + tls_fingerprint_ja4=cdn.tls_fingerprint_ja4, + ) diff --git a/supertab_connect/analytics/ip.py b/supertab_connect/analytics/ip.py new file mode 100644 index 0000000..417f2c3 --- /dev/null +++ b/supertab_connect/analytics/ip.py @@ -0,0 +1,27 @@ +"""Client-IP normalization (mirrors TS `analytics/ip.ts`). + +IPv4 addresses are mapped to their IPv6-mapped form (``::ffff:``); valid IPv6 +addresses pass through unchanged; anything else collapses to the unspecified address. +""" + +import ipaddress + +UNSPECIFIED = "::" + + +def normalize_client_ip(raw: str | None) -> str: + if not raw: + return UNSPECIFIED + trimmed = raw.strip() + if not trimmed: + return UNSPECIFIED + + try: + parsed = ipaddress.ip_address(trimmed) + except ValueError: + return UNSPECIFIED + + if parsed.version == 4: + return f"::ffff:{trimmed}" + # IPv6 passes through unchanged (the original textual form, not a re-compressed one). + return trimmed diff --git a/supertab_connect/analytics/transport.py b/supertab_connect/analytics/transport.py new file mode 100644 index 0000000..d1bd792 --- /dev/null +++ b/supertab_connect/analytics/transport.py @@ -0,0 +1,152 @@ +"""Analytics transports (mirrors TS `analytics/transport.ts`). + +The HTTP transport is fire-and-forget: ``emit`` schedules the POST on the running +event loop and returns immediately, never blocking the request path or raising. + +Unlike a module-level singleton, each ``HttpAnalyticsTransport`` owns its own HTTP +client and the set of in-flight emit tasks, so its lifetime is bounded by its owning +``SupertabConnect`` instance (mirroring the TS SDK). ``aclose`` drains in-flight emits +within a bounded timeout before closing the client, so tasks can never outlive the +client and resurrect a leaked one. +""" + +import asyncio +from dataclasses import asdict + +import httpx + +from supertab_connect._version import _get_sdk_user_agent +from supertab_connect.analytics.types import AnalyticsEvent, AnalyticsTransport +from supertab_connect.common import debug_log, error_log + +ANALYTICS_EVENTS_PATH = "/ingest/events" + +# Upper bound on how long aclose() waits for in-flight emits to finish before cancelling them. +_DEFAULT_FLUSH_TIMEOUT_SECONDS = 5.0 + + +class NoopAnalyticsTransport: + """A transport that discards every event. Used when analytics is disabled.""" + + def emit(self, event: AnalyticsEvent) -> None: + # intentional no-op + return None + + async def aclose(self) -> None: + # Nothing to drain or close; present for lifecycle symmetry with HttpAnalyticsTransport. + return None + + +class HttpAnalyticsTransport: + """Posts events to the Supertab Connect relay, fire-and-forget. + + Owns its HTTP client and in-flight emit tasks; call ``aclose`` (directly or via the + owning client's ``aclose`` / ``async with``) to flush and release them. + """ + + def __init__( + self, + *, + url: str, + api_key: str, + debug: bool = False, + flush_timeout: float = _DEFAULT_FLUSH_TIMEOUT_SECONDS, + ) -> None: + self._url = url + self._api_key = api_key + self._debug = debug + self._flush_timeout = flush_timeout + # Strong references to in-flight emit tasks so asyncio doesn't GC them mid-flight + # (it only keeps weak references to scheduled tasks). + self._tasks: set[asyncio.Task] = set() + self._client: httpx.AsyncClient | None = None + self._closed = False + + def _get_client(self) -> httpx.AsyncClient: + # Lazy: a transport that never emits (e.g. constructed with no running loop) never + # creates — and so never leaks — a client. + if self._client is None or self._client.is_closed: + self._client = httpx.AsyncClient(headers={"User-Agent": _get_sdk_user_agent()}) + return self._client + + def emit(self, event: AnalyticsEvent) -> None: + if self._closed: + debug_log(self._debug, "Skipping analytics emit: transport is closed") + return None + try: + loop = asyncio.get_running_loop() + except RuntimeError: + # No running event loop to schedule onto; analytics is best-effort, so skip. + debug_log(self._debug, "Skipping analytics emit: no running event loop") + return None + + task = loop.create_task(self._send(event)) + self._tasks.add(task) + task.add_done_callback(self._on_task_done) + return None + + def _on_task_done(self, task: asyncio.Task) -> None: + # Backstop: drop the reference and retrieve any exception so it never surfaces as an + # "exception was never retrieved" warning, even if _send's own guard is somehow bypassed. + self._tasks.discard(task) + if task.cancelled(): + return + error = task.exception() + if error is not None: + error_log(self._debug, f"analytics emit task error: {error}") + + async def _send(self, event: AnalyticsEvent) -> None: + # Fail-open: analytics must never block, slow, or alter request handling, so every error + # (transport, serialization, anything) is swallowed here rather than propagating. + # No _closed guard here: aclose() awaits (flush) or cancels every scheduled send before + # closing the client, so a send can neither be silently dropped nor resurrect a client. + try: + response = await self._get_client().post( + self._url, + json=asdict(event), + headers={ + "Authorization": f"Bearer {self._api_key}", + "Content-Type": "application/json", + }, + ) + if not response.is_success: + debug_log(self._debug, f"analytics emit failed: {response.status_code}") + except Exception as error: # noqa: BLE001 — fail-open guarantee, see comment above + error_log(self._debug, f"analytics emit error: {error}") + + async def flush(self) -> None: + """Await in-flight emit tasks, bounded by ``flush_timeout``. + + Does not close the client. Tasks scheduled after the snapshot are not awaited, so the + wait stays bounded even under a steady stream of emits. + """ + tasks = list(self._tasks) + if not tasks: + return + await asyncio.wait(tasks, timeout=self._flush_timeout) + + async def aclose(self) -> None: + """Flush in-flight emits (bounded) and close the HTTP client. + + After this returns, ``emit`` is a no-op. Any emit that outlived the flush timeout is + cancelled so it cannot resurrect the client after it is closed. + """ + self._closed = True + await self.flush() + stragglers = list(self._tasks) + for task in stragglers: + task.cancel() + if stragglers: + await asyncio.gather(*stragglers, return_exceptions=True) + if self._client is not None and not self._client.is_closed: + await self._client.aclose() + self._client = None + + +# Re-exported so callers can rely on structural typing without importing from `types`. +__all__ = [ + "ANALYTICS_EVENTS_PATH", + "AnalyticsTransport", + "HttpAnalyticsTransport", + "NoopAnalyticsTransport", +] diff --git a/supertab_connect/analytics/types.py b/supertab_connect/analytics/types.py new file mode 100644 index 0000000..07bce4b --- /dev/null +++ b/supertab_connect/analytics/types.py @@ -0,0 +1,144 @@ +"""Analytics event schema and transport protocol (mirrors TS `analytics/types.ts`).""" + +from dataclasses import dataclass +from typing import Literal, Protocol, runtime_checkable + +from supertab_connect.types import EnforcementMode, LicenseTokenInvalidReason + +SCHEMA_VERSION = 2 + +SourceCdn = Literal["cloudflare", "fastly", "cloudfront"] + +TokenOutcome = Literal[ + "absent", + "valid", + "expired", + "invalid_signature", + "invalid_audience", + "invalid_resource", + "invalid_issuer", + "malformed", + "server_error", + "not_validated", +] + +FinalAction = Literal["allow", "observe", "block"] + +EnforcementWire = Literal["observe", "enforce", "disabled"] + + +@dataclass(frozen=True) +class Decision: + has_token: bool + token_outcome: TokenOutcome + final_action: FinalAction + enforcement_mode: EnforcementMode + + +@dataclass(frozen=True) +class AnalyticsEvent: + timestamp: str + request_id: str + schema_version: int + # None when the request did not pass through a CDN (e.g. invoked directly via the SDK). + source_cdn: SourceCdn | None + + user_agent: str + client_ip: str + path: str + method: str + referer: str + accept_language: str + + # Classification signals — supplied by the CDN layer (platform-specific). None when not exposed. + request_country: str | None + request_asn: int | None + tls_fingerprint: str | None + + has_token: bool + token_outcome: TokenOutcome + final_action: FinalAction + enforcement_mode: EnforcementWire + + # HTTP Message Signature headers — platform-agnostic, read directly from request headers. + signature_agent: str | None + signature_input: str | None + signature: str | None + + # --- Capture v2 (schema_version 2): spoof-detection signals --- + # Portable header signals — read directly from request headers (every CDN). + sec_fetch_mode: str | None + sec_fetch_site: str | None + sec_fetch_dest: str | None + sec_fetch_user: str | None + sec_ch_ua: str | None + sec_ch_ua_mobile: str | None + sec_ch_ua_platform: str | None + accept: str | None + host: str | None + has_cookies: bool | None + # Lowercased, deduped, sorted request-header names with edge-injected headers + # (cf-*, x-forwarded-*, x-real-ip, …) and the synthesized Host stripped. Non-nullable: [] when none. + header_names: list[str] + + # Query-string derived signals. The raw query is NEVER stored (PII gate → option b); + # only these mechanical derivations are emitted. + query_length: int | None + query_param_count: int | None + query_suspicious: bool | None + + # CDN plumbing — not derivable from the portable Request. Supplied per platform by the + # caller via HandleRequestContext; null when not exposed. + accept_encoding: str | None + http_protocol: str | None + tls_version: str | None + tls_cipher: str | None + tls_client_hello_length: int | None + tls_client_extensions_sha1: str | None + as_organization: str | None + client_tcp_rtt: int | None + cdn_verified_bot_category: str | None + request_priority: str | None + tls_fingerprint_ja4: str | None + + +@dataclass(frozen=True) +class CdnRequestSignals: + """CDN-supplied request signals that cannot be read from the portable httpx ``Request``. + + Extracted per platform by the caller (Cloudflare ``request.cf``, Fastly headers, …) and + threaded through ``HandleRequestContext``. Field names match the wire (snake_case) contract, + so they pass straight through onto the event. + """ + + accept_encoding: str | None = None + http_protocol: str | None = None + tls_version: str | None = None + tls_cipher: str | None = None + tls_client_hello_length: int | None = None + tls_client_extensions_sha1: str | None = None + as_organization: str | None = None + client_tcp_rtt: int | None = None + cdn_verified_bot_category: str | None = None + request_priority: str | None = None + tls_fingerprint_ja4: str | None = None + + +@runtime_checkable +class AnalyticsTransport(Protocol): + def emit(self, event: AnalyticsEvent) -> None: + """Emit an analytics event. Implementations must never block the request path or raise.""" + ... + + +TOKEN_OUTCOME_BY_REASON: dict[LicenseTokenInvalidReason, TokenOutcome] = { + LicenseTokenInvalidReason.MISSING_TOKEN: "absent", + LicenseTokenInvalidReason.EXPIRED: "expired", + LicenseTokenInvalidReason.SIGNATURE_VERIFICATION_FAILED: "invalid_signature", + LicenseTokenInvalidReason.INVALID_AUDIENCE: "invalid_audience", + LicenseTokenInvalidReason.INVALID_ISSUER: "invalid_issuer", + LicenseTokenInvalidReason.INVALID_HEADER: "malformed", + LicenseTokenInvalidReason.INVALID_PAYLOAD: "malformed", + LicenseTokenInvalidReason.INVALID_ALG: "malformed", + LicenseTokenInvalidReason.SERVER_ERROR: "server_error", +} diff --git a/supertab_connect/customer/token.py b/supertab_connect/customer/token.py index 2776cb9..69841f2 100644 --- a/supertab_connect/customer/token.py +++ b/supertab_connect/customer/token.py @@ -15,6 +15,7 @@ from cryptography.hazmat.primitives.asymmetric import ec, rsa from cryptography.hazmat.primitives.serialization import load_pem_private_key +from supertab_connect._version import _get_sdk_user_agent from supertab_connect.common import debug_log, error_log from supertab_connect.exceptions import SupertabConnectError from supertab_connect.customer.content_matcher import _find_best_matching_content @@ -96,6 +97,15 @@ def _evict_expired_license_xml() -> None: def _create_async_client(**kwargs: Any) -> httpx.AsyncClient: kwargs.setdefault("follow_redirects", True) kwargs.setdefault("timeout", httpx.Timeout(_DEFAULT_HTTP_TIMEOUT_SECONDS)) + + headers = kwargs.pop("headers", None) + if headers is None: + headers = {"User-Agent": _get_sdk_user_agent()} + else: + headers = dict(headers) + headers.setdefault("User-Agent", _get_sdk_user_agent()) + kwargs["headers"] = headers + return httpx.AsyncClient(**kwargs) diff --git a/supertab_connect/merchant/client.py b/supertab_connect/merchant/client.py index 0e9f91a..a06002b 100644 --- a/supertab_connect/merchant/client.py +++ b/supertab_connect/merchant/client.py @@ -1,10 +1,29 @@ """High-level merchant client for Supertab Connect.""" +import json from collections.abc import Mapping from typing import ClassVar from httpx import Request +from supertab_connect._version import _get_sdk_version +from supertab_connect.analytics.build_analytics_event import ( + BuildAnalyticsEventContext, + build_analytics_event, +) +from supertab_connect.analytics.transport import ( + ANALYTICS_EVENTS_PATH, + HttpAnalyticsTransport, + NoopAnalyticsTransport, +) +from supertab_connect.analytics.types import ( + TOKEN_OUTCOME_BY_REASON, + AnalyticsTransport, + Decision, + FinalAction, + TokenOutcome, +) +from supertab_connect.common import error_log from supertab_connect.merchant.events import aclose_http_client as aclose_events_http_client from supertab_connect.merchant.license import ( build_block_result, @@ -13,9 +32,11 @@ verify_license_token, ) from supertab_connect.merchant.jwks import aclose_http_client as aclose_jwks_http_client +from supertab_connect.merchant.status import verify_status_challenge from supertab_connect.types import ( BotDetector, EnforcementMode, + HandleRequestContext, HandlerAction, HandlerResult, InvalidLicenseToken, @@ -25,11 +46,17 @@ ) _DEFAULT_BASE_URL = "https://api-connect.supertab.co" +# Analytics is served by the dedicated ingest service, not the API host. Kept separate from +# _base_url (mirroring set_base_url/get_base_url) so the relay can be pointed at a different +# host — or at localhost in dev — without moving token/JWKS/verify traffic. +_DEFAULT_ANALYTICS_BASE_URL = "https://ingest-connect.supertab.co" +_STATUS_PATH = "/.well-known/supertab/status" class SupertabConnect: _instance: ClassVar["SupertabConnect | None"] = None _base_url: ClassVar[str] = _DEFAULT_BASE_URL + _analytics_base_url: ClassVar[str] = _DEFAULT_ANALYTICS_BASE_URL def __new__(cls, config: SupertabConnectConfig, reset: bool = False) -> "SupertabConnect": if not reset and cls._instance is not None: @@ -57,9 +84,28 @@ def __init__(self, config: SupertabConnectConfig, reset: bool = False) -> None: self.bot_detector = config.bot_detector self.debug = config.debug self._base_url_override = config.supertab_base_url + # eventReporting (status endpoint) must reflect whether events are *actually* emitted. + # A configured custom transport (the DI/test seam) emits regardless of analytics_enabled, + # so the effective reporting state includes it — otherwise status could claim + # "eventReporting": false while events flow through the injected transport. + self._analytics_enabled = config.analytics_enabled or config.analytics_transport is not None + self._analytics_transport = self._build_analytics_transport(config) self._initialized = True type(self)._instance = self + def _build_analytics_transport(self, config: SupertabConnectConfig) -> AnalyticsTransport: + if config.analytics_transport is not None: + return config.analytics_transport + if not config.analytics_enabled: + return NoopAnalyticsTransport() + # Precedence: per-instance analytics_base_url > set_analytics_base_url() > the ingest default. + analytics_base_url = config.analytics_base_url or type(self)._analytics_base_url + return HttpAnalyticsTransport( + url=f"{analytics_base_url.rstrip('/')}{ANALYTICS_EVENTS_PATH}", + api_key=config.api_key, + debug=config.debug, + ) + @classmethod def reset_instance(cls) -> None: cls._instance = None @@ -72,6 +118,18 @@ def set_base_url(cls, url: str) -> None: def get_base_url(cls) -> str: return cls._base_url + @classmethod + def set_analytics_base_url(cls, url: str) -> None: + """Override the analytics ingest relay host (e.g. for a non-prod environment or local + development). Independent of set_base_url — token/JWKS/verify traffic is unaffected. + Can also be set per-instance via the ``analytics_base_url`` config option. + """ + cls._analytics_base_url = url + + @classmethod + def get_analytics_base_url(cls) -> str: + return cls._analytics_base_url + @property def base_url(self) -> str: return self._base_url_override or type(self)._base_url @@ -79,6 +137,11 @@ def base_url(self) -> str: async def aclose(self) -> None: await aclose_events_http_client() await aclose_jwks_http_client() + # The analytics transport owns its own client/tasks; drain them if it is closable. + # Injected custom transports may implement `emit` only, so this is duck-typed. + transport_aclose = getattr(self._analytics_transport, "aclose", None) + if callable(transport_aclose): + await transport_aclose() async def __aenter__(self) -> "SupertabConnect": return self @@ -138,17 +201,134 @@ def _detect_bot(self, request: Request) -> bool: return detector(request) - async def handle_request(self, request: Request) -> HandlerResult: + def _emit_analytics( + self, + request: Request, + context: HandleRequestContext | None, + *, + has_token: bool, + token_outcome: TokenOutcome, + final_action: FinalAction, + ) -> None: + try: + event = build_analytics_event( + request, + Decision( + has_token=has_token, + token_outcome=token_outcome, + final_action=final_action, + enforcement_mode=self.enforcement, + ), + BuildAnalyticsEventContext( + source_cdn=context.source_cdn if context else None, + request_id=context.request_id if context else None, + client_ip=context.client_ip if context else None, + request_country=context.request_country if context else None, + request_asn=context.request_asn if context else None, + tls_fingerprint=context.tls_fingerprint if context else None, + cdn_signals=context.cdn_signals if context else None, + ), + ) + self._analytics_transport.emit(event) + except Exception as error: # noqa: BLE001 — analytics must never break request handling + error_log(self.debug, f"failed to build/emit analytics event: {error}") + + @staticmethod + def _request_origin(request: Request) -> str: + """The scheme://host[:port] origin of the request, matching JS `URL.origin`. + + httpx normalizes away default ports (80/443), so they are never appended. + """ + url = request.url + origin = f"{url.scheme}://{url.host}" + if url.port is not None: + origin += f":{url.port}" + return origin + + async def _handle_status_request(self, request: Request, context: HandleRequestContext | None) -> HandlerResult: + """Answer the self-report status probe. + + Serves the live SDK config to a valid backend-signed challenge, else a minimal + 404. Short-circuits ahead of token verification, bot detection, and analytics. + """ + headers = {"Content-Type": "application/json", "Cache-Control": "no-store"} + auth = request.headers.get("authorization", "") + auth_parts = auth.split(None, 1) + token = auth_parts[1] if len(auth_parts) == 2 and auth_parts[0].lower() == "bearer" else "" + ok = ( + await verify_status_challenge( + token, + expected_audience=self._request_origin(request), + base_url=self.base_url, + debug=self.debug, + ) + if token + else False + ) + if not ok: + return { + "action": HandlerAction.RESPOND, + "status": 404, + "body": json.dumps({"supertab": True}), + "headers": headers, + } + body = json.dumps( + { + "runtime": context.source_cdn if context else None, + "component": {"kind": "python-sdk", "version": _get_sdk_version()}, + "enforcement": self.enforcement.value, + "eventReporting": self._analytics_enabled, + } + ) + return { + "action": HandlerAction.RESPOND, + "status": 200, + "body": body, + "headers": headers, + } + + @staticmethod + def _is_status_probe(request: Request) -> bool: + """Match the status probe on the exact ``GET /.well-known/supertab/status`` route only. + + Uses the *raw* path (percent-encoding preserved) rather than ``request.url.path``, which + decodes ``%2F``/``%2E`` — so encoded look-alikes reach the application instead of being + served the SDK's response. Requiring GET keeps non-GET requests to the same path flowing + to the app. Mirrors the TS SDK's ``new URL().pathname`` comparison (plus a method check). + """ + if request.method != "GET": + return False + raw_path = request.url.raw_path.split(b"?", 1)[0].decode("utf-8", "replace") + return raw_path == _STATUS_PATH + + async def handle_request(self, request: Request, context: HandleRequestContext | None = None) -> HandlerResult: + # The self-report status probe short-circuits ahead of everything else: it is answered + # directly (never forwarded to origin) and emits no analytics. + if self._is_status_probe(request): + return await self._handle_status_request(request, context) + auth = request.headers.get("authorization", "") token = None auth_parts = auth.split(None, 1) if len(auth_parts) == 2 and auth_parts[0].lower() == "license": token = auth_parts[1] + has_token = token is not None url = str(request.url) user_agent = request.headers.get("user-agent", "unknown") + # Token present → validate, regardless of bot detection — except in DISABLED + # mode, which short-circuits to ALLOW without verification. if token: if self.enforcement is EnforcementMode.DISABLED: + # DISABLED short-circuits to ALLOW without verifying the token, so we cannot + # honestly claim "valid"; emit "not_validated" so it is not counted as licensed. + self._emit_analytics( + request, + context, + has_token=has_token, + token_outcome="not_validated", + final_action="allow", + ) return {"action": HandlerAction.ALLOW} verification = await verify_and_record_event( @@ -161,22 +341,64 @@ async def handle_request(self, request: Request) -> HandlerResult: request_headers=dict(request.headers.items()), ) if isinstance(verification, InvalidLicenseToken): + self._emit_analytics( + request, + context, + has_token=has_token, + token_outcome=TOKEN_OUTCOME_BY_REASON.get(verification.reason, "malformed"), + final_action="block", + ) return build_block_result( reason=verification.reason, error=verification.error, request_url=url, ) + self._emit_analytics( + request, + context, + has_token=has_token, + token_outcome="valid", + final_action="allow", + ) return {"action": HandlerAction.ALLOW} if not self._detect_bot(request): + self._emit_analytics( + request, + context, + has_token=has_token, + token_outcome="absent", + final_action="allow", + ) return {"action": HandlerAction.ALLOW} - if self.enforcement is EnforcementMode.STRICT: + if self.enforcement is EnforcementMode.ENFORCE: + self._emit_analytics( + request, + context, + has_token=has_token, + token_outcome="absent", + final_action="block", + ) return build_block_result( reason=LicenseTokenInvalidReason.MISSING_TOKEN, error="Authorization header missing or malformed", request_url=url, ) - if self.enforcement is EnforcementMode.SOFT: + if self.enforcement is EnforcementMode.OBSERVE: + self._emit_analytics( + request, + context, + has_token=has_token, + token_outcome="absent", + final_action="observe", + ) return build_signal_result(url) + self._emit_analytics( + request, + context, + has_token=has_token, + token_outcome="absent", + final_action="allow", + ) return {"action": HandlerAction.ALLOW} diff --git a/supertab_connect/merchant/jwks.py b/supertab_connect/merchant/jwks.py index 1115aa6..7b0f011 100644 --- a/supertab_connect/merchant/jwks.py +++ b/supertab_connect/merchant/jwks.py @@ -1,23 +1,35 @@ """JWKS fetching and caching for platform key verification.""" +import asyncio import time from typing import Any import httpx +from supertab_connect._version import _get_sdk_user_agent from supertab_connect.common import debug_log, error_log from supertab_connect.exceptions import JwksKeyNotFoundError JWKS_CACHE_TTL_SECONDS = 48 * 3600 # 48 hours +# Minimum spacing between key-rotation-triggered refreshes, per base_url. A JWT's `kid` is read +# from its unverified header, so an unauthenticated caller (e.g. via the public status endpoint) +# can present tokens with rotating unknown kids; without this floor each one would bypass the +# cache and force a backend JWKS fetch. Well below the TTL, so genuine rotations still recover +# on the first miss. +_JWKS_MIN_REFRESH_INTERVAL_SECONDS = 60 _jwks_cache: dict[str, dict[str, Any]] = {} +# Monotonic timestamp of the last rotation-refresh per base_url, for the cooldown above. +_jwks_last_refresh: dict[str, float] = {} +# Per-base_url locks so concurrent misses coalesce into a single refresh (single-flight). +_jwks_refresh_locks: dict[str, asyncio.Lock] = {} _http_client: httpx.AsyncClient | None = None def _get_http_client() -> httpx.AsyncClient: global _http_client if _http_client is None or _http_client.is_closed: - _http_client = httpx.AsyncClient() + _http_client = httpx.AsyncClient(headers={"User-Agent": _get_sdk_user_agent()}) return _http_client @@ -28,16 +40,19 @@ async def aclose_http_client() -> None: _http_client = None -async def fetch_platform_jwks(base_url: str, *, debug: bool = False) -> dict[str, Any]: +async def fetch_platform_jwks(base_url: str, *, force: bool = False, debug: bool = False) -> dict[str, Any]: """Fetch the platform JWKS from the Supertab well-known endpoint. Results are cached per base_url for 48 hours. Subsequent calls within the TTL return the cached key set without making a network request. + + ``force=True`` bypasses the TTL and refetches; on a fetch failure the previously cached + entry is left intact (the write only happens after a successful response). """ normalized_url = base_url.rstrip("/") now = time.monotonic() cached = _jwks_cache.get(normalized_url) - if cached is not None and (now - cached["cached_at"]) < JWKS_CACHE_TTL_SECONDS: + if not force and cached is not None and (now - cached["cached_at"]) < JWKS_CACHE_TTL_SECONDS: return cached["jwks"] jwks_url = f"{normalized_url}/.well-known/jwks.json/platform" @@ -56,9 +71,36 @@ async def fetch_platform_jwks(base_url: str, *, debug: bool = False) -> dict[str raise +async def refresh_platform_jwks_on_miss(base_url: str, *, debug: bool = False) -> dict[str, Any]: + """Refresh the cached JWKS after a `kid` miss, throttled to protect the backend. + + Called when a token's `kid` is absent from the currently cached key set — normally a sign the + platform rotated its signing keys. Because `kid` comes from an unverified JWT header, an + unauthenticated caller could otherwise force a backend fetch per request; this refetches at + most once per ``_JWKS_MIN_REFRESH_INTERVAL_SECONDS`` per base_url and coalesces concurrent + callers via a per-base_url lock (single-flight). Returns the freshest key set available — + the refreshed set, or the current cache when throttled — and the caller re-checks the `kid`, + failing closed if it is still absent. + """ + normalized_url = base_url.rstrip("/") + lock = _jwks_refresh_locks.setdefault(normalized_url, asyncio.Lock()) + async with lock: + now = time.monotonic() + last = _jwks_last_refresh.get(normalized_url) + if last is not None and (now - last) < _JWKS_MIN_REFRESH_INTERVAL_SECONDS: + debug_log(debug, "Skipping JWKS refresh: a refresh happened within the cooldown window") + cached = _jwks_cache.get(normalized_url) + return cached["jwks"] if cached is not None else {"keys": []} + # Record the attempt before fetching so a failed refetch still spends the cooldown — + # a broken backend must not reopen the per-request fetch amplification. + _jwks_last_refresh[normalized_url] = now + return await fetch_platform_jwks(normalized_url, force=True, debug=debug) + + def clear_jwks_cache() -> None: - """Invalidate all cached JWKS data, forcing a fresh fetch on next call.""" + """Invalidate all cached JWKS data and refresh throttling, forcing a fresh fetch on next call.""" _jwks_cache.clear() + _jwks_last_refresh.clear() def _find_key_by_kid(jwks: dict[str, Any], kid: str | None) -> dict[str, Any]: diff --git a/supertab_connect/merchant/license.py b/supertab_connect/merchant/license.py index 09190f2..f099110 100644 --- a/supertab_connect/merchant/license.py +++ b/supertab_connect/merchant/license.py @@ -14,7 +14,11 @@ from supertab_connect.exceptions import JwksKeyNotFoundError from supertab_connect.merchant.events import record_event from supertab_connect.merchant.headers import to_event_properties -from supertab_connect.merchant.jwks import _find_key_by_kid, clear_jwks_cache, fetch_platform_jwks +from supertab_connect.merchant.jwks import ( + _find_key_by_kid, + fetch_platform_jwks, + refresh_platform_jwks_on_miss, +) from supertab_connect.types import ( AllowHandlerResult, BlockHandlerResult, @@ -200,8 +204,16 @@ async def verify_license_token( return ValidLicenseToken(license_id=license_id, payload=verified_payload) except JwksKeyNotFoundError: if attempt == 0: - debug_log(debug, "Key not found in cached JWKS, clearing cache and retrying...") - clear_jwks_cache() + debug_log(debug, "Key not found in cached JWKS, refreshing and retrying...") + try: + await refresh_platform_jwks_on_miss(supertab_base_url, debug=debug) + except Exception: # noqa: BLE001 — a failed refresh maps to server_error, like the initial fetch + error_log(debug, "Failed to refresh platform JWKS") + return InvalidLicenseToken( + reason=LicenseTokenInvalidReason.SERVER_ERROR, + error=_reason_to_error_description(LicenseTokenInvalidReason.SERVER_ERROR), + license_id=license_id, + ) continue debug_log(debug, "Key not found after JWKS cache refresh") return InvalidLicenseToken( diff --git a/supertab_connect/merchant/status.py b/supertab_connect/merchant/status.py new file mode 100644 index 0000000..3414ce5 --- /dev/null +++ b/supertab_connect/merchant/status.py @@ -0,0 +1,72 @@ +"""Verification of backend-signed status-probe challenges. + +Powers the self-report endpoint (`/.well-known/supertab/status`): the backend signs a +short-lived challenge JWT scoped to the site origin, and the SDK proves it is live and +reports its config only when that challenge verifies. Mirrors the TS SDK `status.ts`. +""" + +from typing import cast + +import jwt +import jwt.algorithms +from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePublicKey + +from supertab_connect.common import debug_log, error_log +from supertab_connect.exceptions import JwksKeyNotFoundError +from supertab_connect.merchant.jwks import ( + _find_key_by_kid, + fetch_platform_jwks, + refresh_platform_jwks_on_miss, +) + +_STATUS_PROBE_PURPOSE = "status-probe" +# Clock skew tolerance for the challenge's exp/iat, matching the TS SDK's 5s. +_CLOCK_TOLERANCE_SECONDS = 5 + + +async def verify_status_challenge( + token: str, + *, + expected_audience: str, + base_url: str, + debug: bool = False, +) -> bool: + """Verify a backend-signed status-probe challenge. + + Returns True only when the token is a valid ES256 JWT signed by the platform JWKS, + scoped to ``expected_audience`` (the site origin), and carrying ``purpose: + status-probe``. Any failure resolves to False rather than raising — the endpoint + falls back to a minimal 404. A stale-key miss triggers one JWKS cache refresh + retry. + """ + + async def _verify() -> bool: + jwks = await fetch_platform_jwks(base_url, debug=debug) + header = jwt.get_unverified_header(token) + jwk_key = _find_key_by_kid(jwks, header.get("kid")) + public_key = cast(EllipticCurvePublicKey, jwt.algorithms.ECAlgorithm.from_jwk(jwk_key)) + payload = jwt.decode( + token, + key=public_key, + algorithms=["ES256"], + audience=expected_audience, + leeway=_CLOCK_TOLERANCE_SECONDS, + # A "short-lived" challenge must actually carry an expiry: without this, a signed + # challenge lacking exp (or iat) verifies and never expires — replayable forever. + # PyJWT validates exp/iat only when present, so require them explicitly. + options={"require": ["exp", "iat"]}, + ) + return payload.get("purpose") == _STATUS_PROBE_PURPOSE + + try: + return await _verify() + except JwksKeyNotFoundError: + debug_log(debug, "Key not found in cached JWKS, refreshing and retrying...") + try: + await refresh_platform_jwks_on_miss(base_url, debug=debug) + return await _verify() + except Exception as error: # noqa: BLE001 — status probe fails closed to a minimal 404 + error_log(debug, f"Status challenge verification failed after JWKS refresh: {error}") + return False + except Exception as error: # noqa: BLE001 — status probe fails closed to a minimal 404 + error_log(debug, f"Status challenge verification failed: {error}") + return False diff --git a/supertab_connect/types.py b/supertab_connect/types.py index 54a0bf4..677074e 100644 --- a/supertab_connect/types.py +++ b/supertab_connect/types.py @@ -3,15 +3,18 @@ from collections.abc import Callable from dataclasses import dataclass, field from enum import StrEnum -from typing import Any, Literal, NotRequired, TypeAlias, TypedDict +from typing import TYPE_CHECKING, Any, Literal, NotRequired, TypeAlias, TypedDict from httpx import Request +if TYPE_CHECKING: + from supertab_connect.analytics.types import AnalyticsTransport, CdnRequestSignals + class EnforcementMode(StrEnum): DISABLED = "disabled" - SOFT = "soft" - STRICT = "strict" + OBSERVE = "observe" + ENFORCE = "enforce" class LicenseTokenInvalidReason(StrEnum): @@ -29,6 +32,7 @@ class LicenseTokenInvalidReason(StrEnum): class HandlerAction(StrEnum): ALLOW = "allow" BLOCK = "block" + RESPOND = "respond" class UsageType(StrEnum): @@ -46,10 +50,36 @@ class UsageType(StrEnum): @dataclass(frozen=True) class SupertabConnectConfig: api_key: str - enforcement: EnforcementMode = EnforcementMode.SOFT + enforcement: EnforcementMode = EnforcementMode.OBSERVE supertab_base_url: str | None = None bot_detector: BotDetector | None = None debug: bool = False + # Enables analytics emission to the Supertab Connect relay. Default: False. + analytics_enabled: bool = False + # Base URL of the analytics ingest relay. Defaults to the dedicated ingest service + # (https://ingest-connect.supertab.co) — separate from the API base URL used for token + # acquisition / JWKS / verification. Override for non-prod or local development. + analytics_base_url: str | None = None + # Internal dependency-injection seam: overrides the default HttpAnalyticsTransport when provided. + # Used by tests to inject in-memory transports. Not a merchant-facing option. + analytics_transport: "AnalyticsTransport | None" = None + + +@dataclass(frozen=True) +class HandleRequestContext: + """Optional CDN-supplied request context for `handle_request`. + + All fields are omitted (None) for direct SDK invocation that did not pass through a CDN. + """ + + source_cdn: Literal["cloudflare", "fastly", "cloudfront"] | None = None + client_ip: str | None = None + request_id: str | None = None + request_country: str | None = None + request_asn: int | None = None + tls_fingerprint: str | None = None + # Capture-v2 CDN plumbing not derivable from the portable Request (e.g. Cloudflare request.cf). + cdn_signals: "CdnRequestSignals | None" = None class AllowHandlerResult(TypedDict): @@ -64,7 +94,20 @@ class BlockHandlerResult(TypedDict): headers: dict[str, str] -HandlerResult: TypeAlias = AllowHandlerResult | BlockHandlerResult +class RespondHandlerResult(TypedDict): + """A fully-formed response the caller must serve verbatim without contacting origin. + + Emitted for the self-report status probe (`/.well-known/supertab/status`), which the + SDK answers directly rather than forwarding. + """ + + action: Literal[HandlerAction.RESPOND] + status: int + body: str + headers: dict[str, str] + + +HandlerResult: TypeAlias = AllowHandlerResult | BlockHandlerResult | RespondHandlerResult @dataclass(frozen=True) diff --git a/tests/analytics/__init__.py b/tests/analytics/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/analytics/test_build_analytics_event.py b/tests/analytics/test_build_analytics_event.py new file mode 100644 index 0000000..72fbe5d --- /dev/null +++ b/tests/analytics/test_build_analytics_event.py @@ -0,0 +1,367 @@ +"""Tests for building relay analytics events.""" + +from dataclasses import asdict, replace +from datetime import datetime, timezone + +import httpx +import pytest + +from supertab_connect.analytics.build_analytics_event import ( + BuildAnalyticsEventContext, + build_analytics_event, +) +from supertab_connect.analytics.types import SCHEMA_VERSION, CdnRequestSignals, Decision +from supertab_connect.types import EnforcementMode + +FIXED_TIME = datetime(2026, 4, 29, 12, 0, 0, tzinfo=timezone.utc) +REQUEST_ID = "req-123" + +BASE_DECISION = Decision( + has_token=False, + token_outcome="absent", + final_action="allow", + enforcement_mode=EnforcementMode.OBSERVE, +) + + +def _make_request( + *, + url: str = "https://example.com/articles/foo?x=1", + method: str = "GET", + headers: dict[str, str] | None = None, +) -> httpx.Request: + return httpx.Request(method, url, headers=headers or {}) + + +def _ctx(**extra) -> BuildAnalyticsEventContext: + base = BuildAnalyticsEventContext(request_id=REQUEST_ID, source_cdn="cloudflare", timestamp=FIXED_TIME) + return replace(base, **extra) + + +def test_returns_event_matching_relay_shape(): + request = _make_request( + headers={ + "user-agent": "Mozilla/5.0", + "referer": "https://example.com/", + "accept-language": "en-US,en;q=0.9", + } + ) + + event = build_analytics_event(request, BASE_DECISION, _ctx(client_ip="1.2.3.4")) + + assert asdict(event) == { + "timestamp": "2026-04-29T12:00:00.000Z", + "request_id": REQUEST_ID, + "schema_version": SCHEMA_VERSION, + "source_cdn": "cloudflare", + "user_agent": "Mozilla/5.0", + "client_ip": "::ffff:1.2.3.4", + "path": "/articles/foo", + "method": "GET", + "referer": "https://example.com/", + "accept_language": "en-US,en;q=0.9", + "request_country": None, + "request_asn": None, + "tls_fingerprint": None, + "has_token": False, + "token_outcome": "absent", + "final_action": "allow", + "enforcement_mode": "observe", + "signature_agent": None, + "signature_input": None, + "signature": None, + # Capture v2 — portable header signals (none of these headers were sent). + "sec_fetch_mode": None, + "sec_fetch_site": None, + "sec_fetch_dest": None, + "sec_fetch_user": None, + "sec_ch_ua": None, + "sec_ch_ua_mobile": None, + "sec_ch_ua_platform": None, + "accept": None, + "host": "example.com", + "has_cookies": False, + "header_names": ["accept-language", "referer", "user-agent"], + "query_length": 3, + "query_param_count": 1, + "query_suspicious": False, + # Capture v2 — CDN plumbing (no cdn_signals in context → None). + "accept_encoding": None, + "http_protocol": None, + "tls_version": None, + "tls_cipher": None, + "tls_client_hello_length": None, + "tls_client_extensions_sha1": None, + "as_organization": None, + "client_tcp_rtt": None, + "cdn_verified_bot_category": None, + "request_priority": None, + "tls_fingerprint_ja4": None, + } + + +def test_passes_through_classification_signals(): + event = build_analytics_event( + _make_request(), + BASE_DECISION, + _ctx(request_country="DE", request_asn=3320, tls_fingerprint="abc123"), + ) + assert event.request_country == "DE" + assert event.request_asn == 3320 + assert event.tls_fingerprint == "abc123" + + +def test_classification_signals_default_to_none(): + event = build_analytics_event(_make_request(), BASE_DECISION, _ctx()) + assert event.request_country is None + assert event.request_asn is None + assert event.tls_fingerprint is None + + +def test_reads_signature_headers_from_request(): + request = _make_request( + headers={ + "signature-agent": "https://agent.example", + "signature-input": "sig1=(...)", + "signature": "sig1=:abc:", + } + ) + event = build_analytics_event(request, BASE_DECISION, _ctx()) + assert event.signature_agent == "https://agent.example" + assert event.signature_input == "sig1=(...)" + assert event.signature == "sig1=:abc:" + + +def test_signature_headers_default_to_none(): + event = build_analytics_event(_make_request(), BASE_DECISION, _ctx()) + assert event.signature_agent is None + assert event.signature_input is None + assert event.signature is None + + +@pytest.mark.parametrize("final_action", ["allow", "observe", "block"]) +def test_passes_through_final_action(final_action): + decision = replace(BASE_DECISION, final_action=final_action) + event = build_analytics_event(_make_request(), decision, _ctx()) + assert event.final_action == final_action + + +@pytest.mark.parametrize( + ("mode", "wire"), + [ + (EnforcementMode.OBSERVE, "observe"), + (EnforcementMode.ENFORCE, "enforce"), + (EnforcementMode.DISABLED, "disabled"), + ], +) +def test_serializes_enforcement_mode_to_wire(mode, wire): + decision = replace(BASE_DECISION, enforcement_mode=mode) + event = build_analytics_event(_make_request(), decision, _ctx()) + assert event.enforcement_mode == wire + + +def test_source_cdn_is_none_for_direct_sdk_invocation(): + event = build_analytics_event(_make_request(), BASE_DECISION, BuildAnalyticsEventContext()) + assert event.source_cdn is None + + +def test_generates_request_id_when_absent(): + event = build_analytics_event(_make_request(), BASE_DECISION, BuildAnalyticsEventContext(timestamp=FIXED_TIME)) + assert event.request_id # a uuid4 string + + +def test_path_preserves_percent_encoding(): + # request.url.path would decode %2F->"/" and %20->" "; the event must keep encoded semantics. + request = _make_request(url="https://example.com/a%2Fb/c%20d?x=1") + event = build_analytics_event(request, BASE_DECISION, _ctx()) + assert event.path == "/a%2Fb/c%20d" + + +def test_path_drops_query_string(): + request = _make_request(url="https://example.com/articles/foo?x=1&y=2") + event = build_analytics_event(request, BASE_DECISION, _ctx()) + assert event.path == "/articles/foo" + + +def test_missing_headers_default_to_empty_strings(): + event = build_analytics_event(_make_request(), BASE_DECISION, _ctx()) + assert event.user_agent == "" + assert event.referer == "" + assert event.accept_language == "" + + +# --- Capture v2 ------------------------------------------------------------------------------- + +BROWSER_HEADERS = { + "user-agent": "Mozilla/5.0", + "sec-fetch-mode": "navigate", + "sec-fetch-site": "none", + "sec-fetch-dest": "document", + "sec-fetch-user": "?1", + "sec-ch-ua": '"Chromium";v="120", "Not(A:Brand";v="24"', + "sec-ch-ua-mobile": "?0", + "sec-ch-ua-platform": '"macOS"', + "accept": "text/html", + "cookie": "session=abc", +} + + +def test_captures_sec_fetch_and_client_hints_from_browser_request(): + event = build_analytics_event(_make_request(headers=BROWSER_HEADERS), BASE_DECISION, _ctx()) + assert event.sec_fetch_mode == "navigate" + assert event.sec_fetch_site == "none" + assert event.sec_fetch_dest == "document" + assert event.sec_fetch_user == "?1" + assert event.sec_ch_ua == '"Chromium";v="120", "Not(A:Brand";v="24"' + assert event.sec_ch_ua_mobile == "?0" + assert event.sec_ch_ua_platform == '"macOS"' + assert event.accept == "text/html" + assert event.has_cookies is True + + +def test_curl_like_request_carries_no_browser_signals(): + event = build_analytics_event(_make_request(headers={"user-agent": "curl/8.0"}), BASE_DECISION, _ctx()) + assert event.sec_fetch_mode is None + assert event.sec_fetch_site is None + assert event.sec_fetch_dest is None + assert event.sec_fetch_user is None + assert event.sec_ch_ua is None + assert event.sec_ch_ua_mobile is None + assert event.sec_ch_ua_platform is None + assert event.has_cookies is False + + +def test_host_falls_back_to_url_host(): + event = build_analytics_event(_make_request(url="https://pub.example.com/a"), BASE_DECISION, _ctx()) + assert event.host == "pub.example.com" + + +def test_truncates_accept_and_sec_ch_ua_to_512_chars(): + long = "a" * 600 + event = build_analytics_event(_make_request(headers={"accept": long, "sec-ch-ua": long}), BASE_DECISION, _ctx()) + assert event.accept == "a" * 512 + assert event.sec_ch_ua == "a" * 512 + + +def test_header_names_lowercased_deduped_sorted(): + event = build_analytics_event( + _make_request(headers={"User-Agent": "x", "Accept": "y", "Referer": "z"}), + BASE_DECISION, + _ctx(), + ) + assert event.header_names == ["accept", "referer", "user-agent"] + + +def test_header_names_strips_edge_injected_headers_across_all_cdns(): + event = build_analytics_event( + _make_request( + headers={ + "user-agent": "x", + # Cloudflare + "cf-connecting-ip": "1.2.3.4", + "cf-ray": "abc", + # Fastly + "fastly-client-ip": "1.2.3.4", + "fastly-client-ja3": "deadbeef", + # CloudFront + "cloudfront-viewer-country": "DE", + "cloudfront-viewer-ja3-fingerprint": "abc", + # shared / SDK routing / synthesized + "x-forwarded-for": "1.2.3.4", + "x-real-ip": "1.2.3.4", + "x-original-request-url": "https://pub.example.com/a", + # portable proxy/CDN service-chain artifacts (esp. Fastly hops) + "cdn-loop": "fastly", + "x-varnish": "12345", + "via": "1.1 varnish", + "surrogate-key": "key-a key-b", + "surrogate-control": "max-age=3600", + } + ), + BASE_DECISION, + _ctx(), + ) + # host is stripped too (httpx synthesizes it; the TS SDK never emits it). + assert event.header_names == ["user-agent"] + + +def test_query_signals_derived_without_storing_raw_query(): + event = build_analytics_event(_make_request(url="https://x.test/p?a=1&b=2&c=3"), BASE_DECISION, _ctx()) + assert event.query_length == len("a=1&b=2&c=3") + assert event.query_param_count == 3 + assert event.query_suspicious is False + # The raw query string must never appear on the event. + assert "a=1&b=2&c=3" not in str(asdict(event)) + + +def test_query_signals_are_zero_for_query_less_url(): + event = build_analytics_event(_make_request(url="https://x.test/p"), BASE_DECISION, _ctx()) + assert event.query_length == 0 + assert event.query_param_count == 0 + assert event.query_suspicious is False + + +@pytest.mark.parametrize( + "url", + [ + "https://x.test/?f=../../etc/passwd", + "https://x.test/?q=UNION%20SELECT%201", + "https://x.test/?x=%3Cscript%3E", + ], +) +def test_query_suspicious_flags_exploit_markers_raw_and_encoded(url): + event = build_analytics_event(_make_request(url=url), BASE_DECISION, _ctx()) + assert event.query_suspicious is True + + +def test_cdn_signals_passthrough_with_truncation(): + event = build_analytics_event( + _make_request(), + BASE_DECISION, + _ctx( + cdn_signals=CdnRequestSignals( + accept_encoding="gzip, br", + http_protocol="HTTP/2", + tls_version="TLSv1.3", + tls_cipher="AEAD-AES128-GCM-SHA256", + tls_client_hello_length=1811, + tls_client_extensions_sha1="4cFD...", + as_organization="o" * 600, + client_tcp_rtt=50, + cdn_verified_bot_category="Search Engine Crawler", + request_priority="weight=256;exclusive=1", + tls_fingerprint_ja4=None, + ) + ), + ) + assert event.accept_encoding == "gzip, br" + assert event.http_protocol == "HTTP/2" + assert event.tls_version == "TLSv1.3" + assert event.tls_cipher == "AEAD-AES128-GCM-SHA256" + assert event.tls_client_hello_length == 1811 + assert event.tls_client_extensions_sha1 == "4cFD..." + assert event.as_organization == "o" * 512 + assert event.client_tcp_rtt == 50 + assert event.cdn_verified_bot_category == "Search Engine Crawler" + assert event.request_priority == "weight=256;exclusive=1" + assert event.tls_fingerprint_ja4 is None + + +def test_cdn_signals_default_to_none_when_absent(): + event = build_analytics_event(_make_request(), BASE_DECISION, _ctx()) + assert event.accept_encoding is None + assert event.http_protocol is None + assert event.tls_version is None + assert event.tls_cipher is None + assert event.tls_client_hello_length is None + assert event.tls_client_extensions_sha1 is None + assert event.as_organization is None + assert event.client_tcp_rtt is None + assert event.cdn_verified_bot_category is None + assert event.request_priority is None + assert event.tls_fingerprint_ja4 is None + + +def test_schema_version_is_2(): + event = build_analytics_event(_make_request(), BASE_DECISION, _ctx()) + assert event.schema_version == 2 diff --git a/tests/analytics/test_ip.py b/tests/analytics/test_ip.py new file mode 100644 index 0000000..a0dbc83 --- /dev/null +++ b/tests/analytics/test_ip.py @@ -0,0 +1,28 @@ +"""Tests for client-IP normalization.""" + +import pytest + +from supertab_connect.analytics.ip import normalize_client_ip + + +def test_maps_ipv4_to_ipv6_mapped_form(): + assert normalize_client_ip("1.2.3.4") == "::ffff:1.2.3.4" + assert normalize_client_ip("192.0.2.1") == "::ffff:192.0.2.1" + + +def test_trims_surrounding_whitespace_before_mapping_ipv4(): + assert normalize_client_ip(" 1.2.3.4 ") == "::ffff:1.2.3.4" + + +def test_passes_ipv6_through_unchanged(): + assert normalize_client_ip("2001:db8::1") == "2001:db8::1" + assert normalize_client_ip("::1") == "::1" + + +@pytest.mark.parametrize("value", [None, "", " "]) +def test_returns_unspecified_for_empty(value): + assert normalize_client_ip(value) == "::" + + +def test_returns_unspecified_for_unrecognized_value(): + assert normalize_client_ip("not-an-ip") == "::" diff --git a/tests/analytics/test_transport.py b/tests/analytics/test_transport.py new file mode 100644 index 0000000..0220c84 --- /dev/null +++ b/tests/analytics/test_transport.py @@ -0,0 +1,254 @@ +"""Tests for analytics transports.""" + +import asyncio +import json + +import httpx +import respx + +from supertab_connect._version import _get_sdk_user_agent +from supertab_connect.analytics.transport import ( + ANALYTICS_EVENTS_PATH, + HttpAnalyticsTransport, + NoopAnalyticsTransport, +) +from supertab_connect.analytics.types import AnalyticsEvent + +RELAY_URL = "https://relay.test/ingest/events" + +FIXTURE_EVENT = AnalyticsEvent( + timestamp="2026-04-29T12:00:00.000Z", + request_id="req-1", + schema_version=2, + source_cdn="cloudflare", + user_agent="ua", + client_ip="::ffff:1.2.3.4", + path="/p", + method="GET", + referer="", + accept_language="en", + request_country="US", + request_asn=13335, + tls_fingerprint="ja3hash", + has_token=False, + token_outcome="absent", + final_action="allow", + enforcement_mode="observe", + signature_agent=None, + signature_input=None, + signature=None, + sec_fetch_mode=None, + sec_fetch_site=None, + sec_fetch_dest=None, + sec_fetch_user=None, + sec_ch_ua=None, + sec_ch_ua_mobile=None, + sec_ch_ua_platform=None, + accept=None, + host="example.com", + has_cookies=False, + header_names=["user-agent"], + query_length=0, + query_param_count=0, + query_suspicious=False, + accept_encoding=None, + http_protocol=None, + tls_version=None, + tls_cipher=None, + tls_client_hello_length=None, + tls_client_extensions_sha1=None, + as_organization=None, + client_tcp_rtt=None, + cdn_verified_bot_category=None, + request_priority=None, + tls_fingerprint_ja4=None, +) + + +def test_analytics_events_path_targets_the_relay_events_route(): + assert ANALYTICS_EVENTS_PATH == "/ingest/events" + + +async def test_posts_json_body_with_bearer_api_key_to_relay_url(): + with respx.mock: + route = respx.post(RELAY_URL).respond(status_code=202) + transport = HttpAnalyticsTransport(url=RELAY_URL, api_key="merchant-api-key") + + transport.emit(FIXTURE_EVENT) + await transport.flush() + + assert route.called + request = route.calls[0].request + assert request.method == "POST" + assert request.headers["authorization"] == "Bearer merchant-api-key" + assert request.headers["content-type"] == "application/json" + assert request.headers["user-agent"] == _get_sdk_user_agent() + assert json.loads(request.content) == { + "timestamp": "2026-04-29T12:00:00.000Z", + "request_id": "req-1", + "schema_version": 2, + "source_cdn": "cloudflare", + "user_agent": "ua", + "client_ip": "::ffff:1.2.3.4", + "path": "/p", + "method": "GET", + "referer": "", + "accept_language": "en", + "request_country": "US", + "request_asn": 13335, + "tls_fingerprint": "ja3hash", + "has_token": False, + "token_outcome": "absent", + "final_action": "allow", + "enforcement_mode": "observe", + "signature_agent": None, + "signature_input": None, + "signature": None, + "sec_fetch_mode": None, + "sec_fetch_site": None, + "sec_fetch_dest": None, + "sec_fetch_user": None, + "sec_ch_ua": None, + "sec_ch_ua_mobile": None, + "sec_ch_ua_platform": None, + "accept": None, + "host": "example.com", + "has_cookies": False, + "header_names": ["user-agent"], + "query_length": 0, + "query_param_count": 0, + "query_suspicious": False, + "accept_encoding": None, + "http_protocol": None, + "tls_version": None, + "tls_cipher": None, + "tls_client_hello_length": None, + "tls_client_extensions_sha1": None, + "as_organization": None, + "client_tcp_rtt": None, + "cdn_verified_bot_category": None, + "request_priority": None, + "tls_fingerprint_ja4": None, + } + + +async def test_does_not_raise_when_request_fails(): + with respx.mock: + respx.post(RELAY_URL).mock(side_effect=httpx.ConnectError("network down")) + transport = HttpAnalyticsTransport(url=RELAY_URL, api_key="t") + + transport.emit(FIXTURE_EVENT) # must not raise + await transport.flush() + + +async def test_does_not_raise_on_non_2xx_responses(): + with respx.mock: + respx.post(RELAY_URL).respond(status_code=500, text="err") + transport = HttpAnalyticsTransport(url=RELAY_URL, api_key="t") + + transport.emit(FIXTURE_EVENT) # must not raise + await transport.flush() + + +async def test_does_not_raise_on_non_http_errors(): + # A non-HTTPError raised on the request path (e.g. unexpected runtime error) must still + # be swallowed so the fire-and-forget task never surfaces an unhandled exception. + with respx.mock: + respx.post(RELAY_URL).mock(side_effect=ValueError("unexpected boom")) + transport = HttpAnalyticsTransport(url=RELAY_URL, api_key="t") + + transport.emit(FIXTURE_EVENT) # must not raise + await transport.flush() + + # No task should retain an unretrieved exception. + assert not transport._tasks + + +def test_emit_without_running_loop_is_a_noop(): + # No running event loop here (sync test) → emit silently skips scheduling. + transport = HttpAnalyticsTransport(url=RELAY_URL, api_key="t") + transport.emit(FIXTURE_EVENT) + + +def test_noop_transport_emit_never_throws(): + transport = NoopAnalyticsTransport() + assert transport.emit(FIXTURE_EVENT) is None + + +async def test_noop_transport_aclose_is_a_noop(): + transport = NoopAnalyticsTransport() + assert await transport.aclose() is None + + +async def test_aclose_drains_pending_emit(): + # Regression: an emit scheduled just before aclose() must be flushed, not dropped — a task + # must never be silently discarded by close. + with respx.mock: + route = respx.post(RELAY_URL).respond(status_code=202) + transport = HttpAnalyticsTransport(url=RELAY_URL, api_key="t") + + transport.emit(FIXTURE_EVENT) + await transport.aclose() + + assert route.called # event was flushed, not dropped + assert not transport._tasks + assert transport._client is None + + +async def test_aclose_closes_the_underlying_client(): + # The lazily-created client must be closed by aclose() — never left open to leak. + with respx.mock: + respx.post(RELAY_URL).respond(status_code=202) + transport = HttpAnalyticsTransport(url=RELAY_URL, api_key="t") + + transport.emit(FIXTURE_EVENT) + await transport.flush() # completes the send; client now exists + client = transport._client + assert client is not None and not client.is_closed + + await transport.aclose() + + assert client.is_closed + assert transport._client is None + + +async def test_emit_after_aclose_is_a_noop(): + with respx.mock: + route = respx.post(RELAY_URL).respond(status_code=202) + transport = HttpAnalyticsTransport(url=RELAY_URL, api_key="t") + + await transport.aclose() + transport.emit(FIXTURE_EVENT) # closed → no-op + + assert not transport._tasks + assert not route.called + + +async def test_aclose_is_bounded_when_send_hangs(): + # A send that never completes must not make aclose() hang: flush is bounded and stragglers + # are cancelled before the client is closed. + release = asyncio.Event() + + async def _hang(request: httpx.Request) -> httpx.Response: + await release.wait() # never released + return httpx.Response(202) + + with respx.mock: + respx.post(RELAY_URL).mock(side_effect=_hang) + transport = HttpAnalyticsTransport(url=RELAY_URL, api_key="t", flush_timeout=0.05) + + transport.emit(FIXTURE_EVENT) + # Let _send start and lazily create the client (it then blocks on `release`). + for _ in range(10): + if transport._client is not None: + break + await asyncio.sleep(0) + client = transport._client + assert client is not None + + # Overall guard well above flush_timeout: aclose must return on its own. + await asyncio.wait_for(transport.aclose(), timeout=1) + + assert not transport._tasks + assert transport._client is None + assert client.is_closed diff --git a/tests/customer/test_tokens.py b/tests/customer/test_tokens.py index ff4ceaf..8309eef 100644 --- a/tests/customer/test_tokens.py +++ b/tests/customer/test_tokens.py @@ -34,6 +34,14 @@ def create_mock_client(**kwargs: Any) -> httpx.AsyncClient: ) +async def test_create_async_client_sets_sdk_user_agent() -> None: + """The customer HTTP client (license.xml + token calls) carries the SDK User-Agent.""" + from supertab_connect._version import _get_sdk_user_agent + + async with _create_async_client() as client: + assert client.headers["User-Agent"] == _get_sdk_user_agent() + + def test_obtain_license_token_fetches_and_caches_token( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/merchant/test_client.py b/tests/merchant/test_client.py index 9832de6..abf0180 100644 --- a/tests/merchant/test_client.py +++ b/tests/merchant/test_client.py @@ -33,13 +33,13 @@ def _reset_supertab_connect_singleton(): def test_supertab_connect_returns_existing_instance_for_same_api_key(): - first = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123", enforcement=EnforcementMode.STRICT)) + first = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123", enforcement=EnforcementMode.ENFORCE)) second = SupertabConnect( - SupertabConnectConfig(api_key="sk_test_123", enforcement=EnforcementMode.SOFT, debug=True) + SupertabConnectConfig(api_key="sk_test_123", enforcement=EnforcementMode.OBSERVE, debug=True) ) assert first is second - assert second.enforcement is EnforcementMode.STRICT + assert second.enforcement is EnforcementMode.ENFORCE assert second.debug is False @@ -154,7 +154,7 @@ async def stub_verify_and_record_event(**kwargs): monkeypatch.setattr("supertab_connect.merchant.client.verify_and_record_event", stub_verify_and_record_event) - client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123", enforcement=EnforcementMode.STRICT)) + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123", enforcement=EnforcementMode.ENFORCE)) result = await client.handle_request( _make_request( { @@ -232,7 +232,7 @@ async def close_jwks(): async def test_handle_request_allows_missing_token_without_bot_detector(): - client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123", enforcement=EnforcementMode.STRICT)) + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123", enforcement=EnforcementMode.ENFORCE)) result = await client.handle_request(_make_request({"User-Agent": "Browser/1.0"})) @@ -243,7 +243,7 @@ async def test_handle_request_allows_missing_token_for_non_bot(): client = SupertabConnect( SupertabConnectConfig( api_key="sk_test_123", - enforcement=EnforcementMode.STRICT, + enforcement=EnforcementMode.ENFORCE, bot_detector=lambda request: False, ) ) @@ -253,11 +253,11 @@ async def test_handle_request_allows_missing_token_for_non_bot(): assert result == {"action": HandlerAction.ALLOW} -async def test_handle_request_blocks_bot_in_strict_mode(): +async def test_handle_request_blocks_bot_in_enforce_mode(): client = SupertabConnect( SupertabConnectConfig( api_key="sk_test_123", - enforcement=EnforcementMode.STRICT, + enforcement=EnforcementMode.ENFORCE, bot_detector=lambda request: True, ) ) @@ -269,11 +269,11 @@ async def test_handle_request_blocks_bot_in_strict_mode(): assert block_result["status"] == 401 -async def test_handle_request_signals_bot_in_soft_mode(): +async def test_handle_request_signals_bot_in_observe_mode(): client = SupertabConnect( SupertabConnectConfig( api_key="sk_test_123", - enforcement=EnforcementMode.SOFT, + enforcement=EnforcementMode.OBSERVE, bot_detector=lambda request: True, ) ) diff --git a/tests/merchant/test_client_analytics.py b/tests/merchant/test_client_analytics.py new file mode 100644 index 0000000..354479c --- /dev/null +++ b/tests/merchant/test_client_analytics.py @@ -0,0 +1,238 @@ +"""Tests for analytics emission wired into the high-level merchant client.""" + +import httpx +import pytest + +from supertab_connect.analytics.types import AnalyticsEvent, AnalyticsTransport, CdnRequestSignals +from supertab_connect.merchant.client import SupertabConnect +from supertab_connect.types import ( + EnforcementMode, + HandleRequestContext, + HandlerAction, + InvalidLicenseToken, + LicenseTokenInvalidReason, + SupertabConnectConfig, + ValidLicenseToken, +) + +from tests.merchant.constants import REQUEST_URL, SUPERTAB_BASE_URL + + +class RecordingTransport: + def __init__(self) -> None: + self.events: list[AnalyticsEvent] = [] + + def emit(self, event: AnalyticsEvent) -> None: + self.events.append(event) + + +class ThrowingTransport: + def emit(self, event: AnalyticsEvent) -> None: + raise RuntimeError("transport blew up") + + +class ClosableRecordingTransport(RecordingTransport): + def __init__(self) -> None: + super().__init__() + self.aclosed = False + + async def aclose(self) -> None: + self.aclosed = True + + +@pytest.fixture(autouse=True) +def _reset_singleton(): + SupertabConnect.reset_instance() + SupertabConnect.set_base_url(SUPERTAB_BASE_URL) + yield + SupertabConnect.reset_instance() + SupertabConnect.set_base_url(SUPERTAB_BASE_URL) + + +def _request(headers: dict[str, str] | None = None) -> httpx.Request: + return httpx.Request("GET", REQUEST_URL, headers=headers or {}) + + +def _client(transport: AnalyticsTransport, **config_kwargs) -> SupertabConnect: + return SupertabConnect( + SupertabConnectConfig(api_key="sk_test_123", analytics_transport=transport, **config_kwargs) + ) + + +def test_constructs_with_only_api_key(): + # Default transport is the Noop transport; construction must not require analytics config. + SupertabConnect(SupertabConnectConfig(api_key="sk_test_123")) + + +async def test_emits_observe_event_for_bot_without_token(): + transport = RecordingTransport() + client = _client(transport, enforcement=EnforcementMode.OBSERVE, bot_detector=lambda request: True) + + result = await client.handle_request( + _request({"User-Agent": "curl/8.0"}), HandleRequestContext(source_cdn="cloudflare") + ) + + assert result["action"] is HandlerAction.ALLOW + assert len(transport.events) == 1 + event = transport.events[0] + assert event.source_cdn == "cloudflare" + assert event.final_action == "observe" + assert event.enforcement_mode == "observe" + assert event.has_token is False + assert event.token_outcome == "absent" + + +async def test_emits_block_event_for_bot_without_token_in_enforce(): + transport = RecordingTransport() + client = _client(transport, enforcement=EnforcementMode.ENFORCE, bot_detector=lambda request: True) + + result = await client.handle_request(_request({"User-Agent": "curl/8.0"})) + + assert result["action"] is HandlerAction.BLOCK + assert transport.events[0].final_action == "block" + assert transport.events[0].token_outcome == "absent" + + +async def test_emits_allow_event_for_non_bot_without_token(): + transport = RecordingTransport() + client = _client(transport, enforcement=EnforcementMode.ENFORCE, bot_detector=lambda request: False) + + result = await client.handle_request(_request({"User-Agent": "Browser/1.0"})) + + assert result == {"action": HandlerAction.ALLOW} + assert transport.events[0].final_action == "allow" + assert transport.events[0].token_outcome == "absent" + + +async def test_emits_not_validated_for_token_in_disabled_mode(): + transport = RecordingTransport() + client = _client(transport, enforcement=EnforcementMode.DISABLED) + + result = await client.handle_request(_request({"Authorization": "License some-token"})) + + assert result == {"action": HandlerAction.ALLOW} + event = transport.events[0] + assert event.has_token is True + assert event.token_outcome == "not_validated" + assert event.final_action == "allow" + assert event.enforcement_mode == "disabled" + + +async def test_emits_valid_for_verified_token(monkeypatch): + async def stub_verify_and_record_event(**kwargs): + return ValidLicenseToken(license_id="lic_test_123", payload={}) + + monkeypatch.setattr("supertab_connect.merchant.client.verify_and_record_event", stub_verify_and_record_event) + transport = RecordingTransport() + client = _client(transport, enforcement=EnforcementMode.ENFORCE) + + result = await client.handle_request(_request({"Authorization": "License signed.jwt"})) + + assert result == {"action": HandlerAction.ALLOW} + assert transport.events[0].has_token is True + assert transport.events[0].token_outcome == "valid" + assert transport.events[0].final_action == "allow" + + +async def test_emits_mapped_outcome_for_invalid_token(monkeypatch): + async def stub_verify_and_record_event(**kwargs): + return InvalidLicenseToken( + reason=LicenseTokenInvalidReason.EXPIRED, + error="License token expired", + license_id="lic_test_123", + ) + + monkeypatch.setattr("supertab_connect.merchant.client.verify_and_record_event", stub_verify_and_record_event) + transport = RecordingTransport() + client = _client(transport, enforcement=EnforcementMode.ENFORCE) + + result = await client.handle_request(_request({"Authorization": "License signed.jwt"})) + + assert result["action"] is HandlerAction.BLOCK + assert transport.events[0].token_outcome == "expired" + assert transport.events[0].final_action == "block" + + +async def test_forwards_classification_signals_from_context(): + transport = RecordingTransport() + client = _client(transport, bot_detector=lambda request: True) + + await client.handle_request( + _request({"User-Agent": "curl/8.0"}), + HandleRequestContext( + source_cdn="fastly", + client_ip="1.2.3.4", + request_id="req-xyz", + request_country="DE", + request_asn=3320, + tls_fingerprint="abc123", + ), + ) + + event = transport.events[0] + assert event.source_cdn == "fastly" + assert event.client_ip == "::ffff:1.2.3.4" + assert event.request_id == "req-xyz" + assert event.request_country == "DE" + assert event.request_asn == 3320 + assert event.tls_fingerprint == "abc123" + + +async def test_forwards_cdn_signals_from_context(): + transport = RecordingTransport() + client = _client(transport, bot_detector=lambda request: True) + + await client.handle_request( + _request({"User-Agent": "curl/8.0"}), + HandleRequestContext( + source_cdn="cloudflare", + cdn_signals=CdnRequestSignals( + tls_version="TLSv1.3", + cdn_verified_bot_category="AI Assistant", + ), + ), + ) + + event = transport.events[0] + assert event.tls_version == "TLSv1.3" + assert event.cdn_verified_bot_category == "AI Assistant" + + +async def test_analytics_failure_does_not_break_request_handling(): + client = _client(ThrowingTransport(), bot_detector=lambda request: True) + + # A throwing transport must not propagate out of handle_request. + result = await client.handle_request(_request({"User-Agent": "curl/8.0"})) + + assert result["action"] is HandlerAction.ALLOW + + +async def test_no_event_emitted_without_context_still_works(): + transport = RecordingTransport() + client = _client(transport, bot_detector=lambda request: True) + + await client.handle_request(_request({"User-Agent": "curl/8.0"})) + + # Direct SDK invocation (no context) → source_cdn is None, request_id auto-generated. + event = transport.events[0] + assert event.source_cdn is None + assert event.request_id + + +async def test_aclose_closes_the_analytics_transport(): + transport = ClosableRecordingTransport() + client = _client(transport) + + async with client: + pass + + assert transport.aclosed is True + + +async def test_aclose_tolerates_emit_only_transport(): + # A custom transport implementing only `emit` (no `aclose`) must not break client teardown. + transport = RecordingTransport() + client = _client(transport) + + async with client: + pass # exit must not raise despite the transport lacking `aclose` diff --git a/tests/merchant/test_client_analytics_base_url.py b/tests/merchant/test_client_analytics_base_url.py new file mode 100644 index 0000000..6eb2a0e --- /dev/null +++ b/tests/merchant/test_client_analytics_base_url.py @@ -0,0 +1,74 @@ +"""Tests for analytics ingest base-URL resolution. + +The analytics relay targets the dedicated ingest service by default, independent of the +API base URL used for token acquisition / JWKS / verification. +""" + +import pytest + +from supertab_connect.analytics.transport import HttpAnalyticsTransport, NoopAnalyticsTransport +from supertab_connect.merchant.client import SupertabConnect +from supertab_connect.types import SupertabConnectConfig + +from tests.merchant.constants import SUPERTAB_BASE_URL + +DEFAULT_INGEST = "https://ingest-connect.supertab.co" + + +@pytest.fixture(autouse=True) +def _reset_singleton(): + SupertabConnect.reset_instance() + original_base_url = SupertabConnect.get_base_url() + original_analytics_base_url = SupertabConnect.get_analytics_base_url() + SupertabConnect.set_base_url(SUPERTAB_BASE_URL) + yield + # Restore the mutable class-level hosts so a mutating test can't leak into the next. + SupertabConnect.set_base_url(original_base_url) + SupertabConnect.set_analytics_base_url(original_analytics_base_url) + SupertabConnect.reset_instance() + + +def _relay_url(**config_kwargs) -> str: + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123", **config_kwargs)) + transport = client._analytics_transport + assert isinstance(transport, HttpAnalyticsTransport) + return transport._url + + +def test_disabled_analytics_uses_noop_transport(): + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123")) + assert isinstance(client._analytics_transport, NoopAnalyticsTransport) + + +def test_defaults_analytics_relay_to_ingest_host(): + assert _relay_url(analytics_enabled=True) == f"{DEFAULT_INGEST}/ingest/events" + + +def test_config_analytics_base_url_overrides_default(): + assert ( + _relay_url(analytics_enabled=True, analytics_base_url="https://ingest.example.com") + == "https://ingest.example.com/ingest/events" + ) + + +def test_set_analytics_base_url_overrides_default(): + SupertabConnect.set_analytics_base_url("https://static.example.com") + assert _relay_url(analytics_enabled=True) == "https://static.example.com/ingest/events" + + +def test_config_analytics_base_url_beats_set_analytics_base_url(): + SupertabConnect.set_analytics_base_url("https://static.example.com") + assert ( + _relay_url(analytics_enabled=True, analytics_base_url="https://perinstance.example.com") + == "https://perinstance.example.com/ingest/events" + ) + + +def test_analytics_host_independent_of_set_base_url(): + SupertabConnect.set_base_url("https://api.example.com") + assert _relay_url(analytics_enabled=True) == f"{DEFAULT_INGEST}/ingest/events" + + +def test_get_analytics_base_url_reflects_setter(): + SupertabConnect.set_analytics_base_url("https://x.example.com") + assert SupertabConnect.get_analytics_base_url() == "https://x.example.com" diff --git a/tests/merchant/test_jwks.py b/tests/merchant/test_jwks.py index 705dcd8..08f5dc3 100644 --- a/tests/merchant/test_jwks.py +++ b/tests/merchant/test_jwks.py @@ -3,6 +3,7 @@ import time from unittest.mock import patch +import httpx import pytest import respx @@ -14,6 +15,7 @@ aclose_http_client, clear_jwks_cache, fetch_platform_jwks, + refresh_platform_jwks_on_miss, ) from tests.merchant.constants import JWKS_URL, SUPERTAB_BASE_URL @@ -30,6 +32,19 @@ async def test_fetch_platform_jwks(jwks_response): assert route.call_count == 1 +async def test_fetch_platform_jwks_sends_sdk_user_agent(jwks_response): + """JWKS requests carry the SDK User-Agent header.""" + from supertab_connect._version import _get_sdk_user_agent + + with respx.mock: + route = respx.get(JWKS_URL).respond(json=jwks_response) + + clear_jwks_cache() + await fetch_platform_jwks(SUPERTAB_BASE_URL) + + assert route.calls[0].request.headers["User-Agent"] == _get_sdk_user_agent() + + async def test_fetch_platform_jwks_caches_result(jwks_response): """Second call returns cached JWKS without a network request.""" with respx.mock: @@ -69,6 +84,63 @@ async def test_clear_jwks_cache_forces_refetch(jwks_response): assert route.call_count == 2 +async def test_refresh_on_miss_refetches_when_not_throttled(jwks_response): + """A first miss forces a refetch even though the cache is still within its TTL.""" + with respx.mock: + route = respx.get(JWKS_URL).respond(json=jwks_response) + + await fetch_platform_jwks(SUPERTAB_BASE_URL) # warm the cache (fetch #1) + assert route.call_count == 1 + + await refresh_platform_jwks_on_miss(SUPERTAB_BASE_URL) # fetch #2, bypassing the TTL + + assert route.call_count == 2 + + +async def test_refresh_on_miss_is_throttled_within_cooldown(jwks_response): + """A second miss inside the cooldown must not hit the backend again — the abuse guard.""" + with respx.mock: + route = respx.get(JWKS_URL).respond(json=jwks_response) + + await fetch_platform_jwks(SUPERTAB_BASE_URL) # fetch #1 + await refresh_platform_jwks_on_miss(SUPERTAB_BASE_URL) # fetch #2 (first refresh) + result = await refresh_platform_jwks_on_miss(SUPERTAB_BASE_URL) # throttled, no fetch + + assert route.call_count == 2 + assert result == jwks_response # served from cache while throttled + + +async def test_refresh_on_miss_resumes_after_cooldown(jwks_response): + """Once the cooldown elapses, a genuine rotation can refresh again.""" + with respx.mock: + route = respx.get(JWKS_URL).respond(json=jwks_response) + + await fetch_platform_jwks(SUPERTAB_BASE_URL) # fetch #1 + await refresh_platform_jwks_on_miss(SUPERTAB_BASE_URL) # fetch #2 + + # Jump past the cooldown window. + later = time.monotonic() + jwks_module._JWKS_MIN_REFRESH_INTERVAL_SECONDS + 1 + with patch("supertab_connect.merchant.jwks.time.monotonic", return_value=later): + await refresh_platform_jwks_on_miss(SUPERTAB_BASE_URL) # fetch #3 + + assert route.call_count == 3 + + +async def test_refresh_on_miss_keeps_last_known_cache_on_failed_refetch(jwks_response): + """A failed forced refetch leaves the previously cached key set intact (fail-safe).""" + with respx.mock: + route = respx.get(JWKS_URL) + route.side_effect = [httpx.Response(200, json=jwks_response), httpx.Response(500)] + + await fetch_platform_jwks(SUPERTAB_BASE_URL) # fetch #1 (good) + with pytest.raises(httpx.HTTPError): + await refresh_platform_jwks_on_miss(SUPERTAB_BASE_URL) # fetch #2 (500) + + # Old keys survive, so a subsequent cached read still works without another fetch. + assert await fetch_platform_jwks(SUPERTAB_BASE_URL) == jwks_response + assert route.call_count == 2 + + def test_find_key_by_kid_returns_matching_key(): """Returns the key matching the given kid.""" jwks = {"keys": [{"kid": "key-1", "kty": "EC"}, {"kid": "key-2", "kty": "EC"}]} diff --git a/tests/merchant/test_status.py b/tests/merchant/test_status.py new file mode 100644 index 0000000..081fb73 --- /dev/null +++ b/tests/merchant/test_status.py @@ -0,0 +1,252 @@ +"""Tests for the self-report status endpoint and its challenge verification.""" + +import json +from datetime import UTC, datetime, timedelta +from typing import cast +from unittest.mock import patch + +import httpx +import jwt +import pytest +import respx +from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, PrivateFormat + +from supertab_connect.analytics.types import AnalyticsEvent +from supertab_connect.merchant.client import SupertabConnect +from supertab_connect.merchant.status import verify_status_challenge +from supertab_connect.types import ( + EnforcementMode, + HandlerAction, + RespondHandlerResult, + SupertabConnectConfig, +) + +from tests.merchant.constants import JWKS_URL, SUPERTAB_BASE_URL + +SITE_ORIGIN = "https://acme.com" +STATUS_URL = f"{SITE_ORIGIN}/.well-known/supertab/status" + + +class RecordingTransport: + def __init__(self) -> None: + self.events: list[AnalyticsEvent] = [] + + def emit(self, event: AnalyticsEvent) -> None: + self.events.append(event) + + +def _sign_challenge( + private_key, + *, + kid: str = "test-kid-1", + audience: str = SITE_ORIGIN, + purpose: str = "status-probe", + exp_delta: timedelta = timedelta(seconds=60), +) -> str: + now = datetime.now(UTC) + payload = {"aud": audience, "purpose": purpose, "iat": now, "exp": now + exp_delta} + pem_bytes = private_key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()) + return jwt.encode(payload, pem_bytes, algorithm="ES256", headers={"kid": kid}) + + +@pytest.fixture() +def sign_challenge(ec_key_pair): + private_key, _ = ec_key_pair + return lambda **kwargs: _sign_challenge(private_key, **kwargs) + + +@pytest.fixture(autouse=True) +def _reset_singleton(): + SupertabConnect.reset_instance() + SupertabConnect.set_base_url(SUPERTAB_BASE_URL) + yield + SupertabConnect.reset_instance() + SupertabConnect.set_base_url(SUPERTAB_BASE_URL) + + +def _status_request(token: str | None = None) -> httpx.Request: + headers = {"Authorization": f"Bearer {token}"} if token is not None else {} + return httpx.Request("GET", STATUS_URL, headers=headers) + + +# --- verify_status_challenge -------------------------------------------------- + + +async def test_accepts_valid_challenge(sign_challenge, mock_jwks): + token = sign_challenge() + assert await verify_status_challenge(token, expected_audience=SITE_ORIGIN, base_url=SUPERTAB_BASE_URL) is True + + +async def test_rejects_wrong_purpose(sign_challenge, mock_jwks): + token = sign_challenge(purpose="nope") + assert await verify_status_challenge(token, expected_audience=SITE_ORIGIN, base_url=SUPERTAB_BASE_URL) is False + + +async def test_rejects_wrong_audience(sign_challenge, mock_jwks): + token = sign_challenge(audience="https://evil.com") + assert await verify_status_challenge(token, expected_audience=SITE_ORIGIN, base_url=SUPERTAB_BASE_URL) is False + + +async def test_rejects_expired_challenge(sign_challenge, mock_jwks): + token = sign_challenge(exp_delta=timedelta(seconds=-30)) + assert await verify_status_challenge(token, expected_audience=SITE_ORIGIN, base_url=SUPERTAB_BASE_URL) is False + + +async def test_rejects_challenge_without_exp(ec_key_pair, mock_jwks): + # A "short-lived" challenge must carry an expiry; one signed without exp verifies otherwise + # (PyJWT validates exp only when present) and would be replayable forever. + private_key, _ = ec_key_pair + now = datetime.now(UTC) + payload = {"aud": SITE_ORIGIN, "purpose": "status-probe", "iat": now} # no exp + pem = private_key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()) + token = jwt.encode(payload, pem, algorithm="ES256", headers={"kid": "test-kid-1"}) + assert await verify_status_challenge(token, expected_audience=SITE_ORIGIN, base_url=SUPERTAB_BASE_URL) is False + + +async def test_rejects_challenge_without_iat(ec_key_pair, mock_jwks): + private_key, _ = ec_key_pair + now = datetime.now(UTC) + payload = {"aud": SITE_ORIGIN, "purpose": "status-probe", "exp": now + timedelta(seconds=60)} # no iat + pem = private_key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()) + token = jwt.encode(payload, pem, algorithm="ES256", headers={"kid": "test-kid-1"}) + assert await verify_status_challenge(token, expected_audience=SITE_ORIGIN, base_url=SUPERTAB_BASE_URL) is False + + +async def test_retries_after_jwks_refresh_on_key_rotation(sign_challenge, jwks_response): + # First fetch returns a stale key set (missing the signing kid); the second returns the fresh one. + stale = {"keys": [{**jwks_response["keys"][0], "kid": "old-kid"}]} + token = sign_challenge() + with respx.mock: + route = respx.get(JWKS_URL) + route.side_effect = [httpx.Response(200, json=stale), httpx.Response(200, json=jwks_response)] + result = await verify_status_challenge(token, expected_audience=SITE_ORIGIN, base_url=SUPERTAB_BASE_URL) + assert result is True + assert route.call_count == 2 + + +async def test_bogus_probes_with_rotating_kids_do_not_amplify_jwks_fetches(sign_challenge, jwks_response): + # An unauthenticated caller submitting challenges with rotating unknown kids must not bypass + # the cache and force a backend JWKS fetch per probe. After the first (cooldown-spending) + # refresh, further misses are throttled to the cache. + with respx.mock: + route = respx.get(JWKS_URL).respond(json=jwks_response) + + for i in range(5): + token = sign_challenge(kid=f"rotating-{i}") + result = await verify_status_challenge(token, expected_audience=SITE_ORIGIN, base_url=SUPERTAB_BASE_URL) + assert result is False # unknown kid never verifies + + # First probe: warm fetch + one refresh; every later probe is throttled → 2 fetches total. + assert route.call_count == 2 + + +# --- handle_request status branch -------------------------------------------- + + +async def test_status_branch_responds_200_with_payload(sign_challenge, mock_jwks): + transport = RecordingTransport() + client = SupertabConnect( + SupertabConnectConfig( + api_key="sk_test_123", + enforcement=EnforcementMode.ENFORCE, + analytics_enabled=True, + analytics_transport=transport, + ) + ) + + result = cast(RespondHandlerResult, await client.handle_request(_status_request(sign_challenge()))) + + assert result["action"] is HandlerAction.RESPOND + assert result["status"] == 200 + assert result["headers"]["Cache-Control"] == "no-store" + assert result["headers"]["Content-Type"] == "application/json" + + body = json.loads(result["body"]) + assert body["enforcement"] == EnforcementMode.ENFORCE.value + assert body["eventReporting"] is True + assert body["runtime"] is None + assert body["component"]["kind"] == "python-sdk" + assert isinstance(body["component"]["version"], str) + assert "sdkVersion" not in body + + # No analytics is emitted for a status probe. + assert transport.events == [] + + +async def test_status_branch_reports_runtime_and_event_reporting_off(sign_challenge, mock_jwks): + from supertab_connect.types import HandleRequestContext + + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123")) + result = cast( + RespondHandlerResult, + await client.handle_request(_status_request(sign_challenge()), HandleRequestContext(source_cdn="cloudflare")), + ) + + body = json.loads(result["body"]) + assert body["runtime"] == "cloudflare" + assert body["eventReporting"] is False + + +async def test_status_branch_reports_event_reporting_on_for_custom_transport(sign_challenge, mock_jwks): + # A custom transport emits regardless of the analytics_enabled flag, so eventReporting must + # report True even though analytics_enabled defaults to False — otherwise status would claim + # "eventReporting": false while events flow through the injected transport. + transport = RecordingTransport() + client = SupertabConnect( + SupertabConnectConfig( + api_key="sk_test_123", + analytics_transport=transport, + ) + ) + + result = cast(RespondHandlerResult, await client.handle_request(_status_request(sign_challenge()))) + + body = json.loads(result["body"]) + assert body["eventReporting"] is True + + +async def test_status_branch_responds_404_on_invalid_challenge(sign_challenge, mock_jwks): + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123")) + result = cast( + RespondHandlerResult, + await client.handle_request(_status_request(sign_challenge(purpose="nope"))), + ) + + assert result["action"] is HandlerAction.RESPOND + assert result["status"] == 404 + assert json.loads(result["body"]) == {"supertab": True} + assert result["headers"]["Cache-Control"] == "no-store" + + +async def test_status_branch_404_when_authorization_absent(): + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123")) + with patch("supertab_connect.merchant.client.verify_status_challenge") as verify: + result = cast(RespondHandlerResult, await client.handle_request(_status_request())) + + assert result["action"] is HandlerAction.RESPOND + assert result["status"] == 404 + verify.assert_not_called() + + +async def test_non_get_to_status_path_is_not_intercepted(): + # Only GET is the status probe; other methods to the same path must reach the application. + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123")) + request = httpx.Request("POST", STATUS_URL, headers={"Authorization": "Bearer whatever"}) + with patch("supertab_connect.merchant.client.verify_status_challenge") as verify: + result = await client.handle_request(request) + + assert result["action"] is not HandlerAction.RESPOND + verify.assert_not_called() + + +async def test_encoded_lookalike_path_is_not_intercepted(): + # request.url.path percent-decodes, but the probe must match the raw path only — an encoded + # look-alike (%2E for '.') is a different route and must flow through to the application. + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123")) + encoded_url = f"{SITE_ORIGIN}/%2Ewell-known/supertab/status" + request = httpx.Request("GET", encoded_url, headers={"Authorization": "Bearer whatever"}) + with patch("supertab_connect.merchant.client.verify_status_challenge") as verify: + result = await client.handle_request(request) + + assert result["action"] is not HandlerAction.RESPOND + verify.assert_not_called() diff --git a/uv.lock b/uv.lock index d8d5a11..69f902b 100644 --- a/uv.lock +++ b/uv.lock @@ -689,7 +689,7 @@ wheels = [ [[package]] name = "supertab-connect-sdk" -version = "0.1.2" +version = "1.0.0" source = { editable = "." } dependencies = [ { name = "httpx" },