From 02434aec6a7b2fc16b76730c2240806bbeb0eef8 Mon Sep 17 00:00:00 2001 From: gimmelovej Date: Wed, 24 Jun 2026 00:04:37 -0300 Subject: [PATCH] fix: use curl_cffi to bypass Cloudflare TLS fingerprinting --- xcompare.py | 119 ++++++++++++++++++++++++++-------------------------- 1 file changed, 60 insertions(+), 59 deletions(-) diff --git a/xcompare.py b/xcompare.py index 6107cb7..9018681 100644 --- a/xcompare.py +++ b/xcompare.py @@ -1,9 +1,8 @@ #!/usr/bin/env python3 """ Find mutual followers between two X (Twitter) accounts without the official API. - -Built and maintained by OffSeq (https://offseq.com/ | https://x.com/offseq | -https://www.linkedin.com/company/offseq | https://infosec.exchange/@offseq). +Built and maintained by OffSeq (https://offseq.com/). +Modified to use curl_cffi to bypass Cloudflare TLS fingerprinting. """ from __future__ import annotations @@ -13,15 +12,20 @@ import os import sys import time -import urllib.error import urllib.parse -import urllib.request from dataclasses import dataclass from typing import Dict, Iterable, List, Optional, Tuple +try: + from curl_cffi import requests as cffi_requests +except ImportError: + print("Error: curl_cffi not installed. Run: pip install curl_cffi", file=sys.stderr) + sys.exit(1) + PUBLIC_WEB_BEARER = urllib.parse.unquote( "AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA" ) + GUEST_ACTIVATE_URL = "https://x.com/i/api/1.1/guest/activate.json" USER_SHOW_URL = "https://x.com/i/api/1.1/users/show.json" USER_BY_USERNAME_V2 = "https://x.com/i/api/2/users/by/username/{username}" @@ -60,7 +64,7 @@ def parse_args() -> argparse.Namespace: "--delay", type=float, default=1.0, - help="Seconds to wait between paginated API calls to stay under rate limits (default: 1).", + help="Seconds to wait between paginated API calls (default: 1).", ) parser.add_argument( "--json", @@ -80,7 +84,7 @@ def parse_args() -> argparse.Namespace: return parser.parse_args() -def read_cookie_source(args: argparse.Namespace) -> Optional[CookieBundle]: +def read_cookie_source(args: argparse.Namespace) -> Optional["CookieBundle"]: raw_cookie: Optional[str] = None if args.cookie: raw_cookie = args.cookie @@ -93,6 +97,7 @@ def read_cookie_source(args: argparse.Namespace) -> Optional[CookieBundle]: if raw_cookie is None: return None + normalized_cookie, bearer = normalize_cookie_blob(raw_cookie) if not normalized_cookie: raise XCompareError("Cookie input was empty; check the provided --cookie/--cookie-file.") @@ -110,24 +115,25 @@ def parse_cookie_string(raw_cookie: str) -> Dict[str, str]: def resolve_bearer_token( - args: argparse.Namespace, bundle: Optional[CookieBundle] + args: argparse.Namespace, bundle: Optional["CookieBundle"] ) -> str: if args.web_bearer: manual = _sanitize_bearer(args.web_bearer) if not manual: raise XCompareError("Provided --web-bearer value was empty.") return manual + env_token = _sanitize_bearer(os.getenv("X_WEB_BEARER_TOKEN")) if env_token: return env_token + if bundle and bundle.bearer: return bundle.bearer + return PUBLIC_WEB_BEARER def normalize_cookie_blob(blob: str) -> Tuple[str, Optional[str]]: - """Accept either a raw Cookie header or a JSON export from DevTools.""" - stripped = blob.strip() if not stripped: return "", None @@ -142,6 +148,7 @@ def normalize_cookie_blob(blob: str) -> Tuple[str, Optional[str]]: cookie_value = cookie_value or "" bearer_value = _sanitize_bearer(bearer_value) return cookie_value, bearer_value + return stripped, None @@ -186,53 +193,55 @@ class XWebSession: cookie_map: Dict[str, str] guest_token: Optional[str] bearer_token: str + _session: object = None # curl_cffi session + + def __post_init__(self): + # Create a persistent curl_cffi session impersonating Chrome + self._session = cffi_requests.Session(impersonate="chrome124") @classmethod def create(cls, raw_cookie: Optional[str], bearer_token: Optional[str]) -> "XWebSession": token = bearer_token or PUBLIC_WEB_BEARER cookie_map = parse_cookie_string(raw_cookie) if raw_cookie else {} guest_token = None - if not raw_cookie: - guest_token = cls._activate_guest_token(token) - return cls( + + instance = cls( raw_cookie=raw_cookie, cookie_map=cookie_map, guest_token=guest_token, bearer_token=token, ) - @staticmethod - def _activate_guest_token(bearer_token: str) -> str: - payload = json.dumps({"client": "web"}).encode() + if not raw_cookie: + instance.guest_token = instance._activate_guest_token(token) + + return instance + + def _activate_guest_token(self, bearer_token: str) -> str: headers = { "Authorization": f"Bearer {bearer_token}", "Content-Type": "application/json", "Accept": "application/json", - "User-Agent": "Mozilla/5.0", "Origin": "https://x.com", "Referer": "https://x.com/", "x-twitter-active-user": "yes", "x-twitter-client-language": "en", "x-twitter-auth-type": "OAuth2Session", } - request = urllib.request.Request( - GUEST_ACTIVATE_URL, - data=payload, - headers=headers, - method="POST", - ) + try: - with urllib.request.urlopen(request) as response: - payload = json.loads(response.read()) - except urllib.error.HTTPError as exc: - body = exc.read().decode() if hasattr(exc, "read") else "" + response = self._session.post( + GUEST_ACTIVATE_URL, + json={"client": "web"}, + headers=headers, + ) + response.raise_for_status() + payload = response.json() + except Exception as exc: raise XCompareError( - f"Guest activation failed ({exc.code} {exc.reason}" - + (f": {body[:200].strip()}" if body else "") - + "). X may be blocking anonymous sessions; pass --cookie with a logged-in browser session." + f"Guest activation failed: {exc}. " + "X may be blocking anonymous sessions; pass --cookie with a logged-in browser session." ) from exc - except urllib.error.URLError as exc: - raise XCompareError(f"Network error: {exc.reason}") from exc token = payload.get("guest_token") if not token: @@ -240,9 +249,8 @@ def _activate_guest_token(bearer_token: str) -> str: return token def headers(self) -> Dict[str, str]: - headers = { + hdrs = { "Authorization": f"Bearer {self.bearer_token}", - "User-Agent": "Mozilla/5.0", "Accept": "application/json, text/plain, */*", "Accept-Language": "en-US,en;q=0.9", "Origin": "https://x.com", @@ -252,25 +260,20 @@ def headers(self) -> Dict[str, str]: "x-twitter-auth-type": "OAuth2Session", } if self.raw_cookie: - headers["Cookie"] = self.raw_cookie + hdrs["Cookie"] = self.raw_cookie if "ct0" in self.cookie_map: - headers["x-csrf-token"] = self.cookie_map["ct0"] + hdrs["x-csrf-token"] = self.cookie_map["ct0"] elif self.guest_token: - headers["x-guest-token"] = self.guest_token - return headers + hdrs["x-guest-token"] = self.guest_token + return hdrs def get(self, url: str, params: Optional[Dict[str, str]] = None) -> Dict: - query = urllib.parse.urlencode(params or {}) - full_url = f"{url}?{query}" if query else url - request = urllib.request.Request(full_url, headers=self.headers()) try: - with urllib.request.urlopen(request) as response: - return json.loads(response.read()) - except urllib.error.HTTPError as exc: - body = exc.read().decode() if hasattr(exc, "read") else "" - raise XCompareError(f"Request failed: {exc.code} {exc.reason}: {body}") from exc - except urllib.error.URLError as exc: - raise XCompareError(f"Network error: {exc.reason}") from exc + response = self._session.get(url, params=params or {}, headers=self.headers()) + response.raise_for_status() + return response.json() + except Exception as exc: + raise XCompareError(f"Request failed: {exc}") from exc def fetch_user(session: XWebSession, username: str) -> Dict: @@ -304,9 +307,7 @@ def _fetch_user_v1(session: XWebSession, username: str) -> Dict: def _fetch_user_v2(session: XWebSession, username: str) -> Dict: url = USER_BY_USERNAME_V2.format(username=username) - params = { - "user.fields": "id,name,username,verified,protected,public_metrics", - } + params = {"user.fields": "id,name,username,verified,protected,public_metrics"} payload = session.get(url, params=params) data = payload.get("data") if not data: @@ -328,11 +329,7 @@ def _looks_like_missing_user(exc: XCompareError) -> bool: def _fetch_user_typeahead(session: XWebSession, username: str) -> Dict: - params = { - "q": username, - "src": "search_box", - "result_type": "users", - } + params = {"q": username, "src": "search_box", "result_type": "users"} payload = session.get("https://x.com/i/api/1.1/search/typeahead.json", params=params) users = payload.get("users", []) username_lower = username.lower() @@ -341,7 +338,6 @@ def _fetch_user_typeahead(session: XWebSession, username: str) -> Dict: if handle == username_lower: return user if users: - # fall back to first if exact match missing return users[0] raise XCompareError(f"Unable to resolve @{username} via typeahead") @@ -366,6 +362,7 @@ def fetch_followers( } payload = session.get(FOLLOWERS_LIST_URL, params=params) users = payload.get("users", []) + for follower in users: collected[follower["id_str"]] = follower if len(collected) >= limit: @@ -373,6 +370,7 @@ def fetch_followers( next_cursor = str(payload.get("next_cursor_str") or payload.get("next_cursor") or "0") cursor = next_cursor + log( f"Fetched {len(collected)} follower(s) so far for user {user_id}" + ("" if cursor in ("0", "0.0") else " (more available)") @@ -380,6 +378,7 @@ def fetch_followers( if cursor in ("0", "0.0") or not users: break + if delay > 0: time.sleep(delay) @@ -428,7 +427,6 @@ def print_table(users: Iterable[Dict]) -> None: if not users: print("No overlapping followers found.") return - print(f"Found {len(users)} overlapping follower(s):\n") for user in users: print(f" • {describe_user(user)}") @@ -438,7 +436,6 @@ def build_logger(quiet: bool): def log(message: str) -> None: if not quiet: print(message, file=sys.stderr) - return log @@ -450,9 +447,12 @@ def main() -> None: cookie_bundle = read_cookie_source(args) bearer_token = resolve_bearer_token(args, cookie_bundle) cookie_value = cookie_bundle.cookie if cookie_bundle else None + session = XWebSession.create(cookie_value, bearer_token=bearer_token) + user_a = fetch_user(session, args.user_a) user_b = fetch_user(session, args.user_b) + log(f"Resolved @{args.user_a} -> {user_a['id_str']}") log(f"Resolved @{args.user_b} -> {user_b['id_str']}") @@ -472,6 +472,7 @@ def main() -> None: ) overlaps = overlapping_followers(followers_a, followers_b) + except XCompareError as exc: print(f"Error: {exc}", file=sys.stderr) sys.exit(1)