diff --git a/connect/__init__.py b/connect/__init__.py index f0dfa1f..2064af6 100644 --- a/connect/__init__.py +++ b/connect/__init__.py @@ -2,20 +2,26 @@ from connect.customer.token import obtain_license_token from connect.exceptions import SupertabConnectError +from connect.merchant.bots import default_bot_detector +from connect.merchant.client import SupertabConnect from connect.merchant.license import verify_license_token from connect.types import ( EnforcementMode, - LicenseTokenInvalidReason, - LicenseTokenVerificationResult, + HandlerAction, + HandlerResult, RSLVerificationResult, + SupertabConnectConfig, ) __all__ = [ "EnforcementMode", - "LicenseTokenInvalidReason", - "LicenseTokenVerificationResult", + "HandlerAction", + "HandlerResult", "RSLVerificationResult", + "SupertabConnect", "SupertabConnectError", + "SupertabConnectConfig", + "default_bot_detector", "obtain_license_token", "verify_license_token", ] diff --git a/connect/_version.py b/connect/_version.py new file mode 100644 index 0000000..cde63cb --- /dev/null +++ b/connect/_version.py @@ -0,0 +1,17 @@ +"""Internal SDK version helpers.""" + +from functools import lru_cache +from importlib.metadata import PackageNotFoundError, version + +_PACKAGE_NAME = "supertab-connect-sdk" +_SDK_NAME = "supertab-connect-sdk-python" + + +@lru_cache(maxsize=1) +def _get_sdk_user_agent() -> str: + try: + package_version = version(_PACKAGE_NAME) + except PackageNotFoundError: + package_version = "unknown" + + return f"{_SDK_NAME}/{package_version}" diff --git a/connect/customer/__init__.py b/connect/customer/__init__.py index 226aa03..c5aa640 100644 --- a/connect/customer/__init__.py +++ b/connect/customer/__init__.py @@ -1 +1,5 @@ """Customer functionality for Supertab Connect.""" + +from connect.customer.token import obtain_license_token + +__all__ = ["obtain_license_token"] diff --git a/connect/customer/content_matcher.py b/connect/customer/content_matcher.py index 0827b49..06f2660 100644 --- a/connect/customer/content_matcher.py +++ b/connect/customer/content_matcher.py @@ -3,7 +3,7 @@ import urllib.parse from connect.common import debug_log -from connect.url_pattern import _score_path_pattern +from connect.url_pattern import score_path_pattern from connect.customer.content_parser import _ContentBlock @@ -48,7 +48,7 @@ def _find_best_matching_content( debug_log(debug, f"Exact match found: {block.url_pattern}") return block - specificity = _score_path_pattern(pattern_path, path) + specificity = score_path_pattern(pattern_path, path) if specificity > best_specificity: best_specificity = specificity best_match = block diff --git a/connect/merchant/__init__.py b/connect/merchant/__init__.py index e69de29..806420a 100644 --- a/connect/merchant/__init__.py +++ b/connect/merchant/__init__.py @@ -0,0 +1,9 @@ +"""Merchant-facing helpers for the Supertab Connect SDK.""" + +from connect.merchant.bots import default_bot_detector +from connect.merchant.client import SupertabConnect + +__all__ = [ + "default_bot_detector", + "SupertabConnect", +] diff --git a/connect/merchant/bots.py b/connect/merchant/bots.py new file mode 100644 index 0000000..31936f2 --- /dev/null +++ b/connect/merchant/bots.py @@ -0,0 +1,84 @@ +"""Merchant bot detection helpers.""" + +from httpx import Request + +_KNOWN_BOT_UA_SUBSTRINGS = ( + "chatgpt-user", + "perplexitybot", + "gptbot", + "anthropic-ai", + "ccbot", + "claude-web", + "claudebot", + "cohere-ai", + "youbot", + "diffbot", + "oai-searchbot", + "meta-externalagent", + "timpibot", + "amazonbot", + "bytespider", + "perplexity-user", + "googlebot", + "bot", + "curl", + "wget", +) + +_CHROMIUM_UA_SUBSTRINGS = ( + "chrome/", + "chromium/", + "edg/", + "edga/", + "opr/", + "opera/", + "samsungbrowser/", +) +_IOS_WEBKIT_BROWSER_UA_SUBSTRINGS = ("crios/", "edgios/", "fxios/") + + +def _browser_lacks_client_hints_support(lower_case_user_agent: str) -> bool: + """Return whether this UA is expected to omit the Sec-CH-UA header. + + Chromium-family browsers generally send Sec-CH-UA, but most browser + user-agent strings still include Mozilla and Safari compatibility tokens. + Keep this exception limited to browsers/platforms that do not reliably + support UA Client Hints: Safari, Firefox, and iOS WebKit browser wrappers. + """ + is_firefox = "firefox/" in lower_case_user_agent or "fxios/" in lower_case_user_agent + is_ios_webkit_browser = any(browser in lower_case_user_agent for browser in _IOS_WEBKIT_BROWSER_UA_SUBSTRINGS) + is_safari = ( + "safari/" in lower_case_user_agent + and "applewebkit/" in lower_case_user_agent + and not any(browser in lower_case_user_agent for browser in _CHROMIUM_UA_SUBSTRINGS) + ) + + return is_firefox or is_ios_webkit_browser or is_safari + + +def default_bot_detector(request: Request) -> bool: + user_agent = request.headers.get("user-agent", "") + accept = request.headers.get("accept", "") + sec_ch_ua = request.headers.get("sec-ch-ua") + accept_language = request.headers.get("accept-language") + + # 1. Basic substring check from known list + lower_case_user_agent = user_agent.lower() + bot_ua_match = any(bot in lower_case_user_agent for bot in _KNOWN_BOT_UA_SUBSTRINGS) + + # 2. Headless browser detection + headless_indicators = "headless" in lower_case_user_agent or "puppeteer" in lower_case_user_agent or not sec_ch_ua + is_browser_missing_sec_ch_ua = ( + "headless" not in lower_case_user_agent and "puppeteer" not in lower_case_user_agent and not sec_ch_ua + ) + + # 3. Suspicious header gaps — many bots omit these + missing_headers = not accept or not accept_language + + # Safari and Mozilla special case: allow if Sec-CH-UA is missing but UA matches Safari/Firefox patterns + if _browser_lacks_client_hints_support(lower_case_user_agent): + if headless_indicators and is_browser_missing_sec_ch_ua: + return False + + # Final decision: any strong indicator is sufficient + return bot_ua_match or headless_indicators or missing_headers diff --git a/connect/merchant/client.py b/connect/merchant/client.py new file mode 100644 index 0000000..cf1bab8 --- /dev/null +++ b/connect/merchant/client.py @@ -0,0 +1,182 @@ +"""High-level merchant client for Supertab Connect.""" + +from collections.abc import Mapping +from typing import ClassVar + +from httpx import Request + +from connect.merchant.events import aclose_http_client as aclose_events_http_client +from connect.merchant.license import ( + build_block_result, + build_signal_result, + verify_and_record_event, + verify_license_token, +) +from connect.merchant.jwks import aclose_http_client as aclose_jwks_http_client +from connect.types import ( + BotDetector, + EnforcementMode, + HandlerAction, + HandlerResult, + InvalidLicenseToken, + LicenseTokenInvalidReason, + RSLVerificationResult, + SupertabConnectConfig, +) + +_DEFAULT_BASE_URL = "https://api-connect.supertab.co" + + +class SupertabConnect: + _instance: ClassVar["SupertabConnect | None"] = None + _base_url: ClassVar[str] = _DEFAULT_BASE_URL + + def __new__(cls, config: SupertabConnectConfig, reset: bool = False) -> "SupertabConnect": + if not reset and cls._instance is not None: + if config.api_key != cls._instance.api_key: + raise ValueError( + "Cannot create a new instance with different configuration. " + "Use reset_instance to clear the existing instance." + ) + return cls._instance + + if reset and cls._instance is not None: + cls.reset_instance() + + return super().__new__(cls) + + def __init__(self, config: SupertabConnectConfig, reset: bool = False) -> None: + if getattr(self, "_initialized", False) and not reset: + return + + if not config.api_key: + raise ValueError("Missing required configuration: api_key is required") + + self.api_key = config.api_key + self.enforcement = config.enforcement + self.bot_detector = config.bot_detector + self.debug = config.debug + self._base_url_override = config.supertab_base_url + self._initialized = True + type(self)._instance = self + + @classmethod + def reset_instance(cls) -> None: + cls._instance = None + + @classmethod + def set_base_url(cls, url: str) -> None: + cls._base_url = url + + @classmethod + def get_base_url(cls) -> str: + return cls._base_url + + @property + def base_url(self) -> str: + return self._base_url_override or type(self)._base_url + + async def aclose(self) -> None: + await aclose_events_http_client() + await aclose_jwks_http_client() + + async def __aenter__(self) -> "SupertabConnect": + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + await self.aclose() + + @classmethod + async def verify( + cls, + *, + token: str, + resource_url: str, + base_url: str | None = None, + debug: bool = False, + ) -> RSLVerificationResult: + result = await verify_license_token( + token, + request_url=resource_url, + supertab_base_url=base_url or cls._base_url, + debug=debug, + ) + + if not isinstance(result, InvalidLicenseToken): + return RSLVerificationResult(valid=True) + + return RSLVerificationResult(valid=False, error=result.error) + + async def verify_and_record( + self, + *, + token: str, + resource_url: str, + user_agent: str = "unknown", + request_headers: Mapping[str, str] | None = None, + debug: bool | None = None, + ) -> RSLVerificationResult: + result = await verify_and_record_event( + token=token, + url=resource_url, + user_agent=user_agent, + supertab_base_url=self.base_url, + debug=self.debug if debug is None else debug, + api_key=self.api_key, + request_headers=request_headers, + ) + + if not isinstance(result, InvalidLicenseToken): + return RSLVerificationResult(valid=True) + + return RSLVerificationResult(valid=False, error=result.error) + + def _detect_bot(self, request: Request) -> bool: + detector: BotDetector | None = self.bot_detector + if detector is None: + return False + + return detector(request) + + async def handle_request(self, request: Request) -> HandlerResult: + 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] + url = str(request.url) + user_agent = request.headers.get("user-agent", "unknown") + + if token: + if self.enforcement is EnforcementMode.DISABLED: + return {"action": HandlerAction.ALLOW} + + verification = await verify_and_record_event( + token=token, + url=url, + user_agent=user_agent, + supertab_base_url=self.base_url, + debug=self.debug, + api_key=self.api_key, + request_headers=dict(request.headers.items()), + ) + if isinstance(verification, InvalidLicenseToken): + return build_block_result( + reason=verification.reason, + error=verification.error, + request_url=url, + ) + return {"action": HandlerAction.ALLOW} + + if not self._detect_bot(request): + return {"action": HandlerAction.ALLOW} + + if self.enforcement is EnforcementMode.STRICT: + return build_block_result( + reason=LicenseTokenInvalidReason.MISSING_TOKEN, + error="Authorization header missing or malformed", + request_url=url, + ) + if self.enforcement is EnforcementMode.SOFT: + return build_signal_result(url) + return {"action": HandlerAction.ALLOW} diff --git a/connect/merchant/events.py b/connect/merchant/events.py new file mode 100644 index 0000000..50f8439 --- /dev/null +++ b/connect/merchant/events.py @@ -0,0 +1,57 @@ +"""Merchant event recording helpers.""" + +from typing import Any + +import httpx + +from connect._version import _get_sdk_user_agent +from connect.common import debug_log, error_log + +_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() + return _http_client + + +async def aclose_http_client() -> None: + global _http_client + if _http_client is not None and not _http_client.is_closed: + await _http_client.aclose() + _http_client = None + + +async def record_event( + *, + api_key: str, + base_url: str, + event_name: str, + properties: dict[str, str], + license_id: str | None = None, + debug: bool = False, +) -> None: + """Record an analytics event without surfacing transport failures.""" + payload: dict[str, Any] = { + "event_name": event_name, + "properties": properties, + } + if license_id is not None: + payload["license_id"] = license_id + + try: + response = await _get_http_client().post( + f"{base_url.rstrip('/')}/events", + json=payload, + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + "User-Agent": _get_sdk_user_agent(), + }, + ) + if not response.is_success: + debug_log(debug, f"Failed to record event: {response.status_code}") + except httpx.HTTPError as error: + error_log(debug, f"Error recording event: {error}") diff --git a/connect/merchant/headers.py b/connect/merchant/headers.py new file mode 100644 index 0000000..5617f13 --- /dev/null +++ b/connect/merchant/headers.py @@ -0,0 +1,29 @@ +"""Header helpers for merchant event analytics.""" + +from collections.abc import Mapping + +_DENIED_HEADERS = { + "authorization", + "cookie", + "set-cookie", + "proxy-authorization", + "x-api-key", + "x-amz-security-token", + "user-agent", + "x-license-auth", +} + + +def to_event_properties( + headers: Mapping[str, str], +) -> dict[str, str]: + """Convert request headers into event properties.""" + result: dict[str, str] = {} + + for key, value in headers.items(): + normalized_key = key.lower() + if normalized_key in _DENIED_HEADERS: + continue + result[f"h_{normalized_key}"] = value + + return result diff --git a/connect/merchant/jwks.py b/connect/merchant/jwks.py index 61511dc..2fb3d5b 100644 --- a/connect/merchant/jwks.py +++ b/connect/merchant/jwks.py @@ -21,6 +21,13 @@ def _get_http_client() -> httpx.AsyncClient: return _http_client +async def aclose_http_client() -> None: + global _http_client + if _http_client is not None and not _http_client.is_closed: + await _http_client.aclose() + _http_client = None + + async def fetch_platform_jwks(base_url: str, *, debug: bool = False) -> dict[str, Any]: """Fetch the platform JWKS from the Supertab well-known endpoint. @@ -54,7 +61,7 @@ def clear_jwks_cache() -> None: _jwks_cache.clear() -def find_key_by_kid(jwks: dict[str, Any], kid: str | None) -> dict[str, Any]: +def _find_key_by_kid(jwks: dict[str, Any], kid: str | None) -> dict[str, Any]: """Find a key in the JWKS by key ID. Raises JwksKeyNotFoundError if no matching key is found. diff --git a/connect/merchant/license.py b/connect/merchant/license.py index 75b6969..c22c141 100644 --- a/connect/merchant/license.py +++ b/connect/merchant/license.py @@ -1,17 +1,23 @@ """License token verification for the Supertab Connect SDK.""" import re -from typing import Any, cast +from collections.abc import Mapping +from typing import cast from urllib.parse import urlparse import jwt import jwt.algorithms from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePublicKey +from connect._version import _get_sdk_user_agent from connect.common import debug_log, error_log from connect.exceptions import JwksKeyNotFoundError -from connect.merchant.jwks import clear_jwks_cache, fetch_platform_jwks, find_key_by_kid +from connect.merchant.events import record_event +from connect.merchant.headers import to_event_properties +from connect.merchant.jwks import _find_key_by_kid, clear_jwks_cache, fetch_platform_jwks from connect.types import ( + AllowHandlerResult, + BlockHandlerResult, HandlerAction, InvalidLicenseToken, LicenseTokenInvalidReason, @@ -36,7 +42,7 @@ def _audience_matches(request_url: str, audience: str) -> bool: return request_url.startswith(normalized_aud + "/") -def _generate_license_link(request_url: str) -> str: +def generate_license_link(request_url: str) -> str: try: parsed = urlparse(request_url) if not parsed.scheme or not parsed.netloc: @@ -181,7 +187,7 @@ async def verify_license_token( ) try: - jwk_key = find_key_by_kid(jwks, header.get("kid")) + jwk_key = _find_key_by_kid(jwks, header.get("kid")) public_key = cast(EllipticCurvePublicKey, jwt.algorithms.ECAlgorithm.from_jwk(jwk_key)) verified_payload = jwt.decode( license_token, @@ -227,11 +233,11 @@ def build_block_result( reason: LicenseTokenInvalidReason, error: str, request_url: str, -) -> dict[str, Any]: +) -> BlockHandlerResult: """Build a block response with appropriate status code and headers.""" rsl_error, status = _reason_to_rsl_error(reason) error_description = _sanitize_header_value(error) - license_link = _generate_license_link(request_url) + license_link = generate_license_link(request_url) return { "action": HandlerAction.BLOCK, @@ -245,9 +251,9 @@ def build_block_result( } -def build_signal_result(request_url: str) -> dict[str, Any]: +def build_signal_result(request_url: str) -> AllowHandlerResult: """Build a soft enforcement signal response with license link headers.""" - license_link = _generate_license_link(request_url) + license_link = generate_license_link(request_url) return { "action": HandlerAction.ALLOW, "headers": { @@ -256,3 +262,39 @@ def build_signal_result(request_url: str) -> dict[str, Any]: "X-RSL-Reason": "missing", }, } + + +async def verify_and_record_event( + *, + token: str, + url: str, + user_agent: str, + supertab_base_url: str, + debug: bool, + api_key: str, + request_headers: Mapping[str, str] | None = None, +) -> LicenseTokenVerificationResult: + verification = await verify_license_token( + token, + request_url=url, + supertab_base_url=supertab_base_url, + debug=debug, + ) + + await record_event( + api_key=api_key, + base_url=supertab_base_url, + event_name="license_used" if isinstance(verification, ValidLicenseToken) else verification.reason, + properties={ + "page_url": url, + "user_agent": user_agent, + "sdk_user_agent": _get_sdk_user_agent(), + "verification_status": "valid" if verification.valid else "invalid", + "verification_reason": "success" if isinstance(verification, ValidLicenseToken) else verification.reason, + **to_event_properties(request_headers or {}), + }, + license_id=verification.license_id, + debug=debug, + ) + + return verification diff --git a/connect/types.py b/connect/types.py index 4ff4858..8bc7299 100644 --- a/connect/types.py +++ b/connect/types.py @@ -1,8 +1,11 @@ """Core types for the Supertab Connect SDK.""" +from collections.abc import Callable from dataclasses import dataclass, field from enum import StrEnum -from typing import Any +from typing import Any, Literal, NotRequired, TypeAlias, TypedDict + +from httpx import Request class EnforcementMode(StrEnum): @@ -28,6 +31,33 @@ class HandlerAction(StrEnum): BLOCK = "block" +BotDetector: TypeAlias = Callable[[Request], bool] + + +@dataclass(frozen=True) +class SupertabConnectConfig: + api_key: str + enforcement: EnforcementMode = EnforcementMode.STRICT + supertab_base_url: str | None = None + bot_detector: BotDetector | None = None + debug: bool = False + + +class AllowHandlerResult(TypedDict): + action: Literal[HandlerAction.ALLOW] + headers: NotRequired[dict[str, str]] + + +class BlockHandlerResult(TypedDict): + action: Literal[HandlerAction.BLOCK] + status: int + body: str + headers: dict[str, str] + + +HandlerResult: TypeAlias = AllowHandlerResult | BlockHandlerResult + + @dataclass(frozen=True) class ValidLicenseToken: valid: bool = field(default=True, init=False) diff --git a/connect/url_pattern.py b/connect/url_pattern.py index c3810aa..d20c693 100644 --- a/connect/url_pattern.py +++ b/connect/url_pattern.py @@ -3,7 +3,7 @@ import re -def _score_path_pattern(pattern: str, path: str) -> int: +def score_path_pattern(pattern: str, path: str) -> int: """Return a specificity score for a matching path pattern, or ``-1``.""" anchored = pattern.endswith("$") normalized_pattern = pattern[:-1] if anchored else pattern diff --git a/examples/merchant_handle_request.py b/examples/merchant_handle_request.py new file mode 100644 index 0000000..9d1db4f --- /dev/null +++ b/examples/merchant_handle_request.py @@ -0,0 +1,51 @@ +"""Example of using `SupertabConnect.handle_request` for request enforcement.""" + +import asyncio +import logging + +import httpx + +from connect import EnforcementMode, HandlerAction, SupertabConnect, SupertabConnectConfig + +logging.basicConfig(level=logging.DEBUG) + +REQUEST_URL = "https://example.com/premium/article" + + +async def main() -> None: + client = SupertabConnect( + SupertabConnectConfig( + api_key="your_api_key", + enforcement=EnforcementMode.STRICT, + debug=True, + ) + ) + + request = httpx.Request( + "GET", + REQUEST_URL, + headers={ + "Authorization": "License your.jwt.token", + "User-Agent": "Mozilla/5.0", + "Accept": "text/html", + "Accept-Language": "en-US", + "Sec-CH-UA": '"Chromium";v="123"', + }, + ) + + async with client: + result = await client.handle_request(request) + + if result["action"] is HandlerAction.BLOCK: + print("BLOCK request") + print(result["status"]) # type: ignore + print(result["headers"]["WWW-Authenticate"]) + return + + print("ALLOW request") + if "headers" in result: + print(result["headers"]) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/merchant_verify_and_record_event.py b/examples/merchant_verify_and_record_event.py new file mode 100644 index 0000000..29808c8 --- /dev/null +++ b/examples/merchant_verify_and_record_event.py @@ -0,0 +1,46 @@ +"""Example of custom request handling with `SupertabConnect.verify_and_record`.""" + +import asyncio +import logging + +from connect import EnforcementMode, SupertabConnect, SupertabConnectConfig + +logging.basicConfig(level=logging.DEBUG) + +REQUEST_URL = "https://example.com/premium/article" + + +async def main() -> None: + client = SupertabConnect( + SupertabConnectConfig( + api_key="your_api_key", + enforcement=EnforcementMode.SOFT, + debug=True, + ) + ) + + token = "your.jwt.token" + user_agent = "Mozilla/5.0" + request_headers = { + "Accept": "text/html", + "Accept-Language": "en-US", + "X-Forwarded-For": "203.0.113.1", + } + + async with client: + result = await client.verify_and_record( + token=token, + resource_url=REQUEST_URL, + user_agent=user_agent, + request_headers=request_headers, + ) + + if not result.valid: + print(f"DENY access: {result.error}") + return + + print("ALLOW access") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/customer/test_content_matcher.py b/tests/customer/test_content_matcher.py index 8a1bf5e..336fb2b 100644 --- a/tests/customer/test_content_matcher.py +++ b/tests/customer/test_content_matcher.py @@ -5,7 +5,7 @@ _find_best_matching_content, ) from connect.customer.content_parser import _parse_content_elements -from connect.url_pattern import _score_path_pattern +from connect.url_pattern import score_path_pattern from tests.customer.conftest import SAMPLE_XML @@ -167,15 +167,15 @@ def test_find_best_matching_content_returns_none_for_empty_blocks() -> None: ) def test_score_path_pattern_handles_all_cases(pattern: str, path: str, expected: int) -> None: """Pattern scoring returns expected specificity for each pattern/path pair.""" - assert _score_path_pattern(pattern, path) == expected + assert score_path_pattern(pattern, path) == expected def test_score_path_pattern_prefers_more_literal_characters() -> None: """More literal characters in the pattern yield a higher score.""" path = "/content/news/article" - broad = _score_path_pattern("/*", path) - mid = _score_path_pattern("/content/*", path) - specific = _score_path_pattern("/content/*/article", path) + broad = score_path_pattern("/*", path) + mid = score_path_pattern("/content/*", path) + specific = score_path_pattern("/content/*/article", path) assert broad < mid < specific diff --git a/tests/merchant/test_bots.py b/tests/merchant/test_bots.py new file mode 100644 index 0000000..cb38474 --- /dev/null +++ b/tests/merchant/test_bots.py @@ -0,0 +1,68 @@ +"""Tests for merchant bot detection helpers.""" + +import httpx + +from connect.merchant.bots import default_bot_detector + +from tests.merchant.constants import REQUEST_URL + + +def _make_request(headers: dict[str, str]) -> httpx.Request: + return httpx.Request("GET", REQUEST_URL, headers=headers) + + +def test_default_bot_detector_flags_known_bot_user_agents(): + request = _make_request( + { + "User-Agent": "GPTBot/1.0", + "Accept": "text/html", + "Accept-Language": "en-US", + "Sec-CH-UA": '"Chromium";v="123"', + } + ) + + assert default_bot_detector(request) is True + + +def test_default_bot_detector_flags_missing_headers(): + request = _make_request( + { + "User-Agent": "CustomBrowser/1.0", + "Sec-CH-UA": '"Chromium";v="123"', + } + ) + + assert default_bot_detector(request) is True + + +def test_default_bot_detector_flags_missing_sec_ch_ua_for_non_safari_agents(): + request = _make_request( + { + "User-Agent": "CustomBrowser/1.0", + "Accept": "text/html", + "Accept-Language": "en-US", + } + ) + + assert default_bot_detector(request) is True + + +def test_default_bot_detector_safari_mozilla_exception_returns_false(): + request = _make_request( + { + "User-Agent": ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15" + ), + "Accept": "text/html", + "Accept-Language": "en-US", + } + ) + + assert default_bot_detector(request) is False + + +def test_default_bot_detector_flags_completely_missing_headers(): + request = _make_request({}) + + assert default_bot_detector(request) is True diff --git a/tests/merchant/test_client.py b/tests/merchant/test_client.py new file mode 100644 index 0000000..7293515 --- /dev/null +++ b/tests/merchant/test_client.py @@ -0,0 +1,298 @@ +"""Tests for the high-level merchant client.""" + +from typing import Any, cast + +import httpx +import pytest + +from connect.merchant.client import SupertabConnect +from connect.types import ( + BlockHandlerResult, + EnforcementMode, + HandlerAction, + InvalidLicenseToken, + LicenseTokenInvalidReason, + SupertabConnectConfig, + ValidLicenseToken, +) + +from tests.merchant.constants import REQUEST_URL, SUPERTAB_BASE_URL + + +def _make_request(headers: dict[str, str] | None = None, url: str = REQUEST_URL) -> httpx.Request: + return httpx.Request("GET", url, headers=headers or {}) + + +@pytest.fixture(autouse=True) +def _reset_supertab_connect_singleton(): + SupertabConnect.reset_instance() + SupertabConnect.set_base_url(SUPERTAB_BASE_URL) + yield + SupertabConnect.reset_instance() + SupertabConnect.set_base_url(SUPERTAB_BASE_URL) + + +def test_supertab_connect_returns_existing_instance_for_same_api_key(): + first = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123", enforcement=EnforcementMode.STRICT)) + second = SupertabConnect( + SupertabConnectConfig(api_key="sk_test_123", enforcement=EnforcementMode.SOFT, debug=True) + ) + + assert first is second + assert second.enforcement is EnforcementMode.STRICT + assert second.debug is False + + +def test_supertab_connect_raises_for_different_api_key_without_reset(): + SupertabConnect(SupertabConnectConfig(api_key="sk_test_123")) + + with pytest.raises(ValueError, match="Cannot create a new instance with different configuration"): + SupertabConnect(SupertabConnectConfig(api_key="sk_test_456")) + + +def test_supertab_connect_reset_replaces_singleton(): + first = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123")) + second = SupertabConnect(SupertabConnectConfig(api_key="sk_test_456"), reset=True) + + assert first is not second + assert second.api_key == "sk_test_456" + + +def test_base_url_property_uses_instance_override_or_class_default(): + SupertabConnect.set_base_url("https://class-default.example") + + default_client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123")) + override_client = SupertabConnect( + SupertabConnectConfig(api_key="sk_test_456", supertab_base_url="https://instance-override.example"), + reset=True, + ) + + assert default_client.base_url == "https://class-default.example" + assert override_client.base_url == "https://instance-override.example" + + +async def test_verify_uses_class_base_url(monkeypatch): + captured: dict[str, Any] = {} + + async def stub_verify_license_token(token: str, *, request_url: str, supertab_base_url: str, debug: bool = False): + captured.update( + { + "token": token, + "request_url": request_url, + "supertab_base_url": supertab_base_url, + "debug": debug, + } + ) + return ValidLicenseToken(license_id="lic_test_123", payload={}) + + monkeypatch.setattr("connect.merchant.client.verify_license_token", stub_verify_license_token) + SupertabConnect.set_base_url("https://override.example") + + result = await SupertabConnect.verify(token="signed.jwt", resource_url=REQUEST_URL, debug=True) + + assert result.valid is True + assert result.error is None + assert captured == { + "token": "signed.jwt", + "request_url": REQUEST_URL, + "supertab_base_url": "https://override.example", + "debug": True, + } + + +async def test_verify_and_record_uses_instance_base_url_override(monkeypatch): + captured: dict[str, Any] = {} + + async def stub_verify_and_record_event(**kwargs): + captured.update(kwargs) + return ValidLicenseToken(license_id="lic_test_123", payload={}) + + monkeypatch.setattr("connect.merchant.client.verify_and_record_event", stub_verify_and_record_event) + + client = SupertabConnect( + SupertabConnectConfig( + api_key="sk_test_123", + supertab_base_url="https://merchant-override.example", + debug=True, + ) + ) + result = await client.verify_and_record( + token="signed.jwt", + resource_url=REQUEST_URL, + user_agent="TestAgent/1.0", + request_headers={"Accept": "text/html"}, + ) + + assert result.valid is True + assert captured["supertab_base_url"] == "https://merchant-override.example" + assert captured["api_key"] == "sk_test_123" + assert captured["request_headers"] == {"Accept": "text/html"} + + +async def test_handle_request_allows_token_when_enforcement_disabled(monkeypatch): + async def fail_verify_and_record_event(**kwargs): + raise AssertionError(f"verify_and_record_event should not be called: {kwargs}") + + monkeypatch.setattr("connect.merchant.client.verify_and_record_event", fail_verify_and_record_event) + + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123", enforcement=EnforcementMode.DISABLED)) + result = await client.handle_request(_make_request({"Authorization": "License signed.jwt"})) + + assert result == {"action": HandlerAction.ALLOW} + + +async def test_handle_request_blocks_invalid_token(monkeypatch): + captured: dict[str, Any] = {} + + async def stub_verify_and_record_event(**kwargs): + captured.update(kwargs) + return InvalidLicenseToken( + reason=LicenseTokenInvalidReason.INVALID_AUDIENCE, + error="The license does not grant access to this resource", + license_id="lic_test_123", + ) + + monkeypatch.setattr("connect.merchant.client.verify_and_record_event", stub_verify_and_record_event) + + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123", enforcement=EnforcementMode.STRICT)) + result = await client.handle_request( + _make_request( + { + "Authorization": "License signed.jwt", + "User-Agent": "Browser/1.0", + "Accept": "text/html", + } + ) + ) + + assert result["action"] is HandlerAction.BLOCK + block_result = cast(BlockHandlerResult, result) + assert block_result["status"] == 403 + assert captured["request_headers"]["authorization"] == "License signed.jwt" + + +async def test_handle_request_allows_valid_token(monkeypatch): + async def stub_verify_and_record_event(**kwargs): + return ValidLicenseToken(license_id="lic_test_123", payload={}) + + monkeypatch.setattr("connect.merchant.client.verify_and_record_event", stub_verify_and_record_event) + + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123")) + result = await client.handle_request( + _make_request( + { + "Authorization": "License signed.jwt", + "User-Agent": "Browser/1.0", + } + ) + ) + + assert result == {"action": HandlerAction.ALLOW} + + +async def test_handle_request_accepts_lowercase_scheme_and_whitespace(monkeypatch): + captured: dict[str, Any] = {} + + async def stub_verify_and_record_event(**kwargs): + captured.update(kwargs) + return ValidLicenseToken(license_id="lic_test_123", payload={}) + + monkeypatch.setattr("connect.merchant.client.verify_and_record_event", stub_verify_and_record_event) + + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123")) + result = await client.handle_request( + _make_request( + { + "Authorization": "license\t signed.jwt", + "User-Agent": "Browser/1.0", + } + ) + ) + + assert result == {"action": HandlerAction.ALLOW} + assert captured["token"] == "signed.jwt" + + +async def test_supertab_connect_async_context_manager_closes_http_clients(monkeypatch): + called: list[str] = [] + + async def close_events(): + called.append("events") + + async def close_jwks(): + called.append("jwks") + + monkeypatch.setattr("connect.merchant.client.aclose_events_http_client", close_events) + monkeypatch.setattr("connect.merchant.client.aclose_jwks_http_client", close_jwks) + + async with SupertabConnect(SupertabConnectConfig(api_key="sk_test_123")): + pass + + assert called == ["events", "jwks"] + + +async def test_handle_request_allows_missing_token_without_bot_detector(): + client = SupertabConnect(SupertabConnectConfig(api_key="sk_test_123", enforcement=EnforcementMode.STRICT)) + + result = await client.handle_request(_make_request({"User-Agent": "Browser/1.0"})) + + assert result == {"action": HandlerAction.ALLOW} + + +async def test_handle_request_allows_missing_token_for_non_bot(): + client = SupertabConnect( + SupertabConnectConfig( + api_key="sk_test_123", + enforcement=EnforcementMode.STRICT, + bot_detector=lambda request: False, + ) + ) + + result = await client.handle_request(_make_request({"User-Agent": "Browser/1.0"})) + + assert result == {"action": HandlerAction.ALLOW} + + +async def test_handle_request_blocks_bot_in_strict_mode(): + client = SupertabConnect( + SupertabConnectConfig( + api_key="sk_test_123", + enforcement=EnforcementMode.STRICT, + bot_detector=lambda request: True, + ) + ) + + result = await client.handle_request(_make_request({"User-Agent": "curl/8.0"})) + + assert result["action"] is HandlerAction.BLOCK + block_result = cast(BlockHandlerResult, result) + assert block_result["status"] == 401 + + +async def test_handle_request_signals_bot_in_soft_mode(): + client = SupertabConnect( + SupertabConnectConfig( + api_key="sk_test_123", + enforcement=EnforcementMode.SOFT, + bot_detector=lambda request: True, + ) + ) + + result = await client.handle_request(_make_request({"User-Agent": "curl/8.0"})) + + assert result["action"] is HandlerAction.ALLOW + assert result["headers"]["X-RSL-Status"] == "token_required" + + +async def test_handle_request_allows_bot_in_disabled_mode(): + client = SupertabConnect( + SupertabConnectConfig( + api_key="sk_test_123", + enforcement=EnforcementMode.DISABLED, + bot_detector=lambda request: True, + ) + ) + + result = await client.handle_request(_make_request({"User-Agent": "curl/8.0"})) + + assert result == {"action": HandlerAction.ALLOW} diff --git a/tests/merchant/test_events.py b/tests/merchant/test_events.py new file mode 100644 index 0000000..e8bab4f --- /dev/null +++ b/tests/merchant/test_events.py @@ -0,0 +1,91 @@ +"""Tests for merchant event recording helpers.""" + +import json +import logging + +import httpx +import respx + +import connect.merchant.events as events_module +from connect.merchant.events import record_event +from connect.merchant.events import aclose_http_client + +from tests.merchant.constants import SUPERTAB_BASE_URL + +EVENTS_URL = f"{SUPERTAB_BASE_URL}/events" + + +async def test_record_event_posts_expected_payload(monkeypatch): + monkeypatch.setattr("connect.merchant.events._get_sdk_user_agent", lambda: "sdk-test/1.2.3") + + with respx.mock: + route = respx.post(EVENTS_URL).respond(status_code=201, json={"ok": True}) + + await record_event( + api_key="sk_test_123", + base_url=SUPERTAB_BASE_URL, + event_name="license_used", + properties={"page_url": "https://example.com/premium/article"}, + license_id="lic_test_123", + ) + + request = route.calls[0].request + assert request.headers["Authorization"] == "Bearer sk_test_123" + assert request.headers["Content-Type"] == "application/json" + assert request.headers["User-Agent"] == "sdk-test/1.2.3" + assert json.loads(request.content) == { + "event_name": "license_used", + "license_id": "lic_test_123", + "properties": {"page_url": "https://example.com/premium/article"}, + } + + +async def test_record_event_logs_non_2xx_responses(caplog): + with respx.mock: + respx.post(EVENTS_URL).respond(status_code=500) + + with caplog.at_level(logging.DEBUG, logger="connect.common"): + await record_event( + api_key="sk_test_123", + base_url=SUPERTAB_BASE_URL, + event_name="license_used", + properties={}, + debug=True, + ) + + assert "Failed to record event: 500" in caplog.text + + +async def test_record_event_swallows_request_failures(caplog): + request = httpx.Request("POST", EVENTS_URL) + + with respx.mock: + respx.post(EVENTS_URL).mock(side_effect=httpx.ConnectError("boom", request=request)) + + with caplog.at_level(logging.ERROR, logger="connect.common"): + await record_event( + api_key="sk_test_123", + base_url=SUPERTAB_BASE_URL, + event_name="license_used", + properties={}, + debug=True, + ) + + assert "Error recording event:" in caplog.text + + +async def test_aclose_http_client_resets_client(monkeypatch): + called = {"aclose": 0} + + class DummyClient: + is_closed = False + + async def aclose(self): + called["aclose"] += 1 + + monkeypatch.setattr("connect.merchant.events._http_client", DummyClient()) + + await aclose_http_client() + + assert called["aclose"] == 1 + assert events_module._http_client is None diff --git a/tests/merchant/test_headers.py b/tests/merchant/test_headers.py new file mode 100644 index 0000000..d5569f4 --- /dev/null +++ b/tests/merchant/test_headers.py @@ -0,0 +1,63 @@ +"""Tests for merchant event header mapping.""" + +from connect.merchant.headers import to_event_properties + + +def test_to_event_properties_lowercases_keys_and_prefixes_them(): + result = to_event_properties( + { + "Accept-Language": "en-US", + "X-Custom": "value", + } + ) + + assert result == { + "h_accept-language": "en-US", + "h_x-custom": "value", + } + + +def test_to_event_properties_drops_denied_headers_regardless_of_casing(): + result = to_event_properties( + { + "Authorization": "License abc123", + "COOKIE": "session=xyz", + "Set-Cookie": "foo=bar", + "Proxy-Authorization": "Basic xxx", + "X-API-Key": "sk_123", + "X-Amz-Security-Token": "amz-token", + "User-Agent": "GPTBot/1.0", + "X-License-Auth": "cf-request-id", + "Accept": "application/json", + } + ) + + assert result == {"h_accept": "application/json"} + + +def test_to_event_properties_keeps_client_ip_headers(): + result = to_event_properties( + { + "X-Forwarded-For": "203.0.113.1", + "X-Real-IP": "203.0.113.2", + "CF-Connecting-IP": "203.0.113.3", + "True-Client-IP": "203.0.113.4", + } + ) + + assert result == { + "h_x-forwarded-for": "203.0.113.1", + "h_x-real-ip": "203.0.113.2", + "h_cf-connecting-ip": "203.0.113.3", + "h_true-client-ip": "203.0.113.4", + } + + +def test_to_event_properties_returns_empty_dict_for_empty_input(): + assert to_event_properties({}) == {} + + +def test_to_event_properties_preserves_values_verbatim(): + result = to_event_properties({"X-Custom": " value with spaces "}) + + assert result["h_x-custom"] == " value with spaces " diff --git a/tests/merchant/test_jwks.py b/tests/merchant/test_jwks.py index 0ed1090..84dcd3a 100644 --- a/tests/merchant/test_jwks.py +++ b/tests/merchant/test_jwks.py @@ -6,8 +6,15 @@ import pytest import respx +import connect.merchant.jwks as jwks_module from connect.exceptions import JwksKeyNotFoundError -from connect.merchant.jwks import JWKS_CACHE_TTL_SECONDS, clear_jwks_cache, fetch_platform_jwks, find_key_by_kid +from connect.merchant.jwks import ( + JWKS_CACHE_TTL_SECONDS, + _find_key_by_kid, + aclose_http_client, + clear_jwks_cache, + fetch_platform_jwks, +) from tests.merchant.constants import JWKS_URL, SUPERTAB_BASE_URL @@ -64,7 +71,7 @@ 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"}]} - result = find_key_by_kid(jwks, "key-2") + result = _find_key_by_kid(jwks, "key-2") assert result == {"kid": "key-2", "kty": "EC"} @@ -74,10 +81,27 @@ def test_find_key_by_kid_raises_on_missing_kid(): jwks = {"keys": [{"kid": "key-1", "kty": "EC"}]} with pytest.raises(JwksKeyNotFoundError, match="no-such-key"): - find_key_by_kid(jwks, "no-such-key") + _find_key_by_kid(jwks, "no-such-key") def test_find_key_by_kid_raises_on_empty_keys(): """Raises JwksKeyNotFoundError when the key set is empty.""" with pytest.raises(JwksKeyNotFoundError): - find_key_by_kid({"keys": []}, "any-kid") + _find_key_by_kid({"keys": []}, "any-kid") + + +async def test_aclose_http_client_resets_client(monkeypatch): + called = {"aclose": 0} + + class DummyClient: + is_closed = False + + async def aclose(self): + called["aclose"] += 1 + + monkeypatch.setattr("connect.merchant.jwks._http_client", DummyClient()) + + await aclose_http_client() + + assert called["aclose"] == 1 + assert jwks_module._http_client is None diff --git a/tests/merchant/test_license.py b/tests/merchant/test_license.py index c559b7a..4dabd43 100644 --- a/tests/merchant/test_license.py +++ b/tests/merchant/test_license.py @@ -1,14 +1,22 @@ """Tests for license token verification and result builders.""" +import json from datetime import timedelta import respx -from connect.merchant.license import build_block_result, build_signal_result, verify_license_token +from connect.merchant.license import ( + build_block_result, + build_signal_result, + verify_and_record_event, + verify_license_token, +) from connect.types import HandlerAction, InvalidLicenseToken, LicenseTokenInvalidReason, ValidLicenseToken from tests.merchant.constants import JWKS_URL, REQUEST_URL, SUPERTAB_BASE_URL +EVENTS_URL = f"{SUPERTAB_BASE_URL}/events" + async def test_verify_valid_token(make_token, jwks_response): """Valid token returns ValidLicenseToken with correct payload.""" @@ -222,3 +230,67 @@ def test_build_block_result_sanitizes_header_value(): assert "\r" not in www_auth assert "\n" not in www_auth assert '\\"' in www_auth + + +async def test_verify_and_record_event_records_license_used_for_valid_token(make_token, jwks_response, monkeypatch): + token = make_token() + monkeypatch.setattr("connect.merchant.license._get_sdk_user_agent", lambda: "sdk-test/1.2.3") + + with respx.mock: + respx.get(JWKS_URL).respond(json=jwks_response) + route = respx.post(EVENTS_URL).respond(status_code=201, json={"ok": True}) + + result = await verify_and_record_event( + token=token, + url=REQUEST_URL, + user_agent="Browser/1.0", + supertab_base_url=SUPERTAB_BASE_URL, + debug=False, + api_key="sk_test_123", + request_headers={ + "Accept": "text/html", + "X-Forwarded-For": "203.0.113.1", + }, + ) + + assert isinstance(result, ValidLicenseToken) + payload = json.loads(route.calls[0].request.content) + assert payload["event_name"] == "license_used" + assert payload["license_id"] == "lic_test_123" + assert payload["properties"] == { + "page_url": REQUEST_URL, + "user_agent": "Browser/1.0", + "sdk_user_agent": "sdk-test/1.2.3", + "verification_status": "valid", + "verification_reason": "success", + "h_accept": "text/html", + "h_x-forwarded-for": "203.0.113.1", + } + + +async def test_verify_and_record_event_records_invalid_reason(make_token, monkeypatch): + token = make_token(audience="https://other-site.com/page") + monkeypatch.setattr("connect.merchant.license._get_sdk_user_agent", lambda: "sdk-test/1.2.3") + + with respx.mock: + route = respx.post(EVENTS_URL).respond(status_code=201, json={"ok": True}) + + result = await verify_and_record_event( + token=token, + url=REQUEST_URL, + user_agent="Browser/1.0", + supertab_base_url=SUPERTAB_BASE_URL, + debug=False, + api_key="sk_test_123", + request_headers={"Accept": "text/html"}, + ) + + assert isinstance(result, InvalidLicenseToken) + payload = json.loads(route.calls[0].request.content) + assert payload["event_name"] == LicenseTokenInvalidReason.INVALID_AUDIENCE.value + assert payload["properties"]["page_url"] == REQUEST_URL + assert payload["properties"]["user_agent"] == "Browser/1.0" + assert payload["properties"]["sdk_user_agent"] == "sdk-test/1.2.3" + assert payload["properties"]["verification_status"] == "invalid" + assert payload["properties"]["verification_reason"] == LicenseTokenInvalidReason.INVALID_AUDIENCE.value + assert payload["properties"]["h_accept"] == "text/html"