Technical Statement: The content of this article is for technical research and learning purposes only. Any data request behavior should comply with the target website's robots.txt protocol and terms of service.
Part 1: Protocol Reverse Engineering
1. Background
When developing the Sky Daily Mission push plugin, we encountered a tricky problem: the original solution relied on users manually filling in Weibo cookies (SUB and XSRF-TOKEN), which brought two pain points:
- Poor User Experience: Ordinary users don’t know how to extract cookies from the browser
- High Maintenance Cost: Cookies have a short validity period and need frequent updates
Directly requesting the Weibo API (e.g., /api/container/getIndex) returns a 432 status code, indicating that even in "visitor" state, the server has a strict session management mechanism.
Technical Goal: Reverse engineer the Weibo H5 visitor authentication protocol to automatically obtain temporary visitor credentials, so the plugin can access Weibo data without requiring users to provide cookies.
2. Protocol Analysis: Three-Phase Authentication Chain
Through packet analysis using Edge DevTools, comparing before and after clearing cookies, we confirmed that the Weibo H5 visitor authentication is a cross-domain three-phase closed loop.
2.1 Authentication Architecture Diagram
2.2 Phase One: Identity Token Issuance
Endpoint: POST https://visitor.passport.weibo.cn/visitor/genvisitor2
Key Parameters:
data = {
'cb': 'visitor_gray_callback',
'tid': '',
'new_tid': 'null'
}
Technical Difficulties:
- JSONP Response Parsing: The response format is
visitor_gray_callback({...})instead of standard JSON, requiring regex extraction - Manual Cookie Injection: The server does not deliver SUB via Set-Cookie; it must be extracted from the response payload and manually written into the CookieJar
Core Code:
resp = session.post(url, headers=headers, data=data)
match = re.search(r'visitor_gray_callback\((.*)\)', resp.text)
json_data = json.loads(match.group(1))
sub = json_data['data']['sub']
session.cookies.set('SUB', sub, domain='.weibo.cn')
2.3 Phase Two: Session Initialization
Endpoint: GET https://m.weibo.cn/
Function: Carry SUB to access the main site, activate the session, and obtain XSRF-TOKEN
Key Points:
- Must carry the SUB cookie obtained from Phase One
- The server delivers XSRF-TOKEN via the Set-Cookie response header
- XSRF-TOKEN is a session-level short-lived token
2.4 Phase Three: Protected Resource Access
Security Mechanism: Double Submit Cookie pattern to defend against CSRF
Client Requirements:
- Cookie level: Carry SUB and XSRF-TOKEN
- Header level: Write the value of XSRF-TOKEN into the
x-xsrf-tokenrequest header
Server-Side Validation:
Header['x-xsrf-token'] == Cookie['XSRF-TOKEN']
3. PoC Implementation
Complete proof-of-concept code:
# -*- coding: utf-8 -*-
import requests
import json
import re
class WeiboH5VisitorAuth:
def __init__(self):
self.session = requests.Session()
self._init_headers()
self.sub = None
self.xsrf_token = None
def _init_headers(self):
"""Configure browser fingerprint"""
headers_base = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'sec-ch-ua': '"Chromium";v="142", "Microsoft Edge";v="142"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'Accept-Encoding': 'gzip, deflate, br, zstd',
'dnt': '1',
}
self.session.headers.update(headers_base)
def step1_obtain_identity_token(self):
"""Phase 1: Obtain SUB"""
url = 'https://visitor.passport.weibo.cn/visitor/genvisitor2'
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'Origin': 'https://visitor.passport.weibo.cn',
}
data = {'cb': 'visitor_gray_callback', 'tid': '', 'new_tid': 'null'}
resp = self.session.post(url, headers=headers, data=data)
match = re.search(r'visitor_gray_callback\((.*)\)', resp.text)
json_data = json.loads(match.group(1))
if json_data.get('retcode') == 20000000:
self.sub = json_data['data']['sub']
self.session.cookies.set('SUB', self.sub, domain='.weibo.cn')
self.session.cookies.set('SUBP', json_data['data']['subp'], domain='.weibo.cn')
def step2_initialize_session(self):
"""Phase 2: Obtain XSRF-TOKEN"""
url = 'https://m.weibo.cn'
headers = {'Referer': 'https://visitor.passport.weibo.cn/'}
self.session.get(url, headers=headers)
self.xsrf_token = self.session.cookies.get('XSRF-TOKEN')
def step3_access_api(self, container_id):
"""Phase 3: Access API"""
api_url = 'https://m.weibo.cn/api/container/getIndex'
params = {'containerid': container_id, 'page': 1, 'count': 10}
headers = {
'Accept': 'application/json, text/plain, */*',
'x-xsrf-token': self.xsrf_token, # Double Submit
'Referer': f'https://m.weibo.cn/u/{container_id}'
}
resp = self.session.get(api_url, params=params, headers=headers)
return resp.json()
def run(self, target_id):
self.step1_obtain_identity_token()
self.step2_initialize_session()
return self.step3_access_api(target_id)
Part 2: Engineering Practice - Integration into AstrBot Plugin
4. Practical Challenges
When integrating the PoC into the production environment, the following challenges were encountered:
- API Differences: Different response formats between the mobile API and the PC API
- Performance Optimization: Avoid repeated authentication and reuse sessions
- Error Handling: Fallback mechanism for network anomalies and authentication failure
- Data Adaptation: HTML tag cleaning and newline preservation
5. Architecture Design
5.1 Dual-Strategy Architecture
An intelligent fallback mechanism was designed:
class Auth:
def __init__(self, config):
self.use_cookie = config.get("cookies", {}).get("enabled", False)
self._visitor_cookies = {}
async def init_visitor_auth(self, session):
"""Cookie-free strategy: Initialize visitor authentication"""
# Implement three-phase authentication
pass
Workflow:
- Prefer using the user-configured cookies (PC API)
- If cookies are not configured or expire, automatically switch to the cookie-free strategy (mobile API)
5.2 Spider Class Refactoring
class Spider:
async def fetch(self, page=0):
# First try the cookie strategy
if self.auth.use_cookie and self.auth._get_cookie():
try:
return await self._fetch_with_cookie(page)
except Exception as e:
logger.warning(f"Cookie strategy failed: {e}, switching to cookie-free strategy")
# Fallback to cookie-free strategy
return await self._fetch_without_cookie(page)
6. Key Technical Points
6.1 API Response Difference Adaptation
PC API (/ajax/statuses/mymblog):
{
"ok": 1,
"data": {
"list": [{
"mblogid": "xxx",
"text_raw": "plain text",
"pic_infos": {...}
}]
}
}
Mobile API (/api/container/getIndex):
{
"ok": 1,
"data": {
"cards": [{
"card_group": [{
"mblog": {
"mid": "xxx",
"text": "<br />HTML text<br />",
"pics": [...]
}
}]
}]
}
}
Adaptation Scheme:
def _parse_mobile_mblogs(self, data):
for card in data.get("cards", []):
for item in card.get("card_group", []):
mblog = item["mblog"]
# 1. Convert <br> to newline
text_raw = re.sub(r'<br\s*/?>', '\n', mblog.get("text", ""))
# 2. Remove other HTML tags
text_raw = re.sub(r'<[^>]+>', '', text_raw).strip()
self._results.append(Blog(
mblogid=mblog.get("mid"),
text_raw=text_raw,
is_long_text=mblog.get("isLongText", False),
use_mobile_api=True
))
6.2 Long Text Optimization
The text returned by the mobile API may be truncated, requiring the long text API:
async def fetch_long_text(self, client, auth):
if not self.is_long_text:
return self.text_raw
if self.use_mobile_api:
api = "https://m.weibo.cn/statuses/extend"
else:
api = "https://weibo.com/ajax/statuses/longtext"
response = await client.get(api, params={"id": self.mblogid})
long_text = response.json()["data"]["longTextContent"]
# Clean HTML and preserve newlines
long_text = re.sub(r'<br\s*/?>', '\n', long_text)
return re.sub(r'<[^>]+>', '', long_text).strip()
6.3 Performance Optimization: Session Reuse
Problem: Each data source requires visitor authentication, resulting in too many requests
Before Optimization (2 data sources):
- Data Source 1: Auth 1 time + List 1 time + Long text auth 1 time + Long text 1 time = 4 times
- Data Source 2: Auth 1 time + List 1 time + Long text auth 1 time + Long text 1 time = 4 times
- Total: 8 requests, 4 authentications
Optimization Scheme:
class Spider:
def __init__(self):
self._client = None # Save client reference
async def _fetch_without_cookie(self, page):
self._client = httpx.AsyncClient()
await self.auth.init_visitor_auth(self._client)
# Get list...
return self # client remains open
# In SkyDaily.get_daily_data
try:
await spider.fetch()
blog = spider.filter_by_regex(pattern).one()
# Reuse spider's client to get long text
if spider._client:
text = await blog.fetch_long_text(spider._client, auth=auth)
finally:
# Ensure client is closed
if spider._client:
await spider._client.aclose()
After Optimization (2 data sources):
- Data Source 1: Auth 1 time + List 1 time + Long text 1 time (reuse client) = 3 times
- Data Source 2: Auth 1 time + List 1 time + Long text 1 time (reuse client) = 3 times
- Total: 6 requests, 2 authentications
6.4 Regular Expression Adaptation
Problem: After cleaning HTML from the mobile API, the super topic tag format changes, causing regex matching failure
Original HTML:
<span class="surl-text">Sky Light Encounter Super Topic</span> 11.22 | Daily Mission
After cleaning:
Sky Light Encounter Super Topic 11.22 | Daily Mission
Solution: Modify the regex from relying on specific formats to keyword matching
# Old regex (depends on # tag)
pattern = r"^#[^#]*Light[^#]*Super Topic]#\s*\d{1,2}\.\d{1,2}\s*"
# New regex (keyword matching)
pattern = r".*Sky Light.*Daily Mission.*"
7. Configuration Design
{
"cookies": {
"enabled": false,
"sub": "",
"xsrf_token": ""
},
"data_sources": [
"7360748659:.*Sky Light.*Daily Mission.*:Did the wanderer tip over today",
"5539106873:^【National Server·Daily Mission Guide】:Chen Chen works hard never giving up"
]
}
Working Logic:
cookies.enabled = trueand cookies are filled → use PC APIcookies.enabled = falseor cookies are empty → automatically use cookie-free strategy
8. Error Handling
async def fetch(self, page=0, max_attempts=3, retry_delay=2):
for attempt in range(max_attempts):
try:
# Attempt to fetch data
pass
except (httpx.HTTPStatusError, httpx.TimeoutException) as e:
if attempt < max_attempts - 1:
logger.warning(f"Request failed, retrying in {retry_delay} seconds...")
await asyncio.sleep(retry_delay)
continue
else:
raise
finally:
# Ensure resource cleanup
if self._client:
await self._client.aclose()
9. Practical Results
Before Optimization:
- Users need to manually fill in cookies
- Plugin fails when cookies expire
- Only supports PC API
After Optimization:
- Zero configuration to use
- Automatically obtains temporary visitor credentials
- Dual-strategy automatic switching
- 25% performance improvement (reduced 2 authentication requests)
Log Example:
[INFO] Using cookie-free strategy to fetch Weibo data
[INFO] Visitor authentication initialized successfully
[INFO] Retrieved 1 Weibo post from "Did the wanderer tip over today"
[INFO] Today's national server guide query completed, successfully obtained 2/2 data sources
Summary
Key Points
- JSONP Response Handling: Use regex to extract JSON, manually inject cookies
- Double Submit CSRF: Read token from cookie and write into header
- Session Reuse: Avoid repeated authentication, reduce network requests
- HTML Cleaning: Preserve newlines, improve text readability
- Error Handling: Implement retry mechanism and graceful degradation
Applicable Scenarios
This solution is suitable for:
- Automated tools that need to access Weibo public data
- Applications that do not want users to manually provide cookies
- Crawler services that need long-term stable operation
Notes
- Comply with Agreements: Strictly follow robots.txt and terms of service
- Frequency Control: Avoid high-frequency requests that trigger risk control
- User-Agent: Use a real browser fingerprint
- Error Handling: Implement proper exception handling and degradation strategies
References
- Weibo Mobile API Documentation
- OWASP CSRF Defense
- AstrBot Plugin Development Documentation
- nonebot2's Sky Daily Mission and Event Query Plugin
Project Address: GitHub - Sky Daily Plugin
The complete code for this article is open source. Stars and PRs are welcome. If you have any questions, please raise an Issue for discussion.