whale-fall-code
2bb83f3685
App 端免签名接口（设备注册 + detail + CDN 直链），一体化流水线 下载→抽音→传COS→记录→删本地，AI 作品四档判定，四份文档： README（使用指南）、PROGRESS（批次进度）、RE（逆向全过程）、AI（判定规则）。 Co-Authored-By: Claude Code <noreply@anthropic.com>
157 lines
5.4 KiB
Python
157 lines
5.4 KiB
Python
"""a_bogus 签名器。
|
||
|
||
两种后端，接口相同、可互换：
|
||
|
||
NodeSigner 调本地 node_signer（真实 webmssdk 跑在伪造浏览器环境里）。
|
||
当前唯一可用的实现，已实测能被抖音接受。
|
||
ABogusSigner 纯 Python 实现。底层 SM3/RC4/自定义 Base64 已在 crypto.py
|
||
验证通过，剩字节拼装顺序待从 ab.js 逆出来。
|
||
|
||
签名器同时决定 User-Agent：a_bogus 把 UA 编进签名，服务端会交叉校验，
|
||
所以请求头的 UA 必须与签名时用的完全一致。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import subprocess
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Dict, List, Protocol, Sequence
|
||
|
||
|
||
class SignError(RuntimeError):
|
||
pass
|
||
|
||
|
||
class SignerNotConfigured(SignError):
|
||
pass
|
||
|
||
|
||
class Signer(Protocol):
|
||
"""client 只依赖这两个成员。"""
|
||
|
||
@property
|
||
def user_agent(self) -> str:
|
||
"""签名时使用的 UA，请求头必须用同一个。"""
|
||
...
|
||
|
||
def sign_url(self, url: str, fp: str = "") -> str:
|
||
"""给一条裸 URL，返回带 a_bogus 的完整 URL。"""
|
||
...
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Node 后端
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
class NodeSigner:
|
||
"""通过 node_signer/sign.js 签名。
|
||
|
||
node 启动 + 加载 347KB 的 SDK 大约 0.7s，所以支持 batch 一次签多条，
|
||
把这份固定开销摊掉。
|
||
"""
|
||
|
||
def __init__(self, signer_dir: str | Path, node: str = "node",
|
||
timeout: float = 120.0) -> None:
|
||
self.dir = Path(signer_dir)
|
||
self.node = node
|
||
self.timeout = timeout
|
||
self._ua: str | None = None
|
||
|
||
script = self.dir / "sign.js"
|
||
if not script.exists():
|
||
raise SignerNotConfigured(f"找不到签名脚本: {script}")
|
||
|
||
def _run(self, args: Sequence[str], stdin: str | None = None) -> dict:
|
||
try:
|
||
proc = subprocess.run(
|
||
[self.node, "sign.js", *args], cwd=self.dir, input=stdin,
|
||
capture_output=True, text=True, encoding="utf-8",
|
||
timeout=self.timeout,
|
||
)
|
||
except FileNotFoundError as exc:
|
||
raise SignerNotConfigured(f"找不到 node 可执行文件: {self.node}") from exc
|
||
except subprocess.TimeoutExpired as exc:
|
||
raise SignError(f"签名超时（>{self.timeout}s）") from exc
|
||
|
||
out = (proc.stdout or "").strip()
|
||
if not out:
|
||
raise SignError(f"签名器无输出，rc={proc.returncode}，stderr={proc.stderr[:400]}")
|
||
try:
|
||
# sign.js 只在最后一行吐 JSON，前面可能有 SDK 自己的噪声
|
||
return json.loads(out.splitlines()[-1])
|
||
except ValueError as exc:
|
||
raise SignError(f"签名器输出不是 JSON: {out[:300]!r}") from exc
|
||
|
||
@property
|
||
def user_agent(self) -> str:
|
||
if self._ua is None:
|
||
probe = self._run(["https://www.douyin.com/aweme/v1/web/aweme/detail/?aweme_id=1", ""])
|
||
self._ua = probe.get("userAgent") or ""
|
||
if not self._ua:
|
||
raise SignError("签名器没有返回 userAgent")
|
||
return self._ua
|
||
|
||
def sign_url(self, url: str, fp: str = "") -> str:
|
||
res = self._run([url, fp])
|
||
if not res.get("ok"):
|
||
raise SignError(f"签名失败: {res.get('error')}")
|
||
self._ua = res.get("userAgent") or self._ua
|
||
return res["signedUrl"]
|
||
|
||
def sign_many(self, urls: List[str], fp: str = "") -> List[str]:
|
||
"""一次 node 启动签完多条。"""
|
||
if not urls:
|
||
return []
|
||
res = self._run(["--batch", fp], stdin=json.dumps(urls))
|
||
if not res.get("ok"):
|
||
raise SignError(f"批量签名失败: {res.get('error')}")
|
||
self._ua = res.get("userAgent") or self._ua
|
||
signed = []
|
||
for item in res["results"]:
|
||
if not item.get("ok"):
|
||
raise SignError(f"某条签名失败: {item.get('error')} url={item.get('rawUrl')}")
|
||
signed.append(item["signedUrl"])
|
||
return signed
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 纯 Python 后端（待补）
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
@dataclass
|
||
class ABogusConfig:
|
||
"""a_bogus 算法常量。必须与线上 webmssdk.js 当前版本一致。"""
|
||
|
||
alphabets: Dict[str, str] = field(default_factory=dict)
|
||
ua_key: bytes = b""
|
||
arguments: List[int] = field(default_factory=list)
|
||
browser_fp: str = ""
|
||
end_string: str = ""
|
||
user_agent: str = ""
|
||
|
||
def validate(self) -> None:
|
||
missing = [n for n in ("alphabets", "ua_key", "arguments",
|
||
"browser_fp", "end_string", "user_agent")
|
||
if not getattr(self, n)]
|
||
if missing:
|
||
raise SignerNotConfigured("a_bogus 常量未配置，缺少: " + ", ".join(missing))
|
||
|
||
|
||
class ABogusSigner:
|
||
"""纯 Python a_bogus。crypto.py 里的 SM3/RC4/Base64 已验证可用。"""
|
||
|
||
def __init__(self, config: ABogusConfig) -> None:
|
||
config.validate()
|
||
self.config = config
|
||
|
||
@property
|
||
def user_agent(self) -> str:
|
||
return self.config.user_agent
|
||
|
||
def sign_url(self, url: str, fp: str = "") -> str:
|
||
raise NotImplementedError(
|
||
"a_bogus 字节拼装顺序待从 ab.js 逆出。目前请用 NodeSigner。"
|
||
)
|