From 456c32512bd5129c5ea9bc8f3d8081b9cefc3bb4 Mon Sep 17 00:00:00 2001 From: Blacksuite Date: Wed, 12 Aug 2026 12:37:47 +0200 Subject: [PATCH] fix: send session cookies when fetching x.com for ClientTransaction init MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _ensure_client_transaction() fetched https://x.com without cookies, so x.com returned the logged-out landing page. That page contains no ondemand.s marker, so ON_DEMAND_FILE_REGEX.search() returns None and get_ondemand_file_url() raises "'NoneType' object has no attribute 'group'". ClientTransaction then stays None and every API request goes out without an x-client-transaction-id header, which X answers with HTTP 404 — including SearchTimeline and account verification endpoints. Measured against x.com: anonymous returns ~35KB with zero occurrences of "ondemand"; the same request with auth_token/ct0 cookies returns ~278KB with the marker present. _build_headers() already builds exactly this Cookie value for API calls; this reuses it for the init fetch. Fixes #78 Fixes #73 Fixes #69 --- tests/test_client.py | 51 +++++++++++++++++++++++++++++++++++++++++++ twitter_cli/client.py | 7 ++++++ 2 files changed, 58 insertions(+) diff --git a/tests/test_client.py b/tests/test_client.py index c1393d3..dd4c15a 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1545,3 +1545,54 @@ def mock_post(operation_name, variables, features=None): assert captured.get("product") == "Latest" assert captured.get("querySource") == "typed_query" + + +# ── TwitterClient._ensure_client_transaction ───────────────────────────── + +class TestEnsureClientTransaction: + """The x.com fetch must be authenticated. + + Anonymous requests receive the logged-out landing page, which carries no + ondemand.s marker, so get_ondemand_file_url() fails and every subsequent + API call goes out without an x-client-transaction-id header. + """ + + def _make_client(self): + client = TwitterClient.__new__(TwitterClient) + client._auth_token = "test_token" + client._ct0 = "test_ct0" + client._cookie_string = None + client._client_transaction = None + client._ct_init_attempted = False + return client + + @patch("twitter_cli.client._gen_ct_headers", return_value={}) + @patch("twitter_cli.client._get_cffi_session") + def test_homepage_fetch_sends_cookies(self, mock_session, mock_ct_headers): + session = MagicMock() + session.get = MagicMock(side_effect=Exception("stop after first fetch")) + mock_session.return_value = session + + client = self._make_client() + client._load_ct_cache = lambda: False + client._ensure_client_transaction() + + _, kwargs = session.get.call_args + cookie = kwargs["headers"]["Cookie"] + assert "auth_token=test_token" in cookie + assert "ct0=test_ct0" in cookie + + @patch("twitter_cli.client._gen_ct_headers", return_value={}) + @patch("twitter_cli.client._get_cffi_session") + def test_full_cookie_string_preferred(self, mock_session, mock_ct_headers): + session = MagicMock() + session.get = MagicMock(side_effect=Exception("stop after first fetch")) + mock_session.return_value = session + + client = self._make_client() + client._cookie_string = "auth_token=a; ct0=b; guest_id=c" + client._load_ct_cache = lambda: False + client._ensure_client_transaction() + + _, kwargs = session.get.call_args + assert kwargs["headers"]["Cookie"] == "auth_token=a; ct0=b; guest_id=c" diff --git a/twitter_cli/client.py b/twitter_cli/client.py index 0436c8e..8232190 100644 --- a/twitter_cli/client.py +++ b/twitter_cli/client.py @@ -1103,6 +1103,13 @@ def _ensure_client_transaction(self): # a different TLS fingerprint on the same IP — a detection vector. cffi_session = _get_cffi_session() ct_headers = _gen_ct_headers() + # x.com serves the logged-out landing page to anonymous requests, + # and that page contains no ondemand.s marker for + # get_ondemand_file_url() to match. Send the session cookies so we + # receive the authenticated app shell instead. + ct_headers["Cookie"] = self._cookie_string or ( + "auth_token=%s; ct0=%s" % (self._auth_token, self._ct0) + ) home_page = cffi_session.get( "https://x.com", headers=ct_headers, timeout=10, )