Repository: yt-dlp/yt-dlp
Pull request #9952 (2024-05-18 06:37:29)

[ie/twitter] Fix auth for x.com migration

Addresses https://github.com/yt-dlp/yt-dlp/issues/9923#issuecomment-2118554769


<details open><summary>Template</summary> <!-- OPEN is intentional -->

### Before submitting a *pull request* make sure you have:
- [x] At least skimmed through [contributing guidelines](https://github.com/yt-dlp/yt-dlp/blob/master/CONTRIBUTING.md#developer-instructions) including [yt-dlp coding conventions](https://github.com/yt-dlp/yt-dlp/blob/master/CONTRIBUTING.md#yt-dlp-coding-conventions)
- [x] [Searched](https://github.com/yt-dlp/yt-dlp/search?q=is%3Apr&type=Issues) the bugtracker for similar pull requests
- [x] Checked the code with [flake8](https://pypi.python.org/pypi/flake8) and [ran relevant tests](https://github.com/yt-dlp/yt-dlp/blob/master/CONTRIBUTING.md#developer-instructions)

### In order to be accepted and merged into yt-dlp each piece of code must be in public domain or released under [Unlicense](http://unlicense.org/). Check all of the following options that apply:
- [x] I am the original author of this code and I am willing to release it under [Unlicense](http://unlicense.org/)

### What is the purpose of your *pull request*?
- [x] Fix or improvement to an extractor (Make sure to add/update tests)

</details>



Patch:
diff --git a/yt_dlp/extractor/twitter.py b/yt_dlp/extractor/twitter.py
index df7f816bd..fc80dade8 100644
--- a/yt_dlp/extractor/twitter.py
+++ b/yt_dlp/extractor/twitter.py
@@ -34,8 +34,8 @@
 
 class TwitterBaseIE(InfoExtractor):
     _NETRC_MACHINE = 'twitter'
-    _API_BASE = 'https://api.twitter.com/1.1/'
-    _GRAPHQL_API_BASE = 'https://twitter.com/i/api/graphql/'
+    _API_BASE = 'https://api.x.com/1.1/'
+    _GRAPHQL_API_BASE = 'https://x.com/i/api/graphql/'
     _BASE_REGEX = r'https?://(?:(?:www|m(?:obile)?)\.)?(?:(?:twitter|x)\.com|twitter3e4tixl4xyajtrzo62zg5vztmjuricljdp2c5kshju4avyoid\.onion)/'
     _AUTH = 'AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA'
     _LEGACY_AUTH = 'AAAAAAAAAAAAAAAAAAAAAIK1zgAAAAAA2tUWuhGZ2JceoId5GwYWU5GspY4%3DUq7gzFoCZs1QfwGoVdvSac3IniczZEYXIcDyumCauIXpcAPorE'
@@ -153,6 +153,14 @@ def _search_dimensions_in_video_url(a_format, video_url):
     def is_logged_in(self):
         return bool(self._get_cookies(self._API_BASE).get('auth_token'))
 
+    # XXX: Temporary workaround until twitter.com => x.com migration is completed
+    def _real_initialize(self):
+        if self.is_logged_in or not self._get_cookies('https://twitter.com/').get('auth_token'):
+            return
+        # User has not yet been migrated to x.com and has passed twitter.com cookies
+        TwitterBaseIE._API_BASE = 'https://api.twitter.com/1.1/'
+        TwitterBaseIE._GRAPHQL_API_BASE = 'https://twitter.com/i/api/graphql/'
+
     @functools.cached_property
     def _selected_api(self):
         return self._configuration_arg('api', ['graphql'], ie_key='Twitter')[0]
@@ -196,17 +204,15 @@ def _perform_login(self, username, password):
         if self.is_logged_in:
             return
 
-        webpage = self._download_webpage('https://twitter.com/', None, 'Downloading login page')
-        guest_token = self._search_regex(
-            r'\.cookie\s*=\s*["\']gt=(\d+);', webpage, 'gt', default=None) or self._fetch_guest_token(None)
+        guest_token = self._fetch_guest_token(None)
         headers = {
             **self._set_base_headers(),
             'content-type': 'application/json',
             'x-guest-token': guest_token,
             'x-twitter-client-language': 'en',
             'x-twitter-active-user': 'yes',
-            'Referer': 'https://twitter.com/',
-            'Origin': 'https://twitter.com',
+            'Referer': 'https://x.com/',
+            'Origin': 'https://x.com',
         }
 
         def build_login_json(*subtask_inputs):
diff --git a/yt_dlp/networking/common.py b/yt_dlp/networking/common.py
index a2217034c..d473e16c5 100644
--- a/yt_dlp/networking/common.py
+++ b/yt_dlp/networking/common.py
@@ -31,6 +31,8 @@
 )
 from ..utils.networking import HTTPHeaderDict, normalize_url
 
+DEFAULT_TIMEOUT = 20
+
 
 def register_preference(*handlers: type[RequestHandler]):
     assert all(issubclass(handler, RequestHandler) for handler in handlers)
@@ -235,7 +237,7 @@ def __init__(
         self._logger = logger
         self.headers = headers or {}
         self.cookiejar = cookiejar if cookiejar is not None else YoutubeDLCookieJar()
-        self.timeout = float(timeout or 20)
+        self.timeout = float(timeout or DEFAULT_TIMEOUT)
         self.proxies = proxies or {}
         self.source_address = source_address
         self.verbose = verbose


Test patch:
diff --git a/test/test_networking.py b/test/test_networking.py
index 994467014..d127cbb94 100644
--- a/test/test_networking.py
+++ b/test/test_networking.py
@@ -6,7 +6,7 @@
 
 import pytest
 
-from yt_dlp.networking.common import Features
+from yt_dlp.networking.common import Features, DEFAULT_TIMEOUT
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
@@ -523,20 +523,17 @@ def test_read_timeout(self, handler):
     def test_connect_timeout(self, handler):
         # nothing should be listening on this port
         connect_timeout_url = 'http://10.255.255.255'
-        with handler(timeout=0.01) as rh:
+        with handler(timeout=0.01) as rh, pytest.raises(TransportError):
             now = time.time()
-            with pytest.raises(TransportError):
-                validate_and_send(
-                    rh, Request(connect_timeout_url))
-            assert 0.01 <= time.time() - now < 20
+            validate_and_send(rh, Request(connect_timeout_url))
+        assert time.time() - now < DEFAULT_TIMEOUT
 
-        with handler() as rh:
-            with pytest.raises(TransportError):
-                # Per request timeout, should override handler timeout
-                now = time.time()
-                validate_and_send(
-                    rh, Request(connect_timeout_url, extensions={'timeout': 0.01}))
-                assert 0.01 <= time.time() - now < 20
+        # Per request timeout, should override handler timeout
+        request = Request(connect_timeout_url, extensions={'timeout': 0.01})
+        with handler() as rh, pytest.raises(TransportError):
+            now = time.time()
+            validate_and_send(rh, request)
+        assert time.time() - now < DEFAULT_TIMEOUT
 
     def test_source_address(self, handler):
         source_address = f'127.0.0.{random.randint(5, 255)}'
diff --git a/test/test_websockets.py b/test/test_websockets.py
index bc9f2187a..aa0dfa2d5 100644
--- a/test/test_websockets.py
+++ b/test/test_websockets.py
@@ -3,11 +3,12 @@
 # Allow direct execution
 import os
 import sys
+import time
 
 import pytest
 
 from test.helper import verify_address_availability
-from yt_dlp.networking.common import Features
+from yt_dlp.networking.common import Features, DEFAULT_TIMEOUT
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
@@ -202,11 +203,26 @@ def test_raise_http_error(self, handler, status):
         ({'timeout': sys.float_info.min}, {}),
         ({}, {'timeout': sys.float_info.min}),
     ])
-    def test_timeout(self, handler, params, extensions):
+    def test_read_timeout(self, handler, params, extensions):
         with handler(**params) as rh:
             with pytest.raises(TransportError):
                 ws_validate_and_send(rh, Request(self.ws_base_url, extensions=extensions))
 
+    def test_connect_timeout(self, handler):
+        # nothing should be listening on this port
+        connect_timeout_url = 'ws://10.255.255.255'
+        with handler(timeout=0.01) as rh, pytest.raises(TransportError):
+            now = time.time()
+            ws_validate_and_send(rh, Request(connect_timeout_url))
+        assert time.time() - now < DEFAULT_TIMEOUT
+
+        # Per request timeout, should override handler timeout
+        request = Request(connect_timeout_url, extensions={'timeout': 0.01})
+        with handler() as rh, pytest.raises(TransportError):
+            now = time.time()
+            ws_validate_and_send(rh, request)
+        assert time.time() - now < DEFAULT_TIMEOUT
+
     def test_cookies(self, handler):
         cookiejar = YoutubeDLCookieJar()
         cookiejar.set_cookie(http.cookiejar.Cookie(