jolly/test_sessions.py

test_sessions.py · 11 KiB · Python Raw

#!/usr/bin/env python3
# SPDX-License-Identifier: AGPL-3.0-only
# Python port of Nitter's tools/test_sessions.nim
#
# Test each cookie session in a JSONL file by hitting the UserMedia endpoint.
#
# Usage:
#   python3 test_sessions.py [sessions_file] [--delay N]
#
# Examples:
#   python3 test_sessions.py
#   python3 test_sessions.py sessions_cookie.jsonl --delay 1
#
# Requires: pip install requests

import base64
import hashlib
import json
import os
import random
import sys
import time
from urllib.parse import quote, urlencode, urlparse

import requests

# ---------------------------------------------------------------- consts.nim

BEARER_TOKEN = "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA"
BEARER_TOKEN2 = "Bearer AAAAAAAAAAAAAAAAAAAAAFXzAwAAAAAAMHCxpeSDG1gLNLghVe8d74hl6k4%3DRUMF4xAQLsbeBhTSRrCiQpJtxoGWeyHrDb5te2jpGskWDFW82F"

GRAPH_USER_MEDIA = "VyudDWQnr9vJNw7GasFz2g/UserMedia"

def _squash(s: str) -> str:
    return s.replace(" ", "").replace("\n", "")

GQL_FEATURES = _squash("""{
  "rweb_video_screen_enabled": false,
  "rweb_cashtags_enabled": true,
  "profile_label_improvements_pcf_label_in_post_enabled": true,
  "responsive_web_profile_redirect_enabled": false,
  "rweb_tipjar_consumption_enabled": false,
  "verified_phone_label_enabled": false,
  "creator_subscriptions_tweet_preview_api_enabled": true,
  "responsive_web_graphql_timeline_navigation_enabled": true,
  "responsive_web_graphql_skip_user_profile_image_extensions_enabled": false,
  "premium_content_api_read_enabled": false,
  "withQuickPromoteEligibilityTweetFields": true
}""")

USER_MEDIA_VARS = _squash("""{
  "userId": "%s", %s
  "count": %s,
  "includePromotedContent": false,
  "withClientEventToken": false,
  "withBirdwatchNotes": false,
  "withVoice": true
}""")

# jack (id=12) - public account with media, same target as ratelimit_probe.py
TEST_VARS = USER_MEDIA_VARS % ("12", "", "20")

# --------------------------------------------------------------------- tid.nim

DEFAULT_KEYWORD = "obfiowerehiring"
PAIRS_URL = ("https://raw.githubusercontent.com/fa0311/"
             "x-client-transaction-id-pair-dict/refs/heads/main/pair.json")
PAIRS_TTL_SEC = 60 * 60  # refresh every hour

_cached_pairs: list = []
_last_cached = 0

def _get_pair() -> dict:
    global _cached_pairs, _last_cached
    if not _cached_pairs or int(time.time()) - _last_cached > PAIRS_TTL_SEC:
        resp = _http.get(PAIRS_URL, timeout=30)
        if resp.status_code == 200:
            pairs = resp.json()
            if not pairs:
                raise ValueError("Parsing pairs failed: " + resp.text)
            _cached_pairs = pairs
            _last_cached = int(time.time())

if not _cached_pairs:
        raise ValueError("Failed to fetch x-client-transaction-id pairs")

return random.choice(_cached_pairs)

def _b64decode_nopad(data: str) -> bytes:
    return base64.b64decode(data + "=" * (-len(data) % 4))

def gen_tid(path: str) -> str:
    pair = _get_pair()

time_now = int(time.time() - 1682924400)
    time_now_bytes = bytes([
        time_now & 0xFF,
        (time_now >> 8) & 0xFF,
        (time_now >> 16) & 0xFF,
        (time_now >> 24) & 0xFF,
    ])

data = "GET!" + path + "!" + str(time_now) + DEFAULT_KEYWORD + pair["animationKey"]
    hash_bytes = hashlib.sha256(data.encode()).digest()
    key_bytes = _b64decode_nopad(pair["verification"])
    bytes_arr = key_bytes + time_now_bytes + hash_bytes[:16] + bytes([3])
    random_num = random.randrange(256)
    tid = bytes([random_num]) + bytes(b ^ random_num for b in bytes_arr)

return base64.b64encode(tid).decode().replace("=", "")

# ------------------------------------------------------- types / auth / session

class Session:
    def __init__(self, kind, id, username, auth_token="", ct0="",
                 oauth_token="", oauth_secret=""):
        self.kind = kind  # "oauth" | "cookie"
        self.id = id
        self.username = username
        self.auth_token = auth_token
        self.ct0 = ct0
        self.oauth_token = oauth_token
        self.oauth_secret = oauth_secret

def pretty(self) -> str:
        if self.id > 0 and self.username:
            result = f"{self.id} ({self.username})"
        elif self.username:
            result = self.username
        elif self.id > 0:
            result = str(self.id)
        else:
            result = "<unknown>"
        return f"{self.kind} {result}"

def parse_session(raw: str) -> Session:
    s = json.loads(raw)
    kind = s.get("kind") or "oauth"
    username = s.get("username", "") or ""

if kind == "oauth":
        token = s.get("oauth_token", "") or ""
        # mirrors Nim's token[0 ..< token.find('-')]
        sid = token[:token.find("-")]
        return Session("oauth", int(sid), username,
                       oauth_token=token,
                       oauth_secret=s.get("oauth_token_secret", "") or "")
    if kind == "cookie":
        raw_id = str(s.get("id", "") or "")
        return Session("cookie", int(raw_id) if raw_id else 0, username,
                       auth_token=s.get("auth_token", "") or "",
                       ct0=s.get("ct0", "") or "")
    raise ValueError("Unknown session kind: " + kind)

# ------------------------------------------------------------------ apiutils.nim

def build_url() -> str:
    params = [("variables", TEST_VARS), ("features", GQL_FEATURES)]
    prefix = "" if GRAPH_USER_MEDIA.startswith("1.1/") else "graphql/"
    return ("https://x.com/i/api/" + prefix + GRAPH_USER_MEDIA
            + "?" + urlencode(params, quote_via=quote))

def gen_headers(session: Session, url: str, skip_tid: bool, disable_tid: bool) -> dict:
    h = {
        "Accept": "*/*",
        "Accept-Encoding": "gzip",
        "Accept-Language": "en-US,en;q=0.9",
        "Content-Type": "application/json",
        "Origin": "https://x.com",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                      "(KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36",
        "X-Twitter-Active-User": "yes",
        "X-Twitter-Client-Language": "en",
        "Priority": "u=1, i",
    }

# only cookie sessions are tested, so the oauth branch is never reached
    path = urlparse(url).path
    h["X-Twitter-Auth-Type"] = "OAuth2Session"
    h["X-Csrf-Token"] = session.ct0
    h["Cookie"] = f"auth_token={session.auth_token}; ct0={session.ct0}"
    h["Referer"] = "https://x.com/"
    h["Sec-Ch-Ua"] = '"Google Chrome";v="142", "Chromium";v="142", "Not A(Brand";v="24"'
    h["Sec-Ch-Ua-Mobile"] = "?0"
    h["Sec-Ch-Ua-Platform"] = "Windows"
    h["Sec-Fetch-Dest"] = "empty"
    h["Sec-Fetch-Mode"] = "cors"
    h["Sec-Fetch-Site"] = "same-origin"
    if disable_tid or skip_tid or "/1.1/" in path:
        h["Authorization"] = BEARER_TOKEN2
    else:
        h["Authorization"] = BEARER_TOKEN
        h["X-Client-Transaction-Id"] = gen_tid(path)
    return h

# ----------------------------------------------------------- test_sessions.nim

# no proxy, and ignore HTTP(S)_PROXY from the environment (Nim's client does too)
_http = requests.Session()
_http.trust_env = False

def parse_cookie_sessions(path: str) -> list:
    result = []
    skipped = 0
    with open(path, encoding="utf-8") as f:
        for line in f:
            s = line.strip()
            if not s:
                continue
            try:
                sess = parse_session(s)
                if sess.kind == "cookie":
                    result.append(sess)
                else:
                    skipped += 1
            except Exception as e:
                print("  [!] Parse error:", e)
    if skipped > 0:
        print("  (skipped", skipped, "non-cookie sessions)")
    return result

def test_session(session: Session, url: str):
    """Returns (ok, code, remaining, limit, reset, detail)."""
    headers = gen_headers(session, url, skip_tid=False, disable_tid=False)

try:
        resp = _http.get(url, headers=headers, timeout=30)
        body = resp.text  # requests transparently handles gzip

remaining = limit = reset = 0
        if "x-rate-limit-reset" in resp.headers:
            reset = int(resp.headers["x-rate-limit-reset"])
        if "x-rate-limit-remaining" in resp.headers:
            remaining = int(resp.headers["x-rate-limit-remaining"])
        if "x-rate-limit-limit" in resp.headers:
            limit = int(resp.headers["x-rate-limit-limit"])

if resp.status_code == 200:
            return (True, resp.status_code, remaining, limit, reset, "")
        detail = body if 1 <= len(body) <= 120 else ""
        return (False, resp.status_code, remaining, limit, reset, detail)
    except Exception as e:
        return (False, 0, 0, 0, 0, str(e)[:120])

def main():
    sessions_path = "sessions.jsonl"
    delay = 500  # ms
    args = sys.argv[1:]
    i = 0

while i < len(args):
        arg = args[i]
        if arg == "--delay":
            i += 1
            if i >= len(args):
                print("Error: --delay requires a value (seconds)")
                sys.exit(1)
            delay = int(float(args[i]) * 1000)
        elif arg in ("--help", "-h"):
            print("Usage: test_sessions.py [sessions_file] [--delay N]")
            return
        else:
            sessions_path = arg
        i += 1

if not os.path.isfile(sessions_path):
        print("File not found:", sessions_path)
        sys.exit(1)

sessions = parse_cookie_sessions(sessions_path)
    if not sessions:
        print("No cookie sessions found in", sessions_path)
        sys.exit(0)

print("Testing", len(sessions), "sessions from", sessions_path)
    print("")

url = build_url()
    n_valid = n_fail = 0
    failures = []

for idx, session in enumerate(sessions):
        num = idx + 1
        prefix = f"[{num}/{len(sessions)}] {session.pretty()}"
        ok, code, remaining, limit, reset, detail = test_session(session, url)

if ok:
            n_valid += 1
            extra = f"  remaining={remaining}/{limit}" if limit > 0 else ""
            if reset > 0:
                mins = max(0, reset - int(time.time())) // 60
                extra += f"  resets in {mins}m ({time.strftime('%H:%M:%S', time.localtime(reset))})"
            print(prefix, "  \u2713 ", code, extra, sep="")
        else:
            n_fail += 1
            d = str(code) if code > 0 else "error: " + detail
            failures.append(session.pretty() + "  " + d)
            print(prefix, "  \u2717 ", d, sep="")

if num < len(sessions) and delay > 0:
            time.sleep(delay / 1000)

print("")
    print(f"Done: {n_valid} valid, {n_fail} invalid  [{len(sessions)} total]")

if failures:
        print("")
        print("Failures:")
        for f in failures:
            print("  " + f)

if __name__ == "__main__":
    main()

1	#!/usr/bin/env python3
2	# SPDX-License-Identifier: AGPL-3.0-only
3	# Python port of Nitter's tools/test_sessions.nim
4	#
5	# Test each cookie session in a JSONL file by hitting the UserMedia endpoint.
6	#
7	# Usage:
8	# python3 test_sessions.py [sessions_file] [--delay N]
9	#
10	# Examples:
11	# python3 test_sessions.py
12	# python3 test_sessions.py sessions_cookie.jsonl --delay 1
13	#
14	# Requires: pip install requests
15
16	import base64
17	import hashlib
18	import json
19	import os
20	import random
21	import sys
22	import time
23	from urllib.parse import quote, urlencode, urlparse
24
25	import requests
26
27	# ---------------------------------------------------------------- consts.nim
28
29	BEARER_TOKEN = "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA"
30	BEARER_TOKEN2 = "Bearer AAAAAAAAAAAAAAAAAAAAAFXzAwAAAAAAMHCxpeSDG1gLNLghVe8d74hl6k4%3DRUMF4xAQLsbeBhTSRrCiQpJtxoGWeyHrDb5te2jpGskWDFW82F"
31
32	GRAPH_USER_MEDIA = "VyudDWQnr9vJNw7GasFz2g/UserMedia"
33
34
35	def _squash(s: str) -> str:
36	return s.replace(" ", "").replace("\n", "")
37
38
39	GQL_FEATURES = _squash("""{
40	"rweb_video_screen_enabled": false,
41	"rweb_cashtags_enabled": true,
42	"profile_label_improvements_pcf_label_in_post_enabled": true,
43	"responsive_web_profile_redirect_enabled": false,
44	"rweb_tipjar_consumption_enabled": false,
45	"verified_phone_label_enabled": false,
46	"creator_subscriptions_tweet_preview_api_enabled": true,
47	"responsive_web_graphql_timeline_navigation_enabled": true,
48	"responsive_web_graphql_skip_user_profile_image_extensions_enabled": false,
49	"premium_content_api_read_enabled": false,
50	"withQuickPromoteEligibilityTweetFields": true
51	}""")
52
53	USER_MEDIA_VARS = _squash("""{
54	"userId": "%s", %s
55	"count": %s,
56	"includePromotedContent": false,
57	"withClientEventToken": false,
58	"withBirdwatchNotes": false,
59	"withVoice": true
60	}""")
61
62	# jack (id=12) - public account with media, same target as ratelimit_probe.py
63	TEST_VARS = USER_MEDIA_VARS % ("12", "", "20")
64
65	# --------------------------------------------------------------------- tid.nim
66
67	DEFAULT_KEYWORD = "obfiowerehiring"
68	PAIRS_URL = ("https://raw.githubusercontent.com/fa0311/"
69	"x-client-transaction-id-pair-dict/refs/heads/main/pair.json")
70	PAIRS_TTL_SEC = 60 * 60 # refresh every hour
71
72	_cached_pairs: list = []
73	_last_cached = 0
74
75
76	def _get_pair() -> dict:
77	global _cached_pairs, _last_cached
78	if not _cached_pairs or int(time.time()) - _last_cached > PAIRS_TTL_SEC:
79	resp = _http.get(PAIRS_URL, timeout=30)
80	if resp.status_code == 200:
81	pairs = resp.json()
82	if not pairs:
83	raise ValueError("Parsing pairs failed: " + resp.text)
84	_cached_pairs = pairs
85	_last_cached = int(time.time())
86
87	if not _cached_pairs:
88	raise ValueError("Failed to fetch x-client-transaction-id pairs")
89
90	return random.choice(_cached_pairs)
91
92
93	def _b64decode_nopad(data: str) -> bytes:
94	return base64.b64decode(data + "=" * (-len(data) % 4))
95
96
97	def gen_tid(path: str) -> str:
98	pair = _get_pair()
99
100	time_now = int(time.time() - 1682924400)
101	time_now_bytes = bytes([
102	time_now & 0xFF,
103	(time_now >> 8) & 0xFF,
104	(time_now >> 16) & 0xFF,
105	(time_now >> 24) & 0xFF,
106	])
107
108	data = "GET!" + path + "!" + str(time_now) + DEFAULT_KEYWORD + pair["animationKey"]
109	hash_bytes = hashlib.sha256(data.encode()).digest()
110	key_bytes = _b64decode_nopad(pair["verification"])
111	bytes_arr = key_bytes + time_now_bytes + hash_bytes[:16] + bytes([3])
112	random_num = random.randrange(256)
113	tid = bytes([random_num]) + bytes(b ^ random_num for b in bytes_arr)
114
115	return base64.b64encode(tid).decode().replace("=", "")
116
117
118	# ------------------------------------------------------- types / auth / session
119
120	class Session:
121	def __init__(self, kind, id, username, auth_token="", ct0="",
122	oauth_token="", oauth_secret=""):
123	self.kind = kind # "oauth" \| "cookie"
124	self.id = id
125	self.username = username
126	self.auth_token = auth_token
127	self.ct0 = ct0
128	self.oauth_token = oauth_token
129	self.oauth_secret = oauth_secret
130
131	def pretty(self) -> str:
132	if self.id > 0 and self.username:
133	result = f"{self.id} ({self.username})"
134	elif self.username:
135	result = self.username
136	elif self.id > 0:
137	result = str(self.id)
138	else:
139	result = "<unknown>"
140	return f"{self.kind} {result}"
141
142
143	def parse_session(raw: str) -> Session:
144	s = json.loads(raw)
145	kind = s.get("kind") or "oauth"
146	username = s.get("username", "") or ""
147
148	if kind == "oauth":
149	token = s.get("oauth_token", "") or ""
150	# mirrors Nim's token[0 ..< token.find('-')]
151	sid = token[:token.find("-")]
152	return Session("oauth", int(sid), username,
153	oauth_token=token,
154	oauth_secret=s.get("oauth_token_secret", "") or "")
155	if kind == "cookie":
156	raw_id = str(s.get("id", "") or "")
157	return Session("cookie", int(raw_id) if raw_id else 0, username,
158	auth_token=s.get("auth_token", "") or "",
159	ct0=s.get("ct0", "") or "")
160	raise ValueError("Unknown session kind: " + kind)
161
162
163	# ------------------------------------------------------------------ apiutils.nim
164
165	def build_url() -> str:
166	params = [("variables", TEST_VARS), ("features", GQL_FEATURES)]
167	prefix = "" if GRAPH_USER_MEDIA.startswith("1.1/") else "graphql/"
168	return ("https://x.com/i/api/" + prefix + GRAPH_USER_MEDIA
169	+ "?" + urlencode(params, quote_via=quote))
170
171
172	def gen_headers(session: Session, url: str, skip_tid: bool, disable_tid: bool) -> dict:
173	h = {
174	"Accept": "/",
175	"Accept-Encoding": "gzip",
176	"Accept-Language": "en-US,en;q=0.9",
177	"Content-Type": "application/json",
178	"Origin": "https://x.com",
179	"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
180	"(KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36",
181	"X-Twitter-Active-User": "yes",
182	"X-Twitter-Client-Language": "en",
183	"Priority": "u=1, i",
184	}
185
186	# only cookie sessions are tested, so the oauth branch is never reached
187	path = urlparse(url).path
188	h["X-Twitter-Auth-Type"] = "OAuth2Session"
189	h["X-Csrf-Token"] = session.ct0
190	h["Cookie"] = f"auth_token={session.auth_token}; ct0={session.ct0}"
191	h["Referer"] = "https://x.com/"
192	h["Sec-Ch-Ua"] = '"Google Chrome";v="142", "Chromium";v="142", "Not A(Brand";v="24"'
193	h["Sec-Ch-Ua-Mobile"] = "?0"
194	h["Sec-Ch-Ua-Platform"] = "Windows"
195	h["Sec-Fetch-Dest"] = "empty"
196	h["Sec-Fetch-Mode"] = "cors"
197	h["Sec-Fetch-Site"] = "same-origin"
198	if disable_tid or skip_tid or "/1.1/" in path:
199	h["Authorization"] = BEARER_TOKEN2
200	else:
201	h["Authorization"] = BEARER_TOKEN
202	h["X-Client-Transaction-Id"] = gen_tid(path)
203	return h
204
205
206	# ----------------------------------------------------------- test_sessions.nim
207
208	# no proxy, and ignore HTTP(S)_PROXY from the environment (Nim's client does too)
209	_http = requests.Session()
210	_http.trust_env = False
211
212
213	def parse_cookie_sessions(path: str) -> list:
214	result = []
215	skipped = 0
216	with open(path, encoding="utf-8") as f:
217	for line in f:
218	s = line.strip()
219	if not s:
220	continue
221	try:
222	sess = parse_session(s)
223	if sess.kind == "cookie":
224	result.append(sess)
225	else:
226	skipped += 1
227	except Exception as e:
228	print(" [!] Parse error:", e)
229	if skipped > 0:
230	print(" (skipped", skipped, "non-cookie sessions)")
231	return result
232
233
234	def test_session(session: Session, url: str):
235	"""Returns (ok, code, remaining, limit, reset, detail)."""
236	headers = gen_headers(session, url, skip_tid=False, disable_tid=False)
237
238	try:
239	resp = _http.get(url, headers=headers, timeout=30)
240	body = resp.text # requests transparently handles gzip
241
242	remaining = limit = reset = 0
243	if "x-rate-limit-reset" in resp.headers:
244	reset = int(resp.headers["x-rate-limit-reset"])
245	if "x-rate-limit-remaining" in resp.headers:
246	remaining = int(resp.headers["x-rate-limit-remaining"])
247	if "x-rate-limit-limit" in resp.headers:
248	limit = int(resp.headers["x-rate-limit-limit"])
249
250	if resp.status_code == 200:
251	return (True, resp.status_code, remaining, limit, reset, "")
252	detail = body if 1 <= len(body) <= 120 else ""
253	return (False, resp.status_code, remaining, limit, reset, detail)
254	except Exception as e:
255	return (False, 0, 0, 0, 0, str(e)[:120])
256
257
258	def main():
259	sessions_path = "sessions.jsonl"
260	delay = 500 # ms
261	args = sys.argv[1:]
262	i = 0
263
264	while i < len(args):
265	arg = args[i]
266	if arg == "--delay":
267	i += 1
268	if i >= len(args):
269	print("Error: --delay requires a value (seconds)")
270	sys.exit(1)
271	delay = int(float(args[i]) * 1000)
272	elif arg in ("--help", "-h"):
273	print("Usage: test_sessions.py [sessions_file] [--delay N]")
274	return
275	else:
276	sessions_path = arg
277	i += 1
278
279	if not os.path.isfile(sessions_path):
280	print("File not found:", sessions_path)
281	sys.exit(1)
282
283	sessions = parse_cookie_sessions(sessions_path)
284	if not sessions:
285	print("No cookie sessions found in", sessions_path)
286	sys.exit(0)
287
288	print("Testing", len(sessions), "sessions from", sessions_path)
289	print("")
290
291	url = build_url()
292	n_valid = n_fail = 0
293	failures = []
294
295	for idx, session in enumerate(sessions):
296	num = idx + 1
297	prefix = f"[{num}/{len(sessions)}] {session.pretty()}"
298	ok, code, remaining, limit, reset, detail = test_session(session, url)
299
300	if ok:
301	n_valid += 1
302	extra = f" remaining={remaining}/{limit}" if limit > 0 else ""
303	if reset > 0:
304	mins = max(0, reset - int(time.time())) // 60
305	extra += f" resets in {mins}m ({time.strftime('%H:%M:%S', time.localtime(reset))})"
306	print(prefix, " \u2713 ", code, extra, sep="")
307	else:
308	n_fail += 1
309	d = str(code) if code > 0 else "error: " + detail
310	failures.append(session.pretty() + " " + d)
311	print(prefix, " \u2717 ", d, sep="")
312
313	if num < len(sessions) and delay > 0:
314	time.sleep(delay / 1000)
315
316	print("")
317	print(f"Done: {n_valid} valid, {n_fail} invalid [{len(sessions)} total]")
318
319	if failures:
320	print("")
321	print("Failures:")
322	for f in failures:
323	print(" " + f)
324
325
326	if __name__ == "__main__":
327	main()