官方口径已确认完整，关键事实清晰。现在撰写正文和源码，把整条链路串起来。
🔑《从0到1：闲鱼开放平台授权登录 + AccessToken 刷新 + 聚石塔部署完整链路》（附Python源码）
结论先拍：闲鱼开放平台没有独立的授权体系——它寄生在淘宝开放平台（TOP）OAuth 2.0 上，走的是标准的授权码模式：
authorize 拿 code → oauth.taobao.com/token 换 SessionKey（即 access_token）→ 后续调用带 session=SessionKey。 整条链路有四个"死就白干"的卡点：① code 只能用一次且 5 分钟过期 ② SessionKey 必须按 shop_id 隔离持久化（它是"哪个淘宝账号授权就只能查哪个商家数据"）③ 刷新机制分应用类型——订购类服务商可用 taobao.top.auth.token.refresh 续 r2，自研/固定时长应用只能重新登录授权 ④ 订单/会员数据 + 高风险写 API 强制聚石塔部署，塔外调用直接失败。 下面从授权 URL 一路写到生产级 TokenManager + 聚石塔架构准入。一、完整链路全景图
┌────────────────────────────────────────────────────────────────┐
│ Step 1 引导授权 │
│ 拼接 https://oauth.taobao.com/authorize?response_type=code │
│ &client_id=APPKEY &redirect_uri=CALLBACK &state=xxx │
│ ↓ 用户登录淘宝/闲鱼账号 → 点击授权 │
├────────────────────────────────────────────────────────────────┤
│ Step 2 回调 + code 换 SessionKey │
│ POST https://oauth.taobao.com/token │
│ grant_type=authorization_code, code, client_id, │
│ client_secret, redirect_uri │
│ ← { access_token(SessionKey), expires_in, refresh_token, │
│ re_expires_in, taobao_user_id, taobao_user_nick } │
├────────────────────────────────────────────────────────────────┤
│ Step 3 持久化 + 调用 API │
│ 每次 TOP 调用: session = SessionKey │
│ 按 shop_id / taobao_user_id 维度缓存 Token │
├────────────────────────────────────────────────────────────────┤
│ Step 4 刷新（分应用类型） │
│ · 订购类服务商: taobao.top.auth.token.refresh (续 r2) │
│ · 自研/固定时长: 重新登录授权 (force_auth=true) │
├────────────────────────────────────────────────────────────────┤
│ Step 5 聚石塔部署（强制） │
│ · 订单/会员数据存塔内 RDS │
│ · 高风险写 API 从塔内发起 │
│ · 塔内→塔外 须走奇门标准接口审批 │
└────────────────────────────────────────────────────────────────┘关键认知：access_token就是 SessionKey，两套叫法同一个东西。淘宝官方文档原文："平台会返回用户授权码 SessionKey（即 access_token）"。
二、授权登录：授权码模式（Step 1 + 2）
2.1 拼接授权 URL（引导用户登录）
https://oauth.taobao.com/authorize? response_type=code &client_id=你的AppKey &redirect_uri=你的回调地址 &state=随机值(防CSRF) &view=web
闲鱼合作方有两种拿 code 的方式：
传统 Web：跳转
oauth.taobao.0.com/authorize，授权后跳回调页带?code=xxx&state=xxx小程序新接入：前端组件直接获取 code（如闲鱼小程序客户端 API），不用跳回调
state必须原样带回，防 CSRF；redirect_uri必须与控制台配置完全一致
2.2 code 换 SessionKey
POST https://oauth.taobao.com/token Content-Type: application/x-www-form-urlencoded grant_type=authorization_code &code=上一步的code &client_id=AppKey &client_secret=AppSecret &redirect_uri=回调地址
成功响应：
{
"access_token": "6101_UMIdX...", // ← 这就是 SessionKey
"token_type": "Bearer",
"expires_in": 86400, // access_token 有效期(秒)
"refresh_token": "xxx",
"re_expires_in": 7200, // refresh_token 有效期
"taobao_user_id": "12345678",
"taobao_user_nick": "卖家昵称"
}⚠️ code 一次性 + 5 分钟过期：被用过或过期必须重新获取；AppSecret绝不能出现在 URL 里，只能放 POST body。
三、Token 刷新：分应用类型的坑（Step 4）
这是最容易踩的雷，官方说得很清楚：
应用类型 | 能否用 refresh_token 续期 | 正确做法 |
|---|---|---|
订购类服务商（发布到服务市场售卖） | ✅ taobao.top.auth.token.refresh 续 r2 时长 | 续期 r2，r1/w1 同订购时长一般不用续 |
自研 / 固定时长（商家后台、新业务） | ❌ 不支持 refresh 延长总时长 | 重新登录授权，或 force_auth=true 刷新有效时间 |
核心规则：
refresh_token只续access_token的 r2 时长，不是总授权时长refresh_expires_in = 0→ 无法刷新，必须重新授权建议过期前（约 30 分钟~1 周，视应用时效档位）主动刷新，不要等 401
官方时效档位：测试环境 24h / 上线环境 code 30 分钟、access_token 30 天、refresh_token 60~180 天
四、聚石塔部署强制规则（Step 5）
只要你的应用标签属于 ERP/进销存、订单管理、WMS、CRM、商品管理、电商财务等 23 类之一，必须入驻聚石塔。关键准入规则：
系统部署：ECS + RDS，应用跑 ECS，数据库只放 RDS（DB 不准装 ECS 上）
数据存储：订单 + 会员数据必须存塔内 RDS，禁止通过自有接口二次开放到塔外
API 调用：所有 TOP 调用必须从塔内发起；高风险写 API（改价、优惠）必须从塔内发起
塔内→塔外：默认禁止，确有需要走奇门标准接口体系 + 平台审批
架构：三层架构，业务逻辑服务端 ECS、数据层 RDS、客户端只做展现
应用隔离：不同应用必须不同 AppKey + 不同 ECS，御城河安全边界隔离
闲鱼合作方官方文档明确："由于集团安全规则，需接入聚石塔后进行调用，用户 id、订单信息等必须存储在聚石塔内的数据库中"。
五、Python：完整生产级实现
# xianyu_auth_platform.py
"""
闲鱼开放平台(淘宝TOP OAuth2.0) 完整链路
- Step1: 授权URL生成 + state(CSRF)校验
- Step2: code 换 SessionKey (access_token)
- Step3: TokenManager (按shop_id持久化 + 自动刷新 + 提前续期)
- Step4: TOP API 调用封装 (自动注入session + 签名)
- Step5: 聚石塔部署合规性自检
"""
import time, hashlib, json, threading, secrets
from typing import Dict, Optional, Any
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
from enum import Enum
import urllib.parse
# ==================== 配置 ====================
@dataclass
class XianyuConfig:
app_key: str = "你的AppKey"
app_secret: str = "你的AppSecret"
redirect_uri: str = "https://yourdomain.com/xianyu/callback"
# 沙箱用 gw.api.tbsandbox.com；生产用 gw.api.taobao.com
api_gateway: str = "https://gw.api.taobao.com/router/rest"
auth_url: str = "https://oauth.taobao.com/authorize"
token_url: str = "https://oauth.taobao.com/token"
is_sandbox: bool = False
# ==================== 应用类型（决定刷新策略）====================
class AppType(Enum):
SELF_USE = "self_use" # 自研/固定时长 → 只能重新授权
PURCHASE_SERVICE = "purchase" # 订购类服务商 → 可refresh续r2
# ==================== Token 数据 ====================
@dataclass
class TokenSet:
shop_id: str
taobao_user_id: str
access_token: str # = SessionKey
refresh_token: str
expires_in: int = 86400
re_expires_in: int = 0 # refresh_token有效期, 0=不可刷新
created_at: float = field(default_factory=time.time)
app_type: AppType = AppType.SELF_USE
@property
def expires_at(self) -> float:
return self.created_at + self.expires_in
def needs_refresh(self, ahead_sec: int = 1800) -> bool:
"""提前 ahead_sec 秒触发刷新"""
return time.time() >= (self.expires_at - ahead_sec)
def can_refresh(self) -> bool:
"""refresh_token 是否仍有效"""
if self.re_expires_in <= 0:
return False
return (self.created_at + self.re_expires_in) > time.time()
# ==================== Step1 + Step2: 授权客户端 ====================
class XianyuAuthClient:
"""授权登录全流程"""
def __init__(self, config: XianyuConfig):
self.cfg = config
def build_authorize_url(self, state: Optional[str] = None,
force_auth: bool = False) -> tuple[str, str]:
"""Step1: 生成授权URL + 返回state供校验"""
state = state or secrets.token_urlsafe(16)
params = {
"response_type": "code",
"client_id": self.cfg.app_key,
"redirect_uri": self.cfg.redirect_uri,
"state": state,
"view": "web",
}
if force_auth:
params["force_auth"] = "true" # 强制重新授权(刷新有效时间)
url = f"{self.cfg.auth_url}?{urllib.parse.urlencode(params)}"
return url, state
def verify_state(self, returned: str, expected: str) -> bool:
"""防CSRF: state必须原样匹配"""
return returned == expected and bool(expected)
def exchange_code(self, code: str) -> TokenSet:
"""Step2: code 换 SessionKey"""
if not code:
raise ValueError("code 不能为空(code只能用一次且5分钟过期)")
# 生产: requests.post(self.cfg.token_url, data=payload)
# 这里演示返回结构
payload = {
"grant_type": "authorization_code",
"code": code,
"client_id": self.cfg.app_key,
"client_secret": self.cfg.app_secret, # ★ body, 绝不放URL
"redirect_uri": self.cfg.redirect_uri,
}
# mock response
resp = self._mock_token_response(code)
return TokenSet(
shop_id=resp["taobao_user_id"],
taobao_user_id=resp["taobao_user_id"],
access_token=resp["access_token"],
refresh_token=resp.get("refresh_token", ""),
expires_in=resp.get("expires_in", 86400),
re_expires_in=resp.get("re_expires_in", 0),
)
def _mock_token_response(self, code: str) -> Dict:
return {
"access_token": f"6101_{code[:10]}",
"refresh_token": f"refresh_{code[:6]}",
"expires_in": 86400,
"re_expires_in": 7776000, # 90天
"taobao_user_id": "12345678",
"taobao_user_nick": "seller_nick",
}
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== Step3: Token 管理器 ====================
class TokenStore(ABC):
"""持久化接口 (DB/Redis实现替换)"""
@abstractmethod
def load(self, shop_id: str) -> Optional[TokenSet]: ...
@abstractmethod
def save(self, token: TokenSet): ...
class MemoryTokenStore(TokenStore):
def __init__(self): self._d: Dict[str, TokenSet] = {}
def load(self, shop_id): return self._d.get(shop_id)
def save(self, token): self._d[token.shop_id] = token
class TokenManager:
"""按shop_id管理Token + 自动刷新"""
def __init__(self, auth: XianyuAuthClient, store: TokenStore,
app_type: AppType = AppType.SELF_USE,
refresh_ahead_sec: int = 1800):
self.auth = auth
self.store = store
self.app_type = app_type
self.ahead = refresh_ahead_sec
self._lock = threading.Lock()
def get(self, shop_id: str) -> Optional[TokenSet]:
"""获取有效Token, 过期则自动刷新"""
token = self.store.load(shop_id)
if token is None:
return None
token.app_type = self.app_type
if not token.needs_refresh(self.ahead):
return token
# 需要刷新
return self._refresh(token)
def _refresh(self, token: TokenSet) -> Optional[TokenSet]:
with self._lock:
# 双检
fresh = self.store.load(token.shop_id)
if fresh and not fresh.needs_refresh(self.ahead):
return fresh
if self.app_type == AppType.PURCHASE_SERVICE and token.can_refresh():
# 订购类: 调 taobao.top.auth.token.refresh 续 r2
new_set = self._call_refresh_api(token.refresh_token)
new_set.shop_id = token.shop_id
self.store.save(new_set)
return new_set
else:
# 自研/固定时长: refresh无法延长总时长 → 需重新授权
raise TokenRefreshRequired(
f"shop={token.shop_id} 需重新授权(force_auth=true重新登录)")
def _call_refresh_api(self, refresh_token: str) -> TokenSet:
"""taobao.top.auth.token.refresh"""
# 生产: 调用 TOP 接口; 这里演示
# grant_type=refresh_token, refresh_token=xxx, client_id, client_secret
return TokenSet(
shop_id="", taobao_user_id="", access_token=f"new_{int(time.time())}",
refresh_token=refresh_token, # ★ 换新access, refresh通常不变或也换
expires_in=86400, re_expires_in=7776000,
app_type=AppType.PURCHASE_SERVICE,
)
def on_callback(self, shop_id: str, code: str, state: str,
expected_state: str) -> TokenSet:
"""授权回调入口"""
if not self.auth.verify_state(state, expected_state):
raise ValueError("state校验失败(CSRF风险)")
token = self.auth.exchange_code(code)
token.shop_id = shop_id
token.app_type = self.app_type
self.store.save(token)
return token
class TokenRefreshRequired(Exception): pass
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== Step3续: TOP API 调用封装 ====================
class TopApiClient:
"""自动注入session(SessionKey) + MD5签名"""
def __init__(self, config: XianyuConfig, token_mgr: TokenManager):
self.cfg = config
self.tokens = token_mgr
def call(self, method: str, shop_id: str, params: Dict) -> Dict:
token = self.tokens.get(shop_id)
if token is None:
raise TokenRefreshRequired(f"shop={shop_id} 未授权")
# 组装签名
all_params = {
"method": method,
"app_key": self.cfg.app_key,
"session": token.access_token, # ★ SessionKey = access_token
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"format": "json", "v": "2.0",
**{k: v for k, v in params.items() if v is not None},
}
all_params["sign"] = self._sign(all_params)
# 生产: requests.post(gateway, data=all_params)
return {"_mock": True, "url": self.cfg.api_gateway, "method": method,
"session": token.access_token[:8] + "..."}
def _sign(self, params: Dict) -> str:
s = self.cfg.app_secret + "".join(
f"{k}{params[k]}" for k in sorted(params) if params[k] is not None) + self.cfg.app_secret
return hashlib.md5(s.encode()).hexdigest().upper()
# ==================== Step5: 聚石塔合规自检 ====================
@dataclass
class JushitaCheck:
app_tags: list # 应用标签, 如 ["ERP", "订单管理"]
data_in_rds: bool # 订单/会员是否存塔内RDS
api_from_inside: bool # TOP调用是否塔内发起
write_from_inside: bool # 高风险写是否塔内
cross_tower_via_qimen: bool # 塔外交互是否走奇门
three_tier: bool # 是否三层架构
yc_isolation: bool # 御城河隔离
FORCE_TAGS = {"ERP/进销存软件","订单管理","仓储管理系统","客户关系管理",
"商品管理","电商财务","全渠道ERP","行业/店铺分析","服务商后台系统"}
def evaluate(self) -> Dict:
must = any(t in self.FORCE_TAGS for t in self.app_tags)
issues = []
if must:
if not self.data_in_rds: issues.append("订单/会员数据必须存塔内RDS")
if not self.api_from_inside: issues.append("TOP API调用必须从塔内发起")
if not self.write_from_inside: issues.append("高风险写API必须从塔内发起")
if not self.three_tier: issues.append("须三层架构(客户端只展现)")
if not self.yc_isolation: issues.append("须配置御城河安全边界隔离")
if not self.cross_tower_via_qimen: issues.append("塔外交互须走奇门标准接口+审批")
return {
"must_jushita": must,
"pass": len(issues) == 0,
"issues": issues,
}
# ==================== 演示 ====================
if __name__ == "__main__":
cfg = XianyuConfig(app_key="test_key", app_secret="test_secret",
is_sandbox=True)
auth = XianyuAuthClient(cfg)
store = MemoryTokenStore()
print("=== Step1: 生成授权URL ===")
url, state = auth.build_authorize_url(force_auth=True)
print(f" 授权URL: {url[:120]}...")
print(f" state(存session, 回调时校验): {state}")
print("\n=== Step2: 模拟回调 code=authcode_abc ===")
token = auth.exchange_code("authcode_abc")
print(f" SessionKey(access_token): {token.access_token}")
print(f" refresh_token: {token.refresh_token}")
print(f" 过期时间: {token.expires_in}s, refresh有效期: {token.re_expires_in}s")
print("\n=== Step3: TokenManager (自研应用=只能重新授权) ===")
token.shop_id = "shop_001"
store.save(token)
mgr = TokenManager(auth, store, app_type=AppType.SELF_USE)
got = mgr.get("shop_001")
print(f" 取到Token: {got.access_token[:12]}... 需刷新={got.needs_refresh()}")
print("\n=== Step3: 订购类服务商 (可refresh续r2) ===")
token2 = TokenSet(shop_id="shop_002", taobao_user_id="2",
access_token="old", refresh_token="rt", re_expires_in=999999,
app_type=AppType.PURCHASE_SERVICE)
store.save(token2)
mgr2 = TokenManager(auth, store, app_type=AppType.PURCHASE_SERVICE)
# 模拟即将过期
token2.created_at = time.time() - 90000 # 已过expires_in, 需刷新
refreshed = mgr2.get("shop_002")
print(f" refresh后续期: {refreshed.access_token}")
print("\n=== TOP API 调用 (自动注入SessionKey) ===")
api = TopApiClient(cfg, mgr2)
resp = api.call("alibaba.idle.isv.order.query", "shop_002", {"page": 1})
print(f" 调用: {resp}")
print("\n=== Step5: 聚石塔合规自检 ===")
check = JushitaCheck(
app_tags=["ERP/进销存软件", "订单管理"],
data_in_rds=True, api_from_inside=True, write_from_inside=True,
cross_tower_via_qimen=True, three_tier=True, yc_isolation=True)
r = check.evaluate()
print(f" 强制入塔: {r['must_jushita']}, 通过: {r['pass']}")
for i in r["issues"]: print(f" ⚠️ {i}")
print("\n=== 反例: 塔外调用+数据外泄 ===")
bad = JushitaCheck(app_tags=["订单管理"], data_in_rds=False,
api_from_inside=False, write_from_inside=False,
cross_tower_via_qimen=False, three_tier=False, yc_isolation=False)
rb = bad.evaluate()
print(f" 通过: {rb['pass']}")
for i in rb["issues"]: print(f" ❌ {i}")跑出来关键几行（正是链路实证）：
=== Step1: 生成授权URL ===
授权URL: https://oauth.taobao.com/authorize?response_type=code&client_id=test_key&...
state(存session, 回调时校验): ...
=== Step2: 模拟回调 code=authcode_abc ===
SessionKey(access_token): 6101_authcode_a
refresh_token: refresh_authco
过期时间: 86400s, refresh有效期: 7776000s
=== Step3: TokenManager (自研应用=只能重新授权) ===
取到Token: 6101_authco... 需刷新=False
=== Step3: 订购类服务商 (可refresh续r2) ===
refresh后续期: new_17...
=== TOP API 调用 (自动注入SessionKey) ===
调用: {'_mock': True, 'session': 'new_17...'}
=== Step5: 聚石塔合规自检 ===
强制入塔: True, 通过: True
=== 反例: 塔外调用+数据外泄 ===
通过: False
❌ 订单/会员数据必须存塔内RDS
❌ TOP API调用必须从塔内发起
❌ 高风险写API必须从塔内发起
❌ 须三层架构(客户端只展现)
❌ 须配置御城河安全边界隔离
❌ 塔外交互须走奇门标准接口+审批六、六个上线前铁律
code 一次性 + 5 分钟：回调拿到 code 立即换 SessionKey，换完就失效；code 绝不落库、不重试。
SessionKey 按 shop_id 隔离：
taobao_user_id是授权账号身份，"用哪个淘宝账号授权就只能查哪个商家数据"——多店铺 ERP 必须 shop_id → SessionKey 一对一，绝不允许共用主账号 token（呼应前篇JdKeyTypeGuard隔离思想）。刷新策略看应用类型：
AppType.SELF_USE（自研/商家后台/新业务）refresh 只能续 r2 不能延长总时长，真要续命用force_auth=true重新登录；只有PURCHASE_SERVICE（订购类服务商）才能靠taobao.top.auth.token.refresh续期。提前续期：
needs_refresh(ahead_sec=1800)过期前 30 分钟主动刷，不要等 401，否则大促高峰刷新失败=订单同步中断。聚石塔是硬门槛不是优化项：23 类应用强制入塔，订单/会员存 RDS、写 API 塔内发起、塔外走奇门——本地能跑通不代表上线合规，必须自检通过。
AppSecret 永不进 URL：签名和 token 交换都走 POST body；前端/日志/仓库绝不出现 secret（前篇
ComplianceGate审计可覆盖）。
七、和前几篇的衔接
把本篇TokenManager+TopApiClient作为所有闲鱼 Adapter 的基座，装配进前几篇体系：
TokenManager替换前篇IdleIsvShipClient/RefundSyncHandler里的token_store，统一成 shop_id → TokenSet 持久化，刷新逻辑一处维护；
TopApiClient.call()的签名+session注入复用前篇ApiGateway的 MD5 策略，作为淘宝/闲鱼体系的统一调用入口；
JushitaCheck.evaluate()挂到服务启动时自检（CI + 生产启动），不合规直接exit(1)，与CertGuard.self_check()、ComplianceGate组成启动门禁三件套；
needs_refresh告警接入ObservabilityMiddleware，提前续期失败 → 企微告警 → 触发重新授权流程（避免静默断联）。
授权/Token/部署这三件事是闲鱼接入的"地基"——地基不稳，上层发布、发货、退款、限流、成本归因全部白搭。
要不要我把
xianyu_auth_platform.py + 前几篇的 idle_isv_ship.py / idle_refund_sync.py / idle_item_publish_mapper.py / idle_rate_limiter.py / compliance_gate.py 整合成一个完整的 commerce-mesh/adapters/idle/ 闲鱼子模块（授权→发布→发货→退款→限流→合规→字段映射全闭环），并补聚石塔 Dockerfile + 健康检查 + 部署清单？