chatgpt-register/platforms/chatgpt/register_http.py

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