feat(qoder): 弄清国际版每日 100 Credits 的「按设备下发」规则 + 一键设备授权登录

排查结论(本机对真实接口做了 30+ 组对照实验)
- GET /sash/api/v1/me/campaigns 只返回一条常驻 VIEW_DETAILS 广告
  (act-20260901-493,BOGO 订阅推广,claimStatus=CLAIMED 与签到无关)
- 无论 Cosy-MachineToken/Code/Type 用 env 0/3/4/prod、裸调用/带 stdin、
  是否附 Cosy-MachineId/OS/Hostname/Origin/Referer/X-*、是否带 clientType,
  都拿不到 CLAIM_BEFIT 轮次;账号本人 balances/achievements 正常
- 三个社区实现(qoder2api-hub、dsh-connect-qoder、CreditDaddy)一致:
  每日轮次按「登录过官方客户端的那台设备」下发,NAS 侧另造的设备看不到它

修复
- _benefits() 改为只认 actionType=CLAIM_BEFIT(不再把 VIEW_DETAILS 当签到),
  窗口过期也排除;claimStatus 未知按可领处理(领取接口幂等)
- 两段式取活动:先裸取拿 uid → 按 uid 生成风控身份 → 再带设备身份重取
- 状态文案改为「未下发每日签到活动」而不是误导性的「今日已领取」

新功能
- Qoder 一键登录(OAuth 设备授权 / PKCE S256):应用内点按钮 → 浏览器确认一次,
  Token 直接绑定本机设备;同 uid 已存在则更新凭证而不是新增重复账号
- 手工导入桌面端设备身份接口 /api/qoder/import_device
- 诊断接口:/api/qoder/{probe,experiment,growth,matrix,desktop}

测试:39 后端 + 25 E2E + 9 前端全绿
This commit is contained in:
AgentCheckin
2026-10-19 12:49:06 +08:00
parent c263edc3e7
commit 556e45d642
11 changed files with 943 additions and 70 deletions
+560 -32
View File
@@ -24,6 +24,7 @@ exactly at midnight, so multi-slot retries matter).
import base64
import json
import time
import uuid
from common import http_post_json, http_get, log, now_ts
import umid
@@ -34,8 +35,15 @@ PATH_CAMPAIGNS = "/sash/api/v1/me/campaigns"
PATH_CLAIM = "/sash/api/v1/me/campaigns/%s/claim"
PATH_REFRESH = "/api/v1/deviceToken/refresh"
# 只有这个 actionType 会真的发积分(VIEW_DETAILS 是订阅推广,不发放)
CLAIM_BENEFIT_ACTION = "CLAIM_BENEFIT"
CLAIMED_STATUS = "CLAIMED"
REFRESH_MARGIN = 72 * 3600 # refresh 72h before expiry (client behaviour)
# 活动 iframe(桌面客户端「每天领 100 Credits」卡片就由它渲染)
GROWTH_URL = "https://openapi.qoder.sh/growth-page/activity-iframe"
# 客户端默认版本(社区实现实测值:缺 Cosy-Version 时 campaigns 可能返回空数组)
DEFAULT_VERSION = "0.3.4"
@@ -78,20 +86,27 @@ def device_headers(account):
def risk_identity(account, force=False):
"""获取(并缓存到账号)设备风控身份 Cosy-MachineToken/Code/Type。
Qoder 服务端(尤其国际版)只对带风控身份的请求下发「每天领 100 Credits」活动。
优先用账号 device 块里已存的;缺失时调用本机 runtime-info 现场生成。
Qoder 服务端只对带风控身份的请求下发「每天领 100 Credits」活动
(国际版尤其严格)。身份必须按账号 uid 生成——uid 从活动接口响应里取,
所以调用前应先跑一次 query_campaigns()。
"""
dev = account.get("device") or {}
if not force and dev.get("machineToken") and dev.get("machineCode") and dev.get("machineType"):
uid = account.get("uid") or ""
have = dev.get("machineToken") and dev.get("machineCode") and dev.get("machineType")
if not force and have and (dev.get("uid") or "") == uid:
return dev["machineToken"], dev["machineCode"], dev["machineType"], None
provider = "qoder" if is_global(account.get("auth") or {}) else "qoder-cn"
ident, err = umid.risk_identity(provider, account.get("uid") or "")
ident, err = umid.risk_identity(provider, uid)
if err:
if have:
# 生成失败时退回旧身份,至少让请求能带上设备标识
return dev["machineToken"], dev["machineCode"], dev["machineType"], err
return None, None, None, err
dev = dict(dev)
dev.update({"machineToken": ident["machineToken"],
"machineCode": ident["machineCode"],
"machineType": ident["machineType"]})
"machineType": ident["machineType"],
"uid": uid})
account["device"] = dev
return dev["machineToken"], dev["machineCode"], dev["machineType"], None
@@ -214,9 +229,9 @@ def _api(account, path, method="GET", body=None):
else http_get(billing_base(account.get("auth") or {}) + path, headers=_headers(account))
def query_campaigns(account):
"""Returns (campaigns, error). campaigns is a list (possibly empty)."""
data, status, err = _api(account, PATH_CAMPAIGNS + "?forceRefresh=true", "GET")
def query_campaigns_full(account):
"""活动接口的完整响应(含 uid / showCampaign / claimable)。"""
data, status, err = _api(account, PATH_CAMPAIGNS + "?clientType=10&forceRefresh=true", "GET")
if status == 401:
return None, "no_session"
if not isinstance(data, dict) or status != 200:
@@ -224,6 +239,20 @@ def query_campaigns(account):
if isinstance(data, dict):
detail = data.get("message") or data.get("errorMessage") or ""
return None, "查询失败(HTTP %s)%s" % (status, (":" + str(detail)[:80]) if detail else "")
return data, None
def query_campaigns(account):
"""Returns (campaigns, error). campaigns is a list (possibly empty).
顺带记录响应里的 uid —— 设备风控身份必须按 uid 生成(见 risk_identity)。
"""
data, err = query_campaigns_full(account)
if err:
return None, err
uid = data.get("uid")
if isinstance(uid, str) and uid:
account["uid"] = uid
campaigns = data.get("campaigns")
return (campaigns if isinstance(campaigns, list) else []), None
@@ -252,31 +281,37 @@ def campaigns_debug(account):
return out
def _benefits(campaigns):
"""CLAIM_BENEFIT campaigns(活动/签到坑位)。
def _window_open(c, now=None):
"""活动是否在有效期内(缺字段视为开放)。"""
now = now if now is not None else now_ts()
start = c.get("startAt") or c.get("beginAt")
end = c.get("endAt")
try:
if start and now < float(start):
return False
if end and now > float(end):
return False
except (TypeError, ValueError):
return True
return True
注意:真实数据里 benefit 可能只有 kind 没有 amount(活动信息在别处),
所以只按 actionType 过滤;kind 明确非 CREDITS 时才排除。
def _benefits(campaigns):
"""可领取的签到轮次:只有 actionType == CLAIM_BENEFIT。
实测(2026-09-27 两区,本机复核 2026-10-19):
同一账号还挂着一条 actionType=VIEW_DETAILS 的订阅推广活动(无 benefit、
claimStatus 却是 CLAIMED)——只看 claimStatus 会把它当成签到活动去「领取」。
而 claimStatus 是判断「本轮是否已领」的唯一可靠字段:响应顶层的
claimable 在「今天已领」和「当前没有活动」两种情况下都是 false。
"""
out = []
for c in campaigns:
if not isinstance(c, dict):
continue
action = c.get("actionType")
claim_status = c.get("claimStatus")
benefit = c.get("benefit")
kind = benefit.get("kind") if isinstance(benefit, dict) else None
# 判定规则(已实测):
# 国内版:actionType == CLAIM_BENEFIT 且 claimStatus 为 CLAIMABLE/CLAIMED
# 国际版:actionType 为 VIEW_DETAILS(点「领取」按钮才领),claimStatus 同样有效
is_benefit_action = action in ("CLAIM_BENEFIT", "VIEW_DETAILS")
has_claim_state = claim_status in ("CLAIMABLE", "CLAIMED")
if not (is_benefit_action or (action is None and has_claim_state)):
if c.get("actionType") != CLAIM_BENEFIT_ACTION:
continue
if not has_claim_state:
continue
# 只排除明确非积分权益(如实物/优惠券)
if kind and kind not in ("CREDITS",):
if not _window_open(c):
continue
out.append(c)
return out
@@ -384,6 +419,8 @@ def checkin_once(account):
if not (at and (not exp or exp > now_ts())):
return {"status": "no_session", "message": msg, "points": 0}
# 两段式取活动:先裸取一次拿到 uid(风控身份按 uid 生成),
# 带上设备身份后再取一次——国际版只有带设备身份的请求才会下发每日活动。
campaigns, err = query_campaigns(account)
if err == "no_session":
ok2, msg2 = ensure_token(account, force=True)
@@ -402,10 +439,13 @@ def checkin_once(account):
if err:
return {"status": "fail", "message": err, "points": 0}
# Qoder 服务端只对带风控身份的请求下发签到活动——先确保身份就绪
tok, code, mtype, rerr = risk_identity(account)
if rerr:
log("qoder risk identity unavailable: %s" % rerr, "WARN")
if tok:
campaigns2, err2 = query_campaigns(account)
if not err2 and campaigns2 is not None:
campaigns = campaigns2
balance, berr = query_balance(account)
if berr == "no_session":
@@ -415,7 +455,9 @@ def checkin_once(account):
balance = None
benefits = _benefits(campaigns)
claimable = [c for c in benefits if c.get("claimStatus") == "CLAIMABLE"]
# claimStatus 未知时按「可领」处理:服务端领取接口幂等(replayed 不再发放),
# 乐观的代价是一次多余请求,保守的代价是把功能永久藏起来。
claimable = [c for c in benefits if c.get("claimStatus") != CLAIMED_STATUS]
if claimable:
got, fail = 0.0, 0
@@ -447,10 +489,398 @@ def checkin_once(account):
return {"status": "already", "message": "今日已领取", "points": 0,
"credits": balance}
if not benefits:
# 没有任何可识别活动:可能是活动未下发/已结束,也可能是 schema 变化
return {"status": "already",
"message": "未发现可领取活动(活动未下发/已结束,或需刷新风控身份)", "points": 0}
return {"status": "retry_later", "message": "活动未下发,下个时段再试", "points": 0}
# 服务端只下发了 VIEW_DETAILS 类推广(或不含每日轮次)——国际版常见原因:
# 活动按「登录过官方客户端的那台设备」下发,本机设备未被绑定。
hint = ("未下发每日签到活动:服务端没有给本机设备下发 CLAIM_BENEFIT 轮次。"
"国际版的活动按设备下发,可在「添加账号 → Qoder → 一键登录」用设备授权"
"把本机绑定到账号后重试。")
return {"status": "retry_later", "message": hint, "points": 0, "credits": balance}
return {"status": "retry_later", "message": "活动未下发,下个时段再试", "points": 0,
"credits": balance}
# ---------------------------------------------------------------------------
# 诊断:原始响应(排查活动 schema / claimStatus 语义差异)
# ---------------------------------------------------------------------------
def _raw_call(account, path, method="GET", body=None):
"""一次 API 调用,返回 {status,error,body}(body 为服务端原始 JSON)。"""
try:
data, status, err = _api(account, path, method, body)
except Exception as e: # pragma: no cover - 网络异常
return {"status": 0, "error": str(e)[:200]}
out = {"status": status}
if err:
out["error"] = str(err)[:200]
out["body"] = data
return out
def _collect_campaign_ids(obj, into):
if isinstance(obj, dict):
for k, v in obj.items():
if k == "campaignId" and isinstance(v, str) and v:
into.append(v)
else:
_collect_campaign_ids(v, into)
elif isinstance(obj, list):
for it in obj:
_collect_campaign_ids(it, into)
def raw_probe(account):
"""转储 Qoder 接口的原始 JSON(排查 claimStatus 语义 / 活动 schema)。
会先记录余额,再对窗口内的活动调用一次领取接口(官方接口幂等,
重复领取返回 replayed,不会重复发放),最后记录余额变化。
"""
out = {"headers": {}, "domain": (account.get("auth") or {}).get("domain")}
for k, v in _headers(account).items():
out["headers"][k] = ("Bearer <redacted>" if k == "Authorization"
else ("<set:%d>" % len(str(v)) if v else ""))
camp = _raw_call(account, PATH_CAMPAIGNS + "?forceRefresh=true")
out["campaigns"] = camp
out["campaigns_no_force"] = _raw_call(account, PATH_CAMPAIGNS)
out["campaigns_v2"] = _raw_call(account, "/sash/api/v2/me/campaigns?forceRefresh=true")
cids = []
_collect_campaign_ids(camp.get("body"), cids)
cids = list(dict.fromkeys(cids))
out["campaign_ids"] = cids
out["usage"] = _raw_call(account, USAGE_PATH)
b0, _ = query_balance(account)
out["balance_before"] = b0
if cids:
cid = cids[0]
out["campaign_detail"] = _raw_call(account, "/sash/api/v1/me/campaigns/" + cid)
out["claim"] = _raw_call(account, PATH_CLAIM % cid, "POST", body={})
time.sleep(1.5)
out["balance_after"] = query_balance(account)[0]
out["usage_after"] = _raw_call(account, USAGE_PATH)
out["campaigns_after_claim"] = _raw_call(account, PATH_CAMPAIGNS + "?forceRefresh=true")
return out
def _summ(data):
"""活动响应摘要(不含 machineToken 等敏感值)。"""
if not isinstance(data, dict):
return {"raw": str(data)[:200]}
out = {"uid": data.get("uid"), "showCampaign": data.get("showCampaign"),
"claimable": data.get("claimable")}
cs = data.get("campaigns")
if isinstance(cs, list):
out["count"] = len(cs)
out["campaigns"] = [{
"key": c.get("campaignKey"),
"actionType": c.get("actionType"),
"claimStatus": c.get("claimStatus"),
"placements": [p.get("type") for p in (c.get("placements") or []) if isinstance(p, dict)],
} for c in cs if isinstance(c, dict)]
return out
def _base_headers(account, risk=False):
"""基础请求头;risk=False 时不带任何 Cosy 设备/风控头。"""
h = {"Accept": "application/json", "Accept-Language": "zh-CN", "User-Agent": "Qoder"}
tok = (account.get("auth") or {}).get("accessToken", "")
if tok:
h["Authorization"] = "Bearer %s" % tok
if risk:
h["Cosy-ClientType"] = "10"
h["Cosy-Version"] = DEFAULT_VERSION
return h
def checkin_experiment(account):
"""定位「每日 100 Credits」活动下发条件的对照实验(只读,不领取)。"""
base = billing_base(account.get("auth") or {})
out = {"domain": base}
# 1) 不带风控身份取一次,拿到 uid / showCampaign / claimable
plain, st, err = http_get(base + PATH_CAMPAIGNS, headers=_base_headers(account, risk=True))
out["step1_plain"] = {"status": st, "error": err, "summary": _summ(plain)}
uid = (plain or {}).get("uid") or account.get("uid") or ""
out["uid"] = uid
# 2) 不同 env 生成的身份(看输出字段差异)
idents = {}
for env in (0, 3, 4):
j, e = umid.run_runtime_info(env, uid)
idents["env%s" % env] = {
"error": e,
"fields": sorted(j.keys()) if isinstance(j, dict) else None,
"preview": {k: (str(v)[:100] if isinstance(v, str) else v)
for k, v in (j or {}).items() if k != "machineToken"},
"token_len": len((j or {}).get("machineToken") or ""),
}
out["identities"] = idents
# 3) 组合对照:env × 是否带 Cosy-MachineId/OS/Hostname × 是否带 clientType 查询参数
m_id = str(uuid.uuid4())
out["machine_id_used"] = m_id
combos = {}
for env in (3, 4):
j, e = umid.run_runtime_info(env, uid)
if e or not isinstance(j, dict):
combos["env%s" % env] = {"error": e}
continue
for mode in ("token_only", "full"):
h = _base_headers(account, risk=True)
h["Cosy-MachineToken"] = j.get("machineToken", "")
h["Cosy-MachineCode"] = j.get("machineCode", "")
h["Cosy-MachineType"] = j.get("machineType", "")
if mode == "full":
h["Cosy-MachineId"] = m_id
h["Cosy-MachineOS"] = "x86_64_linux"
h["Cosy-MachineHostname"] = "fnOS"
for qs, tag in (("", "noq"), ("?clientType=10", "ct10")):
d2, s2, e2 = http_get(base + PATH_CAMPAIGNS + qs, headers=h)
combos["env%s_%s_%s" % (env, mode, tag)] = {
"status": s2, "error": e2, "summary": _summ(d2)}
out["combos"] = combos
return out
def _snips(text, pattern, limit, width):
"""返回 pattern 命中处 ±width 字符的片段(去重、限量)。"""
import re
seen, out = set(), []
for m in re.finditer(pattern, text):
a = max(0, m.start() - width)
b = min(len(text), m.end() + width)
s = text[a:b]
key = s[:120]
if key in seen:
continue
seen.add(key)
out.append(s)
if len(out) >= limit:
break
return out
def growth_probe(account):
"""扒「活动 iframe」页面与其 JS,找出每日签到真正调用的接口。"""
import re
from urllib.parse import urljoin
from common import http_get_text
out = {}
html, status, err = http_get_text(GROWTH_URL, headers={"Accept": "text/html"})
out["page"] = {"status": status, "error": err, "len": len(html or ""),
"head": (html or "")[:2000]}
srcs = re.findall(r'<script[^>]+src=["\']([^"\']+)["\']', html or "", re.I)
out["scripts"] = srcs[:8]
bundles = {}
for s in srcs[:4]:
u = urljoin(GROWTH_URL, s)
js, st, e = http_get_text(u, headers={"Accept": "*/*"})
body = js or ""
paths = sorted(set(re.findall(
r'["\'`](/(?:sash|api|growth|oauth)[A-Za-z0-9_\-/{}.$:]*?)["\'`]', body)))
hits = sorted(set(re.findall(
r'[A-Za-z0-9_./-]*(?:checkin|dailyCheck|daily_credit|claim|credit|campaign)'
r'[A-Za-z0-9_./-]*', body, re.I)))
info = {"status": st, "error": e, "len": len(body),
"paths": paths[:120], "keywords": hits[:120]}
if "/sash/api/v1/me/campaigns" in body:
# 前端构造请求的上下文(请求头 / fetch 选项)
info["api_snips"] = _snips(body, r'/sash/api/v1/me/campaigns', 1, 700)
info["cosy_names"] = sorted(set(re.findall(r'Cosy-[A-Za-z0-9-]+', body)))
info["cosy_snips"] = _snips(body, r'Cosy-[A-Za-z0-9-]+', 6, 300)
info["hdr_snips"] = _snips(body, r'Authorization|machineToken|credentials', 6, 260)
bundles[u] = info
out["bundles"] = bundles
# 候选接口探测(404 = 该路径不存在,其它状态码值得注意)
base = billing_base(account.get("auth") or {})
h = _headers(account)
cands = [
"/sash/api/v1/me/campaigns?clientType=10&forceRefresh=true",
"/sash/api/v1/me/campaigns?clientType=10&placement=USAGE",
"/sash/api/v1/me/dailyCheckin", "/sash/api/v1/me/daily-checkin",
"/sash/api/v1/me/checkin", "/sash/api/v1/me/check-in",
"/sash/api/v1/me/rewards", "/sash/api/v1/me/benefits",
"/sash/api/v1/me/campaigns/claimable",
"/sash/api/v1/growth/campaigns", "/sash/api/v1/me/growth/campaigns",
"/sash/api/v2/me/campaigns", "/sash/api/v1/me/credits/daily",
"/api/v2/me/usage", "/sash/api/v1/me/daily-credits",
]
out["paths"] = {}
for p in cands:
d, st, e = http_get(base + p, headers=h)
out["paths"][p] = {"status": st, "error": e,
"summary": _summ(d) if isinstance(d, dict) else None}
return out
DAILY_ACTION_HINT = "CLAIM_BENEFIT"
def identity_matrix_probe(account):
"""对照实验:哪种 umid 调用方式 + 哪种请求头组合能拿到 CLAIM_BENEFIT。
只读,不会领取(发现 CLAIMABLE 也只报告)。
"""
base = billing_base(account.get("auth") or {})
tok = (account.get("auth") or {}).get("accessToken", "")
auth = {"Authorization": "Bearer %s" % tok} if tok else {}
out = {}
# 1) 裸 bearer 拿 uid
plain_hdr = dict(auth, **{"Accept": "application/json", "User-Agent": "Qoder",
"Cosy-ClientType": "10"})
d, st, e = http_get(base + PATH_CAMPAIGNS, headers=plain_hdr)
out["plain"] = {"status": st, "error": e, "summary": _summ(d)}
uid = (d or {}).get("uid") or account.get("uid") or ""
out["uid"] = uid
# 2) 各种 umid 调用方式
stdin = json.dumps({"account": uid}) + "\n"
modes = {
"env4_stdin": (["4", "--account-stdin"], stdin),
"env4_bare": (["4"], None),
"env3_stdin": (["3", "--account-stdin"], stdin),
"env3_bare": (["3"], None),
"prod_stdin": (["prod", "--account-stdin"], stdin),
"prod_bare": (["prod"], None),
"noarg": ([], None),
}
idents = {}
for name, (args, sd) in modes.items():
j, err = umid.run_runtime_info_raw(args, sd)
idents[name] = {"error": err,
"token_len": len((j or {}).get("machineToken") or ""),
"code": (j or {}).get("machineCode"),
"type": (j or {}).get("machineType"),
"vm": (j or {}).get("vmInfo")}
out["identities"] = idents
# 3) 各身份 × 各请求头组合
def _hdr(ident, extra):
h = dict(auth, **{"Accept": "application/json", "User-Agent": "Qoder",
"Cosy-ClientType": "10", "Cosy-Version": DEFAULT_VERSION})
if ident:
h["Cosy-MachineToken"] = ident.get("machineToken", "")
h["Cosy-MachineCode"] = ident.get("machineCode", "")
h["Cosy-MachineType"] = ident.get("machineType", "")
h.update(extra)
return h
web_extra = {"Origin": "https://qoder.com", "Referer": "https://qoder.com/usage"}
x_extra = {"X-Request-ID": str(uuid.uuid4()), "X-Machine-ID": str(uuid.uuid4()),
"X-Session-ID": str(uuid.uuid4())}
combos = {}
tried = []
for name in ("env4_bare", "env4_stdin", "prod_stdin", "noarg"):
j, _ = umid.run_runtime_info_raw(*modes[name])
if not j:
continue
for label, extra in (("min", {}), ("web", web_extra), ("x", x_extra),
("webx", dict(web_extra, **x_extra))):
h = _hdr(j, extra)
for qs in ("", "?clientType=10"):
key = "%s_%s%s" % (name, label, qs and "_ct10" or "")
if key in combos:
continue
d2, s2, e2 = http_get(base + PATH_CAMPAIGNS + qs, headers=h)
s = _summ(d2)
combos[key] = {"status": s2, "error": e2, "summary": s}
tried.append((key, s))
out["combos"] = combos
# 4) 命中 CLAIM_BENEFIT 的(不领取,只报告)
hits = []
for key, s in tried:
for c in (s.get("campaigns") or []):
if c.get("actionType") == DAILY_ACTION_HINT:
hits.append({"combo": key, "campaign": c,
"top_claimable": s.get("claimable")})
out["hits"] = hits
return out
def _derive_machine_id(uid):
"""官方桌面端同款 machineId(社区实现:md5("machine:"+uid)[:36])。"""
import hashlib
return hashlib.md5(("machine:%s" % uid).encode("utf-8")).hexdigest()[:36]
def desktop_emulation_probe(account):
"""模拟官方桌面端出站头,寻找能拿到 CLAIM_BENEFIT 的组合(只读)。"""
base = billing_base(account.get("auth") or {})
tok = (account.get("auth") or {}).get("accessToken", "")
auth = {"Authorization": "Bearer %s" % tok} if tok else {}
d, st, e = http_get(base + PATH_CAMPAIGNS, headers=dict(auth, **{
"Accept": "application/json", "User-Agent": "Qoder", "Cosy-ClientType": "10"}))
uid = (d or {}).get("uid") or account.get("uid") or ""
out = {"uid": uid, "plain": {"status": st, "error": e, "summary": _summ(d)}}
ident, ierr = umid.risk_identity("qoder", uid)
out["identity_error"] = ierr
out["identity"] = {k: (str(v)[:24] if isinstance(v, str) else v)
for k, v in (ident or {}).items()}
mid = _derive_machine_id(uid)
out["machine_id"] = mid
umid_hdr = {}
if ident:
umid_hdr = {"Cosy-MachineToken": ident["machineToken"],
"Cosy-MachineCode": ident["machineCode"],
"Cosy-MachineType": ident["machineType"]}
win_hdr = {"Cosy-MachineOS": "x86_64_win32", "Cosy-MachineHostname": "DESKTOP-QODER",
"Cosy-MachineId": mid}
web = {"Origin": "https://qoder.com", "Referer": "https://qoder.com/"}
variants = {
# 官方桌面端同款:全套六头 + win32 外形
"desktop_full": dict(umid_hdr, **win_hdr),
"desktop_full_web": dict(umid_hdr, **dict(win_hdr, **web)),
# 去掉 machineId / machineToken(issue #10 的「去掉即恢复」)
"no_machineid": dict(umid_hdr, **{k: v for k, v in win_hdr.items()
if k != "Cosy-MachineId"}),
"no_token": dict({k: v for k, v in umid_hdr.items() if k != "Cosy-MachineToken"},
**win_hdr),
# 只发部分
"only_machineid": {"Cosy-MachineId": mid},
"only_os_host": {"Cosy-MachineOS": "x86_64_win32",
"Cosy-MachineHostname": "DESKTOP-QODER"},
"umid_only": dict(umid_hdr),
# umid + OS(不带 machineId / hostname)
"umid_os": dict(umid_hdr, **{"Cosy-MachineOS": "x86_64_win32"}),
}
out["combos"] = {}
hits = []
for name, extra in variants.items():
h = dict(auth, **{"Accept": "application/json", "User-Agent": "Qoder",
"Cosy-ClientType": "10", "Cosy-Version": "0.4.3"})
h.update(extra)
for qs in ("", "?clientType=10"):
d2, s2, e2 = http_get(base + PATH_CAMPAIGNS + qs, headers=h)
s = _summ(d2)
key = name + ("_ct10" if qs else "")
out["combos"][key] = {"status": s2, "error": e2, "summary": s}
for c in (s.get("campaigns") or []):
if c.get("actionType") == DAILY_ACTION_HINT:
hits.append({"combo": key, "campaign": c,
"claimable": s.get("claimable")})
out["hits"] = hits
# 其它可能暴露资格/成就的接口(只读)
extra_paths = ["/sash/api/v1/me/achievements", "/api/v2/quota/usage",
"/sash/api/v1/me/campaigns/%s/reward"
% "01a05bce-e800-7494-a002-806e4438f483"]
out["extra"] = {}
h = dict(auth, **{"Accept": "application/json", "User-Agent": "Qoder",
"Cosy-ClientType": "10", "Cosy-Version": "0.4.3"})
h.update(umid_hdr)
for p in extra_paths:
d3, s3, e3 = http_get(base + p, headers=h)
body = d3 if isinstance(d3, dict) else None
if body is not None and len(json.dumps(body, ensure_ascii=False)) > 6000:
body = {"_truncated": True, "_keys": list(body.keys())[:20]}
out["extra"][p] = {"status": s3, "error": e3, "body": body}
return out
def probe_domains(account):
@@ -473,5 +903,103 @@ def probe_domains(account):
return out
# ---------------------------------------------------------------------------
# OAuth 设备授权登录(PKCE S256)——「一键登录」,同时也把设备绑到本机
# ---------------------------------------------------------------------------
REALMS = {
"intl": {
"name": "国际版",
"openapi": GLOBAL_BASE,
"website": "https://qoder.com",
"client_id": "e883ade2-e6e3-4d6d-adf7-f92ceff5fdcb",
"redirect_uri": "",
"nonce_dashed": False,
},
"cn": {
"name": "国内版",
"openapi": CN_BASE,
"website": "https://qoder.com.cn",
"client_id": "1c5e33e1-364d-4ce6-b02c-acaa81274a5c",
"redirect_uri": "qoder-work-cn://",
"nonce_dashed": True,
},
}
PATH_DEVICE_POLL = "/api/v1/deviceToken/poll"
def realm_of(domain):
return "cn" if "com.cn" in (domain or "").lower() else "intl"
def _b64url(raw):
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
def device_login_start(realm="intl", machine_id=""):
"""构造 PKCE 设备授权链接。返回 {realm, auth_url, verifier, nonce, machine_id}."""
import hashlib
import os
import urllib.parse
cfg = REALMS.get(realm) or REALMS["intl"]
verifier = _b64url(os.urandom(32))
challenge = _b64url(hashlib.sha256(verifier.encode("ascii")).digest())
nonce = str(uuid.uuid4()) if cfg["nonce_dashed"] else uuid.uuid4().hex
mid = machine_id or str(uuid.uuid4())
params = {"challenge": challenge, "challenge_method": "S256", "nonce": nonce}
if cfg["redirect_uri"]:
params["redirect_uri"] = cfg["redirect_uri"]
params["client_id"] = cfg["client_id"]
params["machine_id"] = mid
return {"realm": realm,
"auth_url": cfg["website"] + "/device/selectAccounts?" + urllib.parse.urlencode(params),
"verifier": verifier, "nonce": nonce, "machine_id": mid,
"openapi": cfg["openapi"]}
def device_login_poll(realm, verifier, nonce, machine_id="", with_device=True):
"""轮询设备授权结果。返回 {status: pending|ok|error, ...}.
带上本机设备身份(Cosy-Machine*)再轮询,让服务端把这次登录绑到本机设备——
国际版的每日 100 Credits 活动只下发给「登录过官方客户端的那台设备」。
"""
import urllib.parse
cfg = REALMS.get(realm) or REALMS["intl"]
q = urllib.parse.urlencode({"nonce": nonce, "verifier": verifier,
"challenge_method": "S256"})
url = cfg["openapi"] + PATH_DEVICE_POLL + "?" + q
headers = {"Accept": "application/json", "User-Agent": "QoderWork"}
if with_device and machine_id:
dev = {"machineId": machine_id, "clientType": "10", "version": DEFAULT_VERSION}
provider = "qoder-cn" if realm == "cn" else "qoder"
ident, _ierr = umid.risk_identity(provider, "")
if ident:
dev.update({"machineToken": ident.get("machineToken", ""),
"machineCode": ident.get("machineCode", ""),
"machineType": ident.get("machineType", "")})
for k, v in device_headers({"device": dev}).items():
if v:
headers[k] = v
data, status, err = http_get(url, headers=headers)
if status not in (200, 202, 404) and with_device:
# 设备头可能让轮询接口不接受:去掉再试一次
return device_login_poll(realm, verifier, nonce, machine_id, with_device=False)
if status in (404, 202):
return {"status": "pending"}
if status != 200 or not isinstance(data, dict):
return {"status": "error", "message": "轮询失败(HTTP %s)%s"
% (status, (":" + str(err)[:120]) if err else "")}
token = data.get("token") or data.get("device_token") or ""
if not token:
return {"status": "pending"}
return {"status": "ok", "accessToken": token,
"refreshToken": data.get("refresh_token") or "",
"uid": str(data.get("user_id") or ""),
"expiresAt": jwt_exp(token),
"realm": realm,
"domain": cfg["openapi"],
"machine_id": machine_id}
def account_display(account):
return account.get("name") or account.get("uid") or "Qoder 账号"