fix: scan resource page JS for GQL hash (Spotify lazy-loads feature chunks)
fetchPlaylist/getAlbum/getTrack GraphQL hashes are not in homepage bundles — they live in lazy-loaded chunks that only appear on the playlist/album/track page. Fetch the specific page HTML to find the extra JS bundles, then scan those for the persisted query hash. Also improved hash regex (no semicolon restriction, wider window, fallback nearest-hex search). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
1 parent
e87d48d6b0
commit
46eb8c826f
1 file changed
+63
-26
+63
-26
@@ -22,49 +22,86 @@ def _totp(secret_str, server_ts):
|
|||||||
(digest[offset+2] & 0xff) << 8 | (digest[offset+3] & 0xff)) % 1_000_000
|
(digest[offset+2] & 0xff) << 8 | (digest[offset+3] & 0xff)) % 1_000_000
|
||||||
return str(code).zfill(6)
|
return str(code).zfill(6)
|
||||||
|
|
||||||
def get_token_and_hashes(session):
|
def _collect_js_urls(html, seen):
|
||||||
"""Fetch TOTP token + extract GraphQL persisted-query hashes from JS bundles."""
|
urls = []
|
||||||
html = session.get('https://open.spotify.com/', headers={
|
|
||||||
'Accept': 'text/html,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.9',
|
|
||||||
'Accept-Encoding': 'identity',
|
|
||||||
}, timeout=20).text
|
|
||||||
|
|
||||||
js_urls, seen = [], set()
|
|
||||||
for m in re.finditer(r'src="((?:https?://[^"]*|/[^"]*?)\.js(?:\?[^"]*)?)"', html):
|
for m in re.finditer(r'src="((?:https?://[^"]*|/[^"]*?)\.js(?:\?[^"]*)?)"', html):
|
||||||
u = m.group(1)
|
u = m.group(1)
|
||||||
if not u.startswith('http'): u = 'https://open.spotify.com' + u
|
if not u.startswith('http'): u = 'https://open.spotify.com' + u
|
||||||
if u not in seen: seen.add(u); js_urls.append(u)
|
if u not in seen: seen.add(u); urls.append(u)
|
||||||
|
return urls
|
||||||
|
|
||||||
|
def _find_gql_hash(js, op):
|
||||||
|
"""Try multiple patterns to find the persisted-query hash for an operation."""
|
||||||
|
for p in [
|
||||||
|
rf'operationName:"{op}"[^{{}}]{{0,1000}}sha256Hash:"([a-f0-9]{{64}})"',
|
||||||
|
rf'sha256Hash:"([a-f0-9]{{64}})"[^{{}}]{{0,500}}operationName:"{op}"',
|
||||||
|
rf'"{op}"[^;]{{0,2000}}sha256Hash:"([a-f0-9]{{64}})"',
|
||||||
|
]:
|
||||||
|
m = re.search(p, js)
|
||||||
|
if m: return m.group(1)
|
||||||
|
# Fallback: any 64-hex string within 3000 chars of the op name
|
||||||
|
pos = js.find(f'"{op}"')
|
||||||
|
if pos >= 0:
|
||||||
|
window = js[max(0, pos - 3000):pos + 3000]
|
||||||
|
m = re.search(r'sha256Hash:"([a-f0-9]{64})"', window)
|
||||||
|
if m: return m.group(1)
|
||||||
|
return None
|
||||||
|
|
||||||
|
def _scan_bundles(session, js_urls, need_totp, need_ops):
|
||||||
|
"""Scan JS bundles for TOTP secret and/or GQL hashes."""
|
||||||
totp_secret = totp_version = None
|
totp_secret = totp_version = None
|
||||||
gql_hashes = {} # operationName -> sha256Hash
|
gql_hashes = {}
|
||||||
GQL_OPS = ['fetchPlaylist', 'getAlbum', 'getTrack']
|
for u in js_urls:
|
||||||
|
if (not need_totp or totp_secret) and all(op in gql_hashes for op in need_ops):
|
||||||
for u in js_urls[:20]:
|
|
||||||
if totp_secret and all(op in gql_hashes for op in GQL_OPS):
|
|
||||||
break
|
break
|
||||||
try:
|
try:
|
||||||
js = session.get(u, headers={'Accept': '*/*', 'Accept-Encoding': 'identity'}, timeout=8).text
|
js = session.get(u, headers={'Accept': '*/*', 'Accept-Encoding': 'identity'}, timeout=8).text
|
||||||
# TOTP secret
|
if need_totp and not totp_secret:
|
||||||
if not totp_secret:
|
|
||||||
m = re.search(r'\{secret:\s*["\']([^"\']+)["\']\s*,\s*version:\s*(\d+)\s*\}', js)
|
m = re.search(r'\{secret:\s*["\']([^"\']+)["\']\s*,\s*version:\s*(\d+)\s*\}', js)
|
||||||
if m:
|
if m:
|
||||||
totp_secret, totp_version = m.group(1), int(m.group(2))
|
totp_secret, totp_version = m.group(1), int(m.group(2))
|
||||||
eprint(f'[spotify] TOTP secret v{totp_version} found')
|
eprint(f'[spotify] TOTP secret v{totp_version} found')
|
||||||
# GraphQL hashes
|
for op in need_ops:
|
||||||
for op in GQL_OPS:
|
|
||||||
if op in gql_hashes: continue
|
if op in gql_hashes: continue
|
||||||
m = re.search(rf'operationName:"{op}"[^;]{{0,600}}sha256Hash:"([a-f0-9]{{64}})"', js)
|
h = _find_gql_hash(js, op)
|
||||||
if not m:
|
if h:
|
||||||
m = re.search(rf'sha256Hash:"([a-f0-9]{{64}})"[^;]{{0,300}}operationName:"{op}"', js)
|
gql_hashes[op] = h
|
||||||
if m:
|
eprint(f'[spotify] GQL hash {op}: {h[:16]}...')
|
||||||
gql_hashes[op] = m.group(1)
|
|
||||||
eprint(f'[spotify] GQL hash {op}: {m.group(1)[:16]}...')
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
eprint(f'[spotify] bundle error: {e}')
|
eprint(f'[spotify] bundle error ({u[-40:]}): {e}')
|
||||||
|
return totp_secret, totp_version, gql_hashes
|
||||||
|
|
||||||
|
def get_token_and_hashes(session, sp_type, sp_id):
|
||||||
|
"""Fetch TOTP token + extract GQL hashes.
|
||||||
|
Scans homepage bundles for TOTP, then the specific resource page for GQL hashes
|
||||||
|
(Spotify lazy-loads feature chunks only on the relevant page)."""
|
||||||
|
page_headers = {
|
||||||
|
'Accept': 'text/html,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.9',
|
||||||
|
'Accept-Encoding': 'identity',
|
||||||
|
}
|
||||||
|
GQL_OP = {'track': 'getTrack', 'album': 'getAlbum', 'playlist': 'fetchPlaylist'}[sp_type]
|
||||||
|
|
||||||
|
# Step 1: homepage — find TOTP secret
|
||||||
|
seen = set()
|
||||||
|
home_html = session.get('https://open.spotify.com/', headers=page_headers, timeout=20).text
|
||||||
|
home_js = _collect_js_urls(home_html, seen)
|
||||||
|
totp_secret, totp_version, gql_hashes = _scan_bundles(session, home_js[:20], True, [GQL_OP])
|
||||||
|
|
||||||
if not totp_secret:
|
if not totp_secret:
|
||||||
raise Exception('TOTP secret not found')
|
raise Exception('TOTP secret not found')
|
||||||
|
|
||||||
|
# Step 2: resource page — find GQL hash (lazy-loaded playlist/album/track chunk)
|
||||||
|
if GQL_OP not in gql_hashes:
|
||||||
|
eprint(f'[spotify] fetching {sp_type} page for GQL hash...')
|
||||||
|
try:
|
||||||
|
page_url = f'https://open.spotify.com/{sp_type}/{sp_id}'
|
||||||
|
page_html = session.get(page_url, headers=page_headers, timeout=15).text
|
||||||
|
extra_js = _collect_js_urls(page_html, seen)
|
||||||
|
_, _, extra_hashes = _scan_bundles(session, extra_js[:25], False, [GQL_OP])
|
||||||
|
gql_hashes.update(extra_hashes)
|
||||||
|
except Exception as e:
|
||||||
|
eprint(f'[spotify] page scan error: {e}')
|
||||||
|
|
||||||
st = session.get('https://open.spotify.com/api/server-time', headers={
|
st = session.get('https://open.spotify.com/api/server-time', headers={
|
||||||
'Accept': 'application/json', 'Referer': 'https://open.spotify.com/',
|
'Accept': 'application/json', 'Referer': 'https://open.spotify.com/',
|
||||||
}, timeout=10).json()
|
}, timeout=10).json()
|
||||||
@@ -295,7 +332,7 @@ def main():
|
|||||||
|
|
||||||
from curl_cffi import requests as cf
|
from curl_cffi import requests as cf
|
||||||
session = cf.Session(impersonate='chrome124')
|
session = cf.Session(impersonate='chrome124')
|
||||||
token, gql_hashes = get_token_and_hashes(session)
|
token, gql_hashes = get_token_and_hashes(session, sp_type, sp_id)
|
||||||
|
|
||||||
GQL_OP = {'track': 'getTrack', 'album': 'getAlbum', 'playlist': 'fetchPlaylist'}[sp_type]
|
GQL_OP = {'track': 'getTrack', 'album': 'getAlbum', 'playlist': 'fetchPlaylist'}[sp_type]
|
||||||
gql_hash = gql_hashes.get(GQL_OP)
|
gql_hash = gql_hashes.get(GQL_OP)
|
||||||
|
|||||||
Reference in new issue
Block a user