`"""cohere_reg_bot.py

Mass‑registration bot for Cohere – IMAP edition
"""

from future import annotations

import argparse
import email
import imaplib
import json
import logging
import os
import random
import re
import string
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from email.message import Message
from typing import Dict, Optional

import requests

────────────────────────────────────────────────────────────────────────────

Constants & Config

────────────────────────────────────────────────────────────────────────────

IMAP_HOST = "mail.linux.do"
IMAP_PORT = 993
IMAP_USER = "@linux.do"
IMAPPASS = ""

COHERE_BASE = "https://production.api.os.cohere.com/rpc/BlobheartAPI"

COHERE_HEADERS_COMMON = {
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:139.0) Gecko/20100101 Firefox/139.0",
"Accept": "/",
"Accept-Language": "en-US,en;q=0.5",
"Accept-Encoding": "gzip, deflate, br, zstd",
"Referer": "https://dashboard.cohere.com/",
"Origin": "https://dashboard.cohere.com",
"DNT": "1",
"Sec-GPC": "1",
"Connection": "keep-alive",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-site",
"Priority": "u=0",
"Pragma": "no-cache",
"Cache-Control": "no-cache",
"TE": "trailers",
"Request-Source": "playground",
}

CONFIRM_LINK_RE = re.compile(r"href=\"([^\"]+confirm_email[^\"]+)\"")
HTML_META_REFRESH_RE = re.compile(r"url=([^\"]+confirm_email[^\"]+)", re.IGNORECASE)

────────────────────────────────────────────────────────────────────────────

Utils

────────────────────────────────────────────────────────────────────────────

def rand_str(n: int, alphabet: str = string.digits) -> str:
"""Return a random string of n characters from alphabet."""
return "".join(random.choices(alphabet, k=n))

def make_cohere_password() -> str:
return "@Co" + "".join(random.choices(string.ascii_letters + string.digits, k=7))

────────────────────────────────────────────────────────────────────────────

HTTP helper with DEBUG dumping

────────────────────────────────────────────────────────────────────────────

def http_request(method: str, url: str, *, logger: logging.Logger, kw) -> requests.Response:
resp = requests.request(method, url, kw)
if logger.isEnabledFor(logging.DEBUG):
req = resp.request
logger.debug("%s %s → %s", req.method, req.url, resp.status_code)
logger.debug("Request headers: %s", json.dumps(dict(req.headers), indent=2)[:2048])
if req.body:
body_preview = req.body if isinstance(req.body, str) else req.body.decode("utf-8", "ignore")
logger.debug("Request body: %s", body_preview[:2048])
logger.debug("Response headers: %s", json.dumps(dict(resp.headers), indent=2)[:2048])
logger.debug("Response body: %s", resp.text[:2048])
resp.raise_for_status()
return resp

────────────────────────────────────────────────────────────────────────────

IMAP helpers

────────────────────────────────────────────────────────────────────────────

def with_imap(logger: logging.Logger):
"""Context manager that yields a logged‑in imaplib.IMAP4_SSL instance."""

class _ImapCtx:
    def __enter__(self):
        self.conn = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT)
        self.conn.login(IMAP_USER, IMAP_PASS)
        self.conn.select("INBOX")
        logger.debug("IMAP connected to %s as %s", IMAP_HOST, IMAP_USER)
        return self.conn

    def __exit__(self, exc_type, exc, tb):
        try:
            self.conn.logout()
        except Exception:  # noqa: BLE001
            logger.exception("IMAP logout failed")

return _ImapCtx()

def find_confirmation_email(imap: imaplib.IMAP4_SSL, recipient: str, logger: logging.Logger, timeout: float = 60.0) -> Optional[str]:
"""Poll IMAP until a confirmation link for recipient is found, or timeout."""
end_time = time.time() + timeout
while time.time() < end_time:

Search for the newest mail to that recipient

    typ, ids = imap.search(None, "TO", f'"{recipient}"')
    if typ != "OK":
        logger.warning("IMAP search returned %s", typ)
        time.sleep(2)
        continue
    id_list = ids[0].split()
    if id_list:
        # use the first one (lowest id)
        msg_id = id_list[0]
        typ, msg_data = imap.fetch(msg_id, "(RFC822)")
        if typ != "OK":
            logger.warning("IMAP fetch %s failed", msg_id)
            time.sleep(2)
            continue
        raw_email = msg_data[0][1]
        message: Message = email.message_from_bytes(raw_email)
        html_part: Optional[str] = None
        if message.is_multipart():
            for part in message.walk():
                ctype = part.get_content_type()
                if ctype == "text/html":
                    html_part = part.get_payload(decode=True).decode(part.get_content_charset() or "utf-8", "ignore")
                    break
        else:
            if message.get_content_type() == "text/html":
                html_part = message.get_payload(decode=True).decode(message.get_content_charset() or "utf-8", "ignore")
        if html_part:
            m = CONFIRM_LINK_RE.search(html_part)
            if not m:
                # Handle meta‑refresh variant (rare)
                m2 = HTML_META_REFRESH_RE.search(html_part)
                confirm_link = m2.group(1) if m2 else None
            else:
                confirm_link = m.group(1)
            if confirm_link:
                logger.debug("Confirmation link found: %s", confirm_link)
                return confirm_link
    time.sleep(2)
return None

────────────────────────────────────────────────────────────────────────────

Worker

────────────────────────────────────────────────────────────────────────────

def worker(idx: int, outfile: str, delay: float) -> None:
log = logging.getLogger(f"worker-{idx}")
try:

1️⃣ Generate alias and password

    alias_local = f"cohere_{rand_str(6)}"
    email_addr = f"{alias_local}@{{ CACHE_ALL_DOMAIN }}"
    domain = "{{ CACHE_ALL_DOMAIN }}"
    password = make_cohere_password()
    log.info("[%d] Using address %s", idx, email_addr)

    # 2️⃣ RegisterWithEmail
    reg_payload = {
        "email": email_addr,
        "password": password,
        "freeCreditCode": "",
        "analyticsData": {
            "utm_campaign": "",
            "utm_channel": "",
            "utm_content": "",
            "utm_medium": "",
            "utm_source": "",
            "ga4_client_id": "",
            "segment_anonymous_id": "",
        },
    }
    headers = {**COHERE_HEADERS_COMMON, "X-Email-Domain": domain, "Content-Type": "application/json"}

    http_request(
        "POST",
        f"{COHERE_BASE}/RegisterWithEmail",
        logger=log,
        headers=headers,
        json=reg_payload,
    )

    time.sleep(delay)  # allow Cohere to send email

    # 3️⃣ Poll IMAP for confirmation e‑mail
    with with_imap(log) as imap:
        confirm_link = find_confirmation_email(imap, email_addr, log, timeout=120.0)
    if not confirm_link:
        raise RuntimeError("Confirmation email not found for %s" % email_addr)

    # 4️⃣ Follow confirm link to obtain access_token cookie
    sess = requests.Session()
    sess.headers.update(COHERE_HEADERS_COMMON)
    resp = sess.get(confirm_link, allow_redirects=True)
    if log.isEnabledFor(logging.DEBUG):
        log.debug("Confirm GET status %s", resp.status_code)
        log.debug("Cookies now: %s", sess.cookies.get_dict())
    if "access_token" not in sess.cookies:
        raise RuntimeError("access_token cookie not set after email confirmation")
    access_token = sess.cookies.get("access_token")

    # 5️⃣ AgreeToOnboarding
    onboard_payload = {
        "agreementType": "PROFILE_SETUP",
        "agreementData": {
            "name": "Zhi",
            "lastName": "Yang",
            "role": "",
            "otherRole": "",
            "use": "",
            "otherUse": "",
            "orgName": "",
            "marketingEmailsOpted": False,
        },
    }
    hdr_auth = {**COHERE_HEADERS_COMMON, "Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}
    http_request(
        "POST",
        f"{COHERE_BASE}/AgreeToOnboarding",
        logger=log,
        headers=hdr_auth,
        json=onboard_payload,
    )

    # 6️⃣ CreateAPIKey
    key_resp = http_request(
        "POST",
        f"{COHERE_BASE}/CreateAPIKey",
        logger=log,
        headers=hdr_auth,
        json={"name": "auto", "keyType": "TRIAL"},
    ).json()
    raw_key = key_resp["key"]["raw_key"]
    log.info("[%d] Key obtained: %s…", idx, raw_key[:8])

    # 7️⃣ Persist
    with threading.Lock():
        with open(outfile, "a", encoding="utf-8") as fp:
            fp.write(f"{raw_key}\n")

except Exception as exc:
    log.exception("Worker %d failed: %s", idx, exc)

────────────────────────────────────────────────────────────────────────────

CLI entry‑point

────────────────────────────────────────────────────────────────────────────

def main() -> None:
ap = argparse.ArgumentParser(description="Automate Cohere account creation via IMAP confirmation.")
ap.add_argument("-n", "--num", type=int, required=True, help="Number of accounts to create")
ap.add_argument("-t", "--threads", type=int, default=1, help="Concurrent threads (default 1)")
ap.add_argument("-o", "--outfile", default="cohere_keys.txt", help="File to append raw keys to")
ap.add_argument("-d", "--debug", action="store_true", help="Enable DEBUG logging")
ap.add_argument("--delay", type=float, default=3.0, help="Seconds to wait before polling mailbox")
args = ap.parse_args()

logging.basicConfig(
    level=logging.DEBUG if args.debug else logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)

with ThreadPoolExecutor(max_workers=args.threads) as pool:
    futs = {pool.submit(worker, i + 1, args.outfile, args.delay): i for i in range(args.num)}
    for f in as_completed(futs):
        f.result()  # propagate exceptions

if name == "main":
try:
main()
except KeyboardInterrupt:
print("Interrupted – exiting…", file=os.sys.stderr)`

    随便折腾一下.

使用方法:

将 {{ CACHE_ALL_DOMAIN }} 替换为你的 CF Cache-All 域名
将常量改为你的IMAP信息
目前问题

正则表达式无法正常匹配链接