"""HTTP/session helpers for ChatGPT registration.""" from __future__ import annotations import json import os import secrets import time import urllib.parse from typing import Any, Callable from .constants import ( CHATGPT_CSRF_URL, CHATGPT_SIGNIN_URL, DEFAULT_PASSWORD_LENGTH, OPENAI_API_ENDPOINTS, OPENAI_PAGE_TYPES, OAUTH_REDIRECT_URI, PASSWORD_CHARSET, generate_random_user_info, ) DEFAULT_ADD_PHONE_OAUTH_OTP_TIMEOUT_SECONDS = 90 def _deduplicate_cross_domain_cookies(session: Any, target_domain: str) -> None: """Remove cookies from non-target domains that conflict with target domain cookies. curl_cffi raises an error when multiple cookies with the same name exist on different domains (e.g. __cf_bm on .auth.openai.com vs .sentinel.openai.com). This helper keeps only the cookie scoped to *target_domain*. """ try: jar = session.cookies target_suffix = target_domain if target_domain.startswith(".") else f".{target_domain}" names_for_target: set[str] = set() for cookie in jar: domain = str(getattr(cookie, "domain", "") or "") if domain == target_suffix or domain == target_domain: names_for_target.add(cookie.name) to_remove: list[Any] = [] for cookie in jar: domain = str(getattr(cookie, "domain", "") or "") if cookie.name in names_for_target and domain != target_suffix and domain != target_domain: to_remove.append(cookie) for cookie in to_remove: jar.clear(domain=getattr(cookie, "domain", ""), path=getattr(cookie, "path", "/"), name=cookie.name) except Exception: pass def _build_sentinel_header( self, sentinel: str, device_id: str, flow: str, *, client: Any | None = None, ) -> str: sentinel_client = client or self.http_client build_header = getattr(sentinel_client, "build_sentinel_header", None) if callable(build_header): try: return str(build_header(device_id=device_id, flow=flow, token=sentinel)) except Exception: pass return json.dumps( { "p": "", "t": "", "c": sentinel, "id": device_id, "flow": flow, }, separators=(",", ":"), ) def _oauth_json_headers(self, *, referer: str, device_id: str) -> dict[str, str]: return { "accept": "application/json", "content-type": "application/json", "origin": "https://auth.openai.com", "referer": referer, "oai-device-id": device_id, "user-agent": self.http_client.default_headers.get("User-Agent", "Mozilla/5.0"), "sec-ch-ua": self.http_client.default_headers.get("sec-ch-ua", ""), "sec-ch-ua-mobile": self.http_client.default_headers.get("sec-ch-ua-mobile", ""), "sec-ch-ua-platform": self.http_client.default_headers.get("sec-ch-ua-platform", ""), } def _is_transient_transport_error(self, exc: Exception) -> bool: message = str(exc or "").lower() markers = ( "connection closed abruptly", "connection timed out", "connection reset", "connection refused", "tls connect error", "recv failure", "send failure", "http/2 stream", "operation timed out", "curl: (7)", "curl: (28)", "curl: (35)", "curl: (52)", "curl: (55)", "curl: (56)", ) return any(marker in message for marker in markers) def _session_request( self, *, session: Any, method: str, url: str, label: str, refresh_session: Callable[[], Any] | None = None, max_attempts: int = 3, retry_delay: float = 1.0, **kwargs: Any, ) -> tuple[Any, Any]: current_session = session last_exc: Exception | None = None for attempt in range(1, max_attempts + 1): try: _deduplicate_cross_domain_cookies(current_session, "auth.openai.com") response = getattr(current_session, method.lower())(url, **kwargs) return response, current_session except Exception as exc: last_exc = exc if attempt >= max_attempts or not self._is_transient_transport_error(exc): raise self._log(f"{label}: transient transport error, retry {attempt}/{max_attempts}: {exc}") if refresh_session is not None: current_session = refresh_session() time.sleep(retry_delay * attempt) if last_exc is not None: raise last_exc raise RuntimeError(f"{label}: request failed without exception") def _refresh_registration_session(self) -> Any: cookie_pairs: dict[str, str] = {} current_session = getattr(self, 'session', None) try: cookies = getattr(current_session, 'cookies', None) if cookies is not None: jar = getattr(cookies, 'jar', None) if jar is not None: for item in list(jar): name = str(getattr(item, 'name', '') or '').strip() value = str(getattr(item, 'value', '') or '').strip() if name: cookie_pairs[name] = value for key, value in dict(cookies).items(): if key: cookie_pairs[str(key)] = str(value) except Exception: pass try: self.http_client.close() except Exception: pass new_session = self.http_client.session try: new_session.cookies.update(cookie_pairs) except Exception: pass self.session = new_session return new_session def _check_ip_location(self) -> tuple[bool, str | None]: try: return self.http_client.check_ip_location() except Exception as exc: self._log(f"check_ip_location failed: {exc}") return False, None def _create_email(self) -> bool: self._last_mailbox_error_kind = "" self._last_mailbox_error_stage = "" self._last_mailbox_error_message = "" if self.email: self.email_info = {"email": self.email} self._log(f"using provided mailbox: {self.email}") return True for attempt in range(1, self.create_email_max_attempts + 1): try: candidate_info = self.email_service.create_email() except Exception as exc: self._last_mailbox_error_stage = "create_email" self._last_mailbox_error_message = str(exc or "").strip() self._last_mailbox_error_kind = "transport_error" if self._is_transient_transport_error(exc) else "provider_error" self._log(f"create_email failed: {exc}") return False candidate_email = str((candidate_info or {}).get("email") or "").strip() if not candidate_email: self._last_mailbox_error_stage = "create_email" self._last_mailbox_error_kind = "provider_error" self._last_mailbox_error_message = "create_email returned no email address" self._log("create_email returned no email address") return False if self.mailbox_dedupe_store is not None and not self.mailbox_dedupe_store.reserve(candidate_email): self._log( f"duplicate mailbox discarded ({attempt}/{self.create_email_max_attempts}): {candidate_email}" ) continue self.email_info = candidate_info self.email = candidate_email self._reserved_email = candidate_email self._log(f"created mailbox: {self.email}") return True self._log("create_email exhausted unique mailbox retries") self._last_mailbox_error_stage = "create_email" self._last_mailbox_error_kind = "provider_error" self._last_mailbox_error_message = "create_email exhausted unique mailbox retries" return False def _init_session(self) -> bool: try: self.session = self.http_client.session return True except Exception as exc: self._log(f"init_session failed: {exc}") return False def _start_oauth_via_chatgpt(self) -> bool: """Initiate OAuth flow via chatgpt.com (csrf + signin/openai) to avoid add_phone.""" import uuid as _uuid from .oauth import OAuthStart if self.session is None: return False try: csrf_resp = self.session.get(CHATGPT_CSRF_URL, timeout=15) if csrf_resp.status_code != 200: self._log(f"chatgpt csrf failed: {csrf_resp.status_code}") return False csrf_token = csrf_resp.json().get("csrfToken", "") if not csrf_token: self._log("chatgpt csrf token empty") return False device_id = str(_uuid.uuid4()) self.session.cookies.set("oai-did", device_id, domain=".openai.com") signin_url = f"{CHATGPT_SIGNIN_URL}?prompt=login&ext-oai-did={device_id}" signin_resp = self.session.post( signin_url, data=f"callbackUrl=https%3A%2F%2Fchatgpt.com%2F&csrfToken={csrf_token}&json=true", headers={"content-type": "application/x-www-form-urlencoded"}, timeout=15, allow_redirects=False, ) if signin_resp.status_code != 200: self._log(f"chatgpt signin failed: {signin_resp.status_code}") return False auth_url = signin_resp.json().get("url", "") if not auth_url: self._log("chatgpt signin returned no url") return False parsed = urllib.parse.urlparse(auth_url) params = dict(urllib.parse.parse_qsl(parsed.query)) self.oauth_start = OAuthStart( auth_url=auth_url, state=params.get("state", ""), code_verifier="", redirect_uri=OAUTH_REDIRECT_URI, ) self._log("oauth flow initialized via chatgpt.com") return True except Exception as exc: self._log(f"chatgpt oauth init failed: {exc}") return False def _start_oauth(self) -> bool: try: self.oauth_start = self.oauth_manager.start_oauth() self._log("oauth flow initialized") return True except Exception as exc: self._log(f"oauth init failed: {exc}") return False def _get_device_id(self) -> str | None: if not self.oauth_start or self.session is None: return None try: self.session.get(self.oauth_start.auth_url, timeout=15) device_id = str(self.session.cookies.get("oai-did") or "").strip() if device_id: self._log(f"device_id acquired: {device_id}") return device_id self._log("device_id missing from oauth bootstrap cookies") return None except Exception as exc: self._log(f"get_device_id failed: {exc}") return None def _check_sentinel(self, did: str, *, flow: str = "authorize_continue") -> str | None: try: token = self.http_client.check_sentinel(did, flow=flow) if token: self._log("sentinel token acquired") else: self._log("sentinel token unavailable") return token except Exception as exc: self._log(f"check_sentinel failed: {exc}") return None def _submit_signup_form(self, did: str, sen_token: str | None) -> SignupFormResult: if self.session is None or not self.email: return SignupFormResult(success=False, error_message="session or email missing") try: self._last_signup_http_status = None self._last_signup_error_code = "" self._last_signup_error_message = "" self._last_signup_error_body = "" signup_body = json.dumps( { "username": {"value": self.email, "kind": "email"}, "screen_hint": "signup", } ) headers = { "referer": "https://auth.openai.com/create-account", "accept": "application/json", "content-type": "application/json", } if sen_token: sentinel = self._build_sentinel_header( sen_token, did, "authorize_continue", ) headers["openai-sentinel-token"] = sentinel response = self.session.post( OPENAI_API_ENDPOINTS["signup"], headers=headers, data=signup_body, ) self._log(f"signup form status: {response.status_code}") self._last_signup_http_status = int(response.status_code) if response.status_code != 200: self._last_signup_error_body = str(response.text or "")[:240] try: error_payload = response.json() except Exception: error_payload = None if isinstance(error_payload, dict): error = error_payload.get("error") if isinstance(error, dict): self._last_signup_error_code = str(error.get("code") or "").strip() self._last_signup_error_message = str(error.get("message") or "").strip() return SignupFormResult( success=False, error_message=f"HTTP {response.status_code}: {response.text[:200]}", ) try: response_data = response.json() except Exception as exc: return SignupFormResult(success=False, error_message=f"signup json parse failed: {exc}") page_type = str(((response_data.get("page") or {}).get("type")) or "").strip() is_existing = page_type == OPENAI_PAGE_TYPES["EMAIL_OTP_VERIFICATION"] self._is_existing_account = is_existing if is_existing: self._log("existing account detected; switching to login-like OTP flow") else: self._log(f"signup page type: {page_type or 'unknown'}") self._log(f"signup response data keys: {list(response_data.keys()) if isinstance(response_data, dict) else 'not dict'}") self._log(f"signup response data: {json.dumps(response_data, default=str)[:500]}") return SignupFormResult( success=True, page_type=page_type, is_existing_account=is_existing, response_data=response_data, ) except Exception as exc: self._log(f"submit_signup_form failed: {exc}") return SignupFormResult(success=False, error_message=str(exc)) def _register_password(self) -> bool: if self.session is None or not self.email: return False try: if not self.password: self.password = self._generate_password() _deduplicate_cross_domain_cookies(self.session, "auth.openai.com") device_id = getattr(self, "_last_device_id", "") or "" sentinel_token = getattr(self, "_last_sentinel_token", None) try: self.session.get( "https://auth.openai.com/create-account/password", headers={"referer": "https://auth.openai.com/create-account"}, ) except Exception: pass payload = json.dumps({"password": self.password, "username": self.email}) headers: dict[str, str] = { "referer": "https://auth.openai.com/create-account/password", "accept": "application/json", "content-type": "application/json", } if device_id: headers["oai-device-id"] = device_id if sentinel_token: headers["openai-sentinel-token"] = self._build_sentinel_header( sentinel_token, device_id, "authorize_continue" ) response = self.session.post( OPENAI_API_ENDPOINTS["register"], headers=headers, data=payload, ) self._log(f"register password status: {response.status_code}") if response.status_code != 200: self._log(f"register password failed body: {response.text[:240]}") return False return True except Exception as exc: self._log(f"register_password failed: {exc}") return False def _send_verification_code(self) -> bool: if self.session is None: return False try: self._signup_otp_before_ids = self._capture_mailbox_ids() self._otp_sent_at = time.time() self._log( "send otp mailbox baseline captured: " f"{len(self._signup_otp_before_ids)} existing ids" ) response, session = self._session_request( session=self.session, method="GET", url=OPENAI_API_ENDPOINTS["send_otp"], label="send otp", refresh_session=self._refresh_registration_session, headers={ "referer": "https://auth.openai.com/create-account/password", "accept": "application/json", }, ) self.session = session self._log(f"send otp status: {response.status_code}") return response.status_code == 200 except Exception as exc: self._log(f"send_verification_code failed: {exc}") return False def _create_user_account(self) -> bool: if self.session is None: return False try: self._last_create_account_http_status = None self._last_create_account_error_code = "" self._last_create_account_error_message = "" self._last_create_account_error_body = "" user_info = generate_random_user_info() self._log(f"generated profile: {user_info['name']} / {user_info['birthdate']}") _deduplicate_cross_domain_cookies(self.session, "auth.openai.com") device_id = getattr(self, "_last_device_id", "") or "" sentinel_token = getattr(self, "_last_sentinel_token", None) headers = { "referer": "https://auth.openai.com/about-you", "accept": "application/json", "content-type": "application/json", } if device_id: headers["oai-device-id"] = device_id if sentinel_token: headers["openai-sentinel-token"] = self._build_sentinel_header( sentinel_token, device_id, "authorize_continue" ) response = self.session.post( OPENAI_API_ENDPOINTS["create_account"], headers=headers, data=json.dumps(user_info), ) self._last_create_account_http_status = int(response.status_code) self._log(f"create account status: {response.status_code}") if response.status_code != 200: self._last_create_account_error_body = str(response.text or "")[:240] self._log(f"create account body: {self._last_create_account_error_body}") try: error_payload = response.json() except Exception: error_payload = {} error_info = error_payload.get("error") if isinstance(error_payload, dict) else {} if isinstance(error_info, dict): self._last_create_account_error_code = str(error_info.get("code") or "").strip() self._last_create_account_error_message = str(error_info.get("message") or "").strip() if self._last_create_account_error_code or self._last_create_account_error_message: self._log( "create account classified error: " f"code={self._last_create_account_error_code or '-'} " f"message={self._last_create_account_error_message or '-'}" ) return False try: create_resp = response.json() self._log(f"create account response keys: {list(create_resp.keys())}") # Store continue_url from response (bypass workspace flow) curl = str(create_resp.get("continue_url") or "").strip() page_info = create_resp.get("page") or {} page_type = str(page_info.get("type") or "").strip() if isinstance(page_info, dict) else "" continue_host = "" continue_kind = "unknown" if curl: parsed_curl = urllib.parse.urlparse(curl) continue_host = parsed_curl.netloc if "callback/openai" in curl: continue_kind = "callback_openai" elif "add-phone" in curl: continue_kind = "add_phone" elif "workspace" in curl: continue_kind = "workspace" else: continue_kind = f"other:{parsed_curl.path[:40]}" self._log( f"create_account result: page_type={page_type}, " f"continue_kind={continue_kind}, continue_host={continue_host}" ) if curl: self._create_account_continue_url = curl self._log(f"continue_url from create_account: {curl[:120]}") except Exception: pass return True except Exception as exc: self._log(f"create_user_account failed: {exc}") return False def _email_domain(self) -> str: email = str(self.email or "").strip() if "@" not in email: return "" return email.rsplit("@", 1)[-1].strip().lower() def _load_add_phone_oauth_max_attempts(self) -> int: raw = str(os.getenv("ZHUCE6_ADD_PHONE_OAUTH_MAX_ATTEMPTS", "2") or "2").strip() try: value = int(raw) except Exception: value = 2 return max(1, min(value, 3)) def _load_wait_otp_timeout_seconds(self) -> int: raw = str(os.getenv("ZHUCE6_WAIT_OTP_TIMEOUT_SECONDS", "180") or "180").strip() try: value = int(raw) except Exception: value = 180 return max(60, min(value, 300)) def _load_add_phone_oauth_otp_timeout_seconds(self) -> int: raw = str( os.getenv( "ZHUCE6_ADD_PHONE_OAUTH_OTP_TIMEOUT_SECONDS", str(DEFAULT_ADD_PHONE_OAUTH_OTP_TIMEOUT_SECONDS), ) or str(DEFAULT_ADD_PHONE_OAUTH_OTP_TIMEOUT_SECONDS) ).strip() try: value = int(raw) except Exception: value = DEFAULT_ADD_PHONE_OAUTH_OTP_TIMEOUT_SECONDS return max(30, min(value, 180)) def _load_post_create_login_delay_seconds(self) -> int: raw = str(os.getenv("ZHUCE6_POST_CREATE_LOGIN_DELAY_SECONDS", "8") or "8").strip() try: value = int(raw) except Exception: value = 8 return max(0, min(value, 600)) def _metadata(self, extra: dict[str, Any] | None = None) -> dict[str, Any]: payload: dict[str, Any] = { "email_domain": self._email_domain(), "signup_http_status": self._last_signup_http_status, "signup_error_code": self._last_signup_error_code, "signup_error_message": self._last_signup_error_message, "signup_auth_reset_count": getattr(self, "_signup_auth_reset_count", 0), "mailbox_error_kind": getattr(self, "_last_mailbox_error_kind", ""), "mailbox_error_stage": getattr(self, "_last_mailbox_error_stage", ""), "mailbox_error_message": getattr(self, "_last_mailbox_error_message", ""), "create_account_http_status": self._last_create_account_http_status, "create_account_error_code": self._last_create_account_error_code, "create_account_error_message": self._last_create_account_error_message, } if self._last_signup_error_body: payload["signup_error_body"] = self._last_signup_error_body if self._last_create_account_error_body: payload["create_account_error_body"] = self._last_create_account_error_body if self._last_otp_wait_failure_reason: payload["otp_wait_failure_reason"] = self._last_otp_wait_failure_reason if self._last_otp_wait_diagnostics: payload.update(self._last_otp_wait_diagnostics) if extra: payload.update(extra) return payload def _generate_password(self, length: int = DEFAULT_PASSWORD_LENGTH) -> str: required = [ secrets.choice("abcdefghijklmnopqrstuvwxyz"), secrets.choice("ABCDEFGHIJKLMNOPQRSTUVWXYZ"), secrets.choice("0123456789"), secrets.choice("!@#$%&*"), ] rest = [secrets.choice(PASSWORD_CHARSET) for _ in range(length - len(required))] combined = required + rest secrets.SystemRandom().shuffle(combined) return "".join(combined) def _auth_url(self, url: str) -> str: candidate = str(url or "").strip() if not candidate: return "" return urllib.parse.urljoin("https://auth.openai.com", candidate)