修改sdk解析改为一次性

platforms/chatgpt/sentinel_batch.py

@@ -0,0 +1,497 @@
+"""Standalone Sentinel SDK batch token helper."""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import uuid
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Iterable, Mapping, Optional, Protocol
+
+from core.browser_runtime import (
+    ensure_browser_display_available,
+    resolve_browser_headless,
+)
+from core.config_store import ConfigStore, config_store
+from core.proxy_pool import ProxyPool, proxy_pool
+from core.proxy_utils import build_playwright_proxy_config, normalize_proxy_url
+
+
+DEFAULT_SDK_VERSION = "20260219f9f6"
+DEFAULT_FRAME_URL = (
+    f"https://sentinel.openai.com/backend-api/sentinel/frame.html?sv={DEFAULT_SDK_VERSION}"
+)
+DEFAULT_SDK_URL = f"https://sentinel.openai.com/sentinel/{DEFAULT_SDK_VERSION}/sdk.js"
+DEFAULT_USER_AGENT = (
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+    "AppleWebKit/537.36 (KHTML, like Gecko) "
+    "Chrome/136.0.7103.92 Safari/537.36"
+)
+DEFAULT_OUT = Path(tempfile.gettempdir()) / "sentinel_multi_helper_out.json"
+DEFAULT_FLOW_SPECS: tuple["FlowSpec", ...]
+
+
+@dataclass(frozen=True)
+class FlowSpec:
+    internal_name: str
+    alias: str
+    page_url: str
+    needs_session_observer_token: bool = False
+
+
+DEFAULT_FLOW_SPECS = (
+    FlowSpec(
+        internal_name="authorize_continue",
+        alias="authorize-continue",
+        page_url="https://auth.openai.com/create-account",
+    ),
+    FlowSpec(
+        internal_name="username_password_create",
+        alias="username-password-create",
+        page_url="https://auth.openai.com/create-account/password",
+    ),
+    FlowSpec(
+        internal_name="password_verify",
+        alias="password-verify",
+        page_url="https://auth.openai.com/log-in/password",
+    ),
+    FlowSpec(
+        internal_name="oauth_create_account",
+        alias="oauth-create-account",
+        page_url="https://auth.openai.com/about-you",
+        needs_session_observer_token=True,
+    ),
+)
+
+
+@dataclass(frozen=True)
+class SentinelBatchConfig:
+    frame_url: str
+    sdk_url: str
+    user_agent: str
+    output_path: Path
+    proxy: Optional[str]
+    flows: tuple[FlowSpec, ...]
+    headless: bool = True
+    headless_reason: str = ""
+
+
+@dataclass
+class FlowTokenResult:
+    flow: str
+    page_url: str
+    sentinel_token: Optional[str] = None
+    sentinel_so_token: Optional[str] = None
+    error: Optional[str] = None
+
+    def to_dict(self) -> dict[str, object]:
+        data: dict[str, object] = {
+            "flow": self.flow,
+            "pageUrl": self.page_url,
+        }
+        if self.sentinel_token:
+            data["sentinel-token"] = self.sentinel_token
+        if self.sentinel_so_token:
+            data["sentinel-so-token"] = self.sentinel_so_token
+        if self.error:
+            data["error"] = self.error
+        return data
+
+
+@dataclass
+class SentinelBatchResult:
+    generated_at: str
+    device_id: str
+    proxy: Optional[str]
+    frame_url: str
+    sdk_url: str
+    user_agent: str
+    flows: dict[str, FlowTokenResult] = field(default_factory=dict)
+
+    @property
+    def has_errors(self) -> bool:
+        return any(item.error for item in self.flows.values())
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "generatedAt": self.generated_at,
+            "deviceId": self.device_id,
+            "proxy": self.proxy or "",
+            "frameUrl": self.frame_url,
+            "sdkUrl": self.sdk_url,
+            "userAgent": self.user_agent,
+            "flows": {alias: item.to_dict() for alias, item in self.flows.items()},
+        }
+
+    def to_json(self) -> str:
+        return json.dumps(self.to_dict(), ensure_ascii=False, indent=2)
+
+
+class ProxySelector(Protocol):
+    def select_proxy(self) -> Optional[str]: ...
+
+
+class SentinelProvider(ABC):
+    @abstractmethod
+    def __enter__(self) -> "SentinelProvider":
+        raise NotImplementedError
+
+    @abstractmethod
+    def __exit__(self, exc_type, exc, tb) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_flow_token(self, flow: FlowSpec) -> str:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_session_observer_token(self, flow: FlowSpec) -> str:
+        raise NotImplementedError
+
+    @abstractmethod
+    def resolved_sdk_url(self) -> str:
+        raise NotImplementedError
+
+
+class ConfigBackedProxySelector:
+    """Resolve proxy from explicit config/env values and then proxy pool."""
+
+    EXPLICIT_PROXY_KEYS = (
+        "PROXY_SERVER",
+        "proxy_server",
+        "sentinel_proxy_server",
+        "sentinel.proxy_server",
+        "proxy_url",
+        "proxy.url",
+    )
+
+    def __init__(
+        self,
+        config: ConfigStore = config_store,
+        pool: ProxyPool = proxy_pool,
+    ) -> None:
+        self._config = config
+        self._pool = pool
+
+    def _get_first(self, keys: Iterable[str]) -> str:
+        for key in keys:
+            value = str(self._config.get(key, "") or "").strip()
+            if value:
+                return value
+        return ""
+
+    def _build_from_global_proxy_config(self) -> str:
+        enabled = str(self._config.get("proxy.enabled", "") or "").strip().lower()
+        if enabled in {"0", "false", "no", "off"}:
+            return ""
+
+        host = str(self._config.get("proxy.host", "") or "").strip()
+        if not host:
+            return ""
+
+        scheme = str(self._config.get("proxy.type", "http") or "http").strip() or "http"
+        port = str(self._config.get("proxy.port", "") or "").strip()
+        user = str(self._config.get("proxy.username", "") or "").strip()
+        password = str(self._config.get("proxy.password", "") or "").strip()
+
+        auth = ""
+        if user:
+            auth = user
+            if password:
+                auth += f":{password}"
+            auth += "@"
+
+        if port:
+            return f"{scheme}://{auth}{host}:{port}"
+        return f"{scheme}://{auth}{host}"
+
+    def select_proxy(self) -> Optional[str]:
+        explicit_proxy = self._get_first(self.EXPLICIT_PROXY_KEYS)
+        if explicit_proxy:
+            return normalize_proxy_url(explicit_proxy)
+
+        global_proxy = self._build_from_global_proxy_config()
+        if global_proxy:
+            return normalize_proxy_url(global_proxy)
+
+        return normalize_proxy_url(self._pool.get_next() or None)
+
+
+class ConfigResolver:
+    """Resolve runtime config with clear precedence rules."""
+
+    def __init__(
+        self,
+        config: ConfigStore = config_store,
+        proxy_selector: Optional[ProxySelector] = None,
+        environ: Optional[Mapping[str, str]] = None,
+    ) -> None:
+        self._config = config
+        self._proxy_selector = proxy_selector or ConfigBackedProxySelector(config=config)
+        self._environ = environ if environ is not None else os.environ
+
+    def _get(self, key: str, default: str = "") -> str:
+        if key in self._environ:
+            value = str(self._environ.get(key, "") or "").strip()
+            if value:
+                return value
+        return str(self._config.get(key, default) or default).strip()
+
+    def _resolve_output_path(self) -> Path:
+        out_value = self._get("OUT", "")
+        return Path(out_value).expanduser() if out_value else DEFAULT_OUT
+
+    def _resolve_flows(self) -> tuple[FlowSpec, ...]:
+        flow_alias_map = {
+            spec.internal_name: spec for spec in DEFAULT_FLOW_SPECS
+        } | {spec.alias: spec for spec in DEFAULT_FLOW_SPECS}
+        flows_raw = self._get("FLOWS", "")
+        if not flows_raw:
+            return DEFAULT_FLOW_SPECS
+
+        selected: list[FlowSpec] = []
+        for item in flows_raw.split(","):
+            key = item.strip()
+            if not key:
+                continue
+            spec = flow_alias_map.get(key)
+            if not spec:
+                raise ValueError(f"Unsupported sentinel flow: {key}")
+            if spec not in selected:
+                selected.append(spec)
+        return tuple(selected) or DEFAULT_FLOW_SPECS
+
+    def resolve(self) -> SentinelBatchConfig:
+        requested_headless = None
+        if "HEADLESS" in self._environ:
+            requested_headless = self._environ.get("HEADLESS", "").strip().lower() not in {
+                "",
+                "0",
+                "false",
+                "no",
+                "off",
+            }
+
+        headless, reason = resolve_browser_headless(requested_headless)
+        proxy = self._proxy_selector.select_proxy()
+        return SentinelBatchConfig(
+            frame_url=self._get("FRAME_URL", DEFAULT_FRAME_URL) or DEFAULT_FRAME_URL,
+            sdk_url=self._get("SDK_URL", DEFAULT_SDK_URL) or DEFAULT_SDK_URL,
+            user_agent=self._get("UA", DEFAULT_USER_AGENT) or DEFAULT_USER_AGENT,
+            output_path=self._resolve_output_path(),
+            proxy=proxy,
+            flows=self._resolve_flows(),
+            headless=headless,
+            headless_reason=reason,
+        )
+
+
+class PlaywrightSentinelProvider(SentinelProvider):
+    """Fetch Sentinel tokens through the browser SDK in a single session."""
+
+    def __init__(
+        self,
+        config: SentinelBatchConfig,
+        *,
+        device_id: str,
+        timeout_ms: int = 60000,
+    ) -> None:
+        self._config = config
+        self._device_id = device_id
+        self._timeout_ms = timeout_ms
+        self._playwright = None
+        self._browser = None
+        self._context = None
+        self._page = None
+        self._resolved_sdk_url = config.sdk_url
+
+    def __enter__(self) -> "PlaywrightSentinelProvider":
+        from playwright.sync_api import sync_playwright
+
+        ensure_browser_display_available(self._config.headless)
+        self._playwright = sync_playwright().start()
+
+        launch_kwargs: dict[str, object] = {
+            "headless": self._config.headless,
+            "args": [
+                "--no-sandbox",
+                "--disable-blink-features=AutomationControlled",
+            ],
+        }
+        proxy_config = build_playwright_proxy_config(self._config.proxy)
+        if proxy_config:
+            launch_kwargs["proxy"] = proxy_config
+
+        self._browser = self._playwright.chromium.launch(**launch_kwargs)
+        self._context = self._browser.new_context(
+            user_agent=self._config.user_agent,
+            locale="en-US",
+            viewport={"width": 1920, "height": 1080},
+            ignore_https_errors=True,
+        )
+        self._context.add_cookies(
+            [
+                {
+                    "name": "oai-did",
+                    "value": self._device_id,
+                    "url": "https://sentinel.openai.com/",
+                    "path": "/",
+                    "secure": True,
+                    "sameSite": "Lax",
+                },
+                {
+                    "name": "oai-did",
+                    "value": self._device_id,
+                    "url": "https://auth.openai.com/",
+                    "path": "/",
+                    "secure": True,
+                    "sameSite": "Lax",
+                },
+            ]
+        )
+        self._page = self._context.new_page()
+        self._page.goto(self._config.frame_url, wait_until="load", timeout=self._timeout_ms)
+        self._ensure_sdk_loaded()
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:
+        if self._browser is not None:
+            self._browser.close()
+        if self._playwright is not None:
+            self._playwright.stop()
+
+    def _ensure_sdk_loaded(self) -> None:
+        assert self._page is not None
+        self._page.wait_for_load_state("load", timeout=self._timeout_ms)
+        try:
+            self._page.wait_for_function(
+                "() => !!window.SentinelSDK",
+                timeout=min(self._timeout_ms, 15000),
+            )
+        except Exception:
+            self._inject_expected_sdk()
+            self._page.wait_for_function(
+                "() => !!window.SentinelSDK",
+                timeout=min(self._timeout_ms, 30000),
+            )
+
+        self._resolved_sdk_url = self._page.evaluate(
+            """
+            (expectedSdkUrl) => {
+                const scripts = Array.from(document.scripts || [])
+                    .map((item) => item.src)
+                    .filter(Boolean);
+                return scripts.find((src) => src.includes('/sdk.js')) || expectedSdkUrl;
+            }
+            """,
+            self._config.sdk_url,
+        )
+
+    def _inject_expected_sdk(self) -> None:
+        assert self._page is not None
+        self._page.evaluate(
+            """
+            async (sdkUrl) => {
+                const existing = Array.from(document.scripts || [])
+                    .some((item) => item.src === sdkUrl);
+                if (existing) return;
+                await new Promise((resolve, reject) => {
+                    const script = document.createElement('script');
+                    script.src = sdkUrl;
+                    script.async = true;
+                    script.onload = () => resolve(true);
+                    script.onerror = () => reject(new Error(`Failed to load ${sdkUrl}`));
+                    document.head.appendChild(script);
+                });
+            }
+            """,
+            self._config.sdk_url,
+        )
+
+    def _invoke_sdk(self, method: str, flow: FlowSpec) -> str:
+        assert self._page is not None
+        result = self._page.evaluate(
+            """
+            async ({ flow, methodName }) => {
+                if (!window.SentinelSDK) {
+                    throw new Error('SentinelSDK missing');
+                }
+                const target = window.SentinelSDK[methodName];
+                if (typeof target !== 'function') {
+                    throw new Error(`SentinelSDK.${methodName} missing`);
+                }
+                if (typeof window.SentinelSDK.init === 'function') {
+                    await window.SentinelSDK.init(flow);
+                }
+                return await target.call(window.SentinelSDK, flow);
+            }
+            """,
+            {"flow": flow.internal_name, "methodName": method},
+        )
+        token = str(result or "").strip()
+        if not token:
+            raise RuntimeError(f"Empty {method} result for {flow.internal_name}")
+        return token
+
+    def get_flow_token(self, flow: FlowSpec) -> str:
+        return self._invoke_sdk("token", flow)
+
+    def get_session_observer_token(self, flow: FlowSpec) -> str:
+        return self._invoke_sdk("sessionObserverToken", flow)
+
+    def resolved_sdk_url(self) -> str:
+        return self._resolved_sdk_url
+
+
+class SentinelProviderFactory:
+    def create(
+        self,
+        config: SentinelBatchConfig,
+        *,
+        device_id: str,
+    ) -> SentinelProvider:
+        return PlaywrightSentinelProvider(config=config, device_id=device_id)
+
+
+class SentinelBatchService:
+    def __init__(
+        self,
+        provider_factory: Optional[SentinelProviderFactory] = None,
+        *,
+        device_id_factory=None,
+    ) -> None:
+        self._provider_factory = provider_factory or SentinelProviderFactory()
+        self._device_id_factory = device_id_factory or (lambda: str(uuid.uuid4()))
+
+    def generate(self, config: SentinelBatchConfig) -> SentinelBatchResult:
+        device_id = self._device_id_factory()
+        result = SentinelBatchResult(
+            generated_at=datetime.now(timezone.utc).isoformat(),
+            device_id=device_id,
+            proxy=config.proxy,
+            frame_url=config.frame_url,
+            sdk_url=config.sdk_url,
+            user_agent=config.user_agent,
+        )
+        provider = self._provider_factory.create(config, device_id=device_id)
+        with provider:
+            result.sdk_url = provider.resolved_sdk_url()
+            for flow in config.flows:
+                item = FlowTokenResult(flow=flow.internal_name, page_url=flow.page_url)
+                try:
+                    item.sentinel_token = provider.get_flow_token(flow)
+                    if flow.needs_session_observer_token:
+                        item.sentinel_so_token = provider.get_session_observer_token(flow)
+                except Exception as exc:  # pragma: no cover - exercised via tests
+                    item.error = str(exc)
+                result.flows[flow.alias] = item
+        return result
+
+
+def write_batch_result(result: SentinelBatchResult, output_path: Path) -> None:
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(result.to_json(), encoding="utf-8")

scripts/sentinel_multi_helper.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from platforms.chatgpt.sentinel_batch import (  # noqa: E402
+    ConfigResolver,
+    SentinelBatchService,
+    write_batch_result,
+)
+
+
+def main() -> int:
+    resolver = ConfigResolver()
+    config = resolver.resolve()
+    service = SentinelBatchService()
+    result = service.generate(config)
+    write_batch_result(result, config.output_path)
+    print(result.to_json())
+    print(str(config.output_path))
+    return 1 if result.has_errors else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_chatgpt_sentinel_batch.py

@@ -0,0 +1,247 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from platforms.chatgpt.sentinel_batch import (
+    ConfigResolver,
+    DEFAULT_FLOW_SPECS,
+    DEFAULT_FRAME_URL,
+    DEFAULT_OUT,
+    DEFAULT_SDK_URL,
+    DEFAULT_USER_AGENT,
+    ConfigBackedProxySelector,
+    FlowSpec,
+    SentinelBatchConfig,
+    SentinelBatchService,
+)
+
+
+class _FakeConfigStore:
+    def __init__(self, values=None):
+        self.values = values or {}
+
+    def get(self, key, default=""):
+        return self.values.get(key, default)
+
+
+class _FakeProxyPool:
+    def __init__(self, proxy=None):
+        self.proxy = proxy
+        self.calls = 0
+
+    def get_next(self):
+        self.calls += 1
+        return self.proxy
+
+
+class _FakeProxySelector:
+    def __init__(self, proxy):
+        self.proxy = proxy
+
+    def select_proxy(self):
+        return self.proxy
+
+
+class _FakeProvider:
+    def __init__(self):
+        self.token_calls = []
+        self.so_calls = []
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return None
+
+    def get_flow_token(self, flow):
+        self.token_calls.append(flow.internal_name)
+        return f'{{"flow":"{flow.internal_name}","kind":"token"}}'
+
+    def get_session_observer_token(self, flow):
+        self.so_calls.append(flow.internal_name)
+        return f'{{"flow":"{flow.internal_name}","kind":"so"}}'
+
+    def resolved_sdk_url(self):
+        return "https://sentinel.openai.com/sentinel/20260219f9f6/sdk.js"
+
+
+class _FakeProviderFactory:
+    def __init__(self, provider):
+        self.provider = provider
+        self.last_config = None
+        self.last_device_id = None
+
+    def create(self, config, *, device_id):
+        self.last_config = config
+        self.last_device_id = device_id
+        return self.provider
+
+
+class ConfigBackedProxySelectorTests(unittest.TestCase):
+    def test_prefers_explicit_proxy_server_over_proxy_pool(self):
+        selector = ConfigBackedProxySelector(
+            config=_FakeConfigStore({"PROXY_SERVER": "http://configured:8080"}),
+            pool=_FakeProxyPool("http://pool:8080"),
+        )
+
+        self.assertEqual(selector.select_proxy(), "http://configured:8080")
+
+    def test_builds_proxy_from_global_proxy_settings(self):
+        selector = ConfigBackedProxySelector(
+            config=_FakeConfigStore(
+                {
+                    "proxy.enabled": "true",
+                    "proxy.type": "http",
+                    "proxy.host": "127.0.0.1",
+                    "proxy.port": "7890",
+                }
+            ),
+            pool=_FakeProxyPool("http://pool:8080"),
+        )
+
+        self.assertEqual(selector.select_proxy(), "http://127.0.0.1:7890")
+
+    def test_falls_back_to_proxy_pool_when_no_global_proxy(self):
+        pool = _FakeProxyPool("socks5://pool:1080")
+        selector = ConfigBackedProxySelector(
+            config=_FakeConfigStore({}),
+            pool=pool,
+        )
+
+        self.assertEqual(selector.select_proxy(), "socks5h://pool:1080")
+        self.assertEqual(pool.calls, 1)
+
+    def test_returns_none_when_proxy_pool_empty(self):
+        selector = ConfigBackedProxySelector(
+            config=_FakeConfigStore({}),
+            pool=_FakeProxyPool(None),
+        )
+
+        self.assertIsNone(selector.select_proxy())
+
+
+class ConfigResolverTests(unittest.TestCase):
+    def test_uses_env_and_defaults(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            resolver = ConfigResolver(
+                config=_FakeConfigStore({}),
+                proxy_selector=_FakeProxySelector("http://pool:8888"),
+                environ={
+                    "OUT": str(Path(tmp_dir) / "out.json"),
+                    "FRAME_URL": "https://example.com/frame.html",
+                },
+            )
+
+            config = resolver.resolve()
+
+        self.assertEqual(config.frame_url, "https://example.com/frame.html")
+        self.assertEqual(config.sdk_url, DEFAULT_SDK_URL)
+        self.assertEqual(config.user_agent, DEFAULT_USER_AGENT)
+        self.assertEqual(config.output_path, Path(tmp_dir) / "out.json")
+        self.assertEqual(config.proxy, "http://pool:8888")
+        self.assertEqual(config.flows, DEFAULT_FLOW_SPECS)
+
+    def test_flows_can_be_selected_by_alias_or_internal_name(self):
+        resolver = ConfigResolver(
+            config=_FakeConfigStore({}),
+            proxy_selector=_FakeProxySelector(None),
+            environ={
+                "FLOWS": "authorize_continue,oauth-create-account",
+            },
+        )
+
+        config = resolver.resolve()
+
+        self.assertEqual(
+            [item.internal_name for item in config.flows],
+            ["authorize_continue", "oauth_create_account"],
+        )
+
+    def test_invalid_flow_raises_error(self):
+        resolver = ConfigResolver(
+            config=_FakeConfigStore({}),
+            proxy_selector=_FakeProxySelector(None),
+            environ={"FLOWS": "bad-flow"},
+        )
+
+        with self.assertRaises(ValueError):
+            resolver.resolve()
+
+    def test_default_out_path_uses_windows_safe_temp_location(self):
+        resolver = ConfigResolver(
+            config=_FakeConfigStore({}),
+            proxy_selector=_FakeProxySelector(None),
+            environ={},
+        )
+
+        config = resolver.resolve()
+
+        self.assertEqual(config.output_path, DEFAULT_OUT)
+
+
+class SentinelBatchServiceTests(unittest.TestCase):
+    def test_requests_session_observer_only_for_oauth_create_account(self):
+        provider = _FakeProvider()
+        factory = _FakeProviderFactory(provider)
+        service = SentinelBatchService(
+            provider_factory=factory,
+            device_id_factory=lambda: "device-fixed",
+        )
+        config = SentinelBatchConfig(
+            frame_url=DEFAULT_FRAME_URL,
+            sdk_url=DEFAULT_SDK_URL,
+            user_agent=DEFAULT_USER_AGENT,
+            output_path=Path(tempfile.gettempdir()) / "out.json",
+            proxy=None,
+            flows=DEFAULT_FLOW_SPECS,
+            headless=True,
+            headless_reason="default:true",
+        )
+
+        result = service.generate(config)
+
+        self.assertEqual(factory.last_device_id, "device-fixed")
+        self.assertEqual(
+            provider.token_calls,
+            [flow.internal_name for flow in DEFAULT_FLOW_SPECS],
+        )
+        self.assertEqual(provider.so_calls, ["oauth_create_account"])
+        self.assertFalse(result.has_errors)
+        self.assertEqual(
+            result.flows["oauth-create-account"].sentinel_so_token,
+            '{"flow":"oauth_create_account","kind":"so"}',
+        )
+
+    def test_result_serializes_stable_output_shape(self):
+        provider = _FakeProvider()
+        service = SentinelBatchService(
+            provider_factory=_FakeProviderFactory(provider),
+            device_id_factory=lambda: "device-fixed",
+        )
+        config = SentinelBatchConfig(
+            frame_url=DEFAULT_FRAME_URL,
+            sdk_url=DEFAULT_SDK_URL,
+            user_agent=DEFAULT_USER_AGENT,
+            output_path=Path(tempfile.gettempdir()) / "out.json",
+            proxy="http://pool:8080",
+            flows=(
+                FlowSpec(
+                    internal_name="authorize_continue",
+                    alias="authorize-continue",
+                    page_url="https://auth.openai.com/create-account",
+                ),
+            ),
+            headless=True,
+            headless_reason="default:true",
+        )
+
+        payload = service.generate(config).to_dict()
+
+        self.assertEqual(payload["deviceId"], "device-fixed")
+        self.assertEqual(payload["proxy"], "http://pool:8080")
+        self.assertIn("authorize-continue", payload["flows"])
+        self.assertIn("sentinel-token", payload["flows"]["authorize-continue"])
+
+
+if __name__ == "__main__":
+    unittest.main()