Merchant feature parity

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

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}"

connect/customer/__init__.py

@@ -1 +1,5 @@
 """Customer functionality for Supertab Connect."""
+
+from connect.customer.token import obtain_license_token
+
+__all__ = ["obtain_license_token"]

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

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",
+]

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

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}