《数据合规前置：OAuth2 / HMAC 签名 / 脱敏 / 最小权限在二手ERP对接中的落地》（附Python源码）

不是所有平台都用 OAuth2。

• 闲鱼/淘宝开放平台：OAuth2 授权码 + 三方 session

• eBay：OAuth2 Client Credentials / Authorization Code

• Mercari：OAuth2 体系（Mercari ID Platform 基于 OAuth 2.0 / OIDC）

• Back Market：卖家后台生成 opaque token，Authorization: Basic <token>，没有 OAuth2、没有 scope、没有 refresh

所以“数据合规前置”不是统一接 OAuth2，而是：按平台能力分层——有 OAuth2 用 OAuth2，只有 opaque token 就当长期凭证管，webhook 统一走 HMAC，PII 统一脱敏，服务账号统一最小权限。

一、四件事分别解决什么问题

二、统一凭证仓库（先别把 secret 写代码里）

# secrets_vault.py
import os, json, hashlib
from dataclasses import dataclass
from typing import Optional

class SecretVault:
    """
    生产用 HashiCorp Vault / AWS Secrets Manager。
    这里用环境变量+内存做最小可运行版。
    """

    def __init__(self, backend: str = "env"):
        self.backend = backend

    def get(self, key: str) -> Optional[str]:
        return os.environ.get(f"ERP_SECRET_{key}")

    def get_oauth_client(self, platform: str):
        return {
            "client_id": self.get(f"{platform}_CLIENT_ID"),
            "client_secret": self.get(f"{platform}_CLIENT_SECRET"),
            "token_url": self.get(f"{platform}_TOKEN_URL"),
            "scope": self.get(f"{platform}_SCOPE"),
        }

    def get_hmac_secret(self, platform: str) -> str:
        return self.get(f"{platform}_WEBHOOK_SECRET") or ""

    def get_opaque_token(self, platform: str) -> str:
        # Back Market 这类
        return self.get(f"{platform}_OPAQUE_TOKEN") or ""

• OAuth2 client_secret：只在令牌交换时出现，不进日志

• opaque token（Back Market）：当长期凭证管，写进 Vault，支持人工吊销+更换

• HMAC secret：按平台隔离，季度轮换

三、OAuth2 客户端（eBay / Mercari 风格）

# oauth2_client.py
import time, requests
from dataclasses import dataclass

@dataclass
class OAuth2Token:
    access_token: str
    token_type: str = "Bearer"
    expires_in: int = 0
    scope: str = ""
    issued_at: float = 0.0

    @property
    def expired(self, skew: int = 60) -> bool:
        if not self.access_token:
            return True
        return time.time() >= (self.issued_at + self.expires_in - skew)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class OAuth2Client:
    def __init__(self, client_id: str, client_secret: str,
                 token_url: str, scope: str = ""):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = token_url
        self.scope = scope
        self._token: OAuth2Token | None = None

    def get_token(self) -> OAuth2Token:
        if self._token and not self._token.expired:
            return self._token

        resp = requests.post(
            self.token_url,
            data={"grant_type": "client_credentials", "scope": self.scope},
            auth=(self.client_id, self.client_secret),
            timeout=10,
        )
        resp.raise_for_status()
        d = resp.json()
        self._token = OAuth2Token(
            access_token=d["access_token"],
            token_type=d.get("token_type", "Bearer"),
            expires_in=int(d.get("expires_in", 3600)),
            scope=d.get("scope", self.scope),
            issued_at=time.time(),
        )
        return self._token

    def auth_header(self) -> dict:
        t = self.get_token()
        return {"Authorization": f"{t.token_type} {t.access_token}"}

最小权限要点：scope 写死成 orders:read shipments:write listings:read，不要申请 finance:read 和 pii:full。

四、Back Market 的“非 OAuth2”正确姿势

# backmarket_auth.py
class BackMarketAuth:
    """
    Back Market 用 Back Office 生成的 opaque token:
    Authorization: Basic <token>
    没有 scope / refresh / expiry。
    对策：当长期凭证 + 人工轮换 + 按区域分 token。
    """

    def __init__(self, opaque_token: str, region: str = "fr"):
        self.token = opaque_token
        self.region = region
        self.base_url = {
            "fr": "https://www.backmarket.fr",
            "com": "https://www.backmarket.com",
            "co.jp": "https://www.backmarket.co.jp",
        }[region]

    def auth_header(self) -> dict:
        return {"Authorization": f"Basic {self.token}"}

    def companion_headers(self) -> dict:
        # Back Market 强制要求
        return {
            "Accept-Language": self.region,
            "User-Agent": "BM-YourCompany-erp-sync",
        }

• 一个 region 一个 token，EU 出事不连累 NA

• token 不进 Git，不进前端

• 换密码/token 有 runbook，避免“账号一改全部渠道掉线”

五、HMAC 签名校验（webhook 防伪造 + 防重放）

# webhook_hmac.py
import hmac, hashlib, time
from typing import Dict

class WebhookVerifier:
    def __init__(self, secret: str, tolerance_sec: int = 300):
        self.secret = secret.encode()
        self.tolerance_sec = tolerance_sec

    def sign(self, raw_body: bytes, ts: int) -> str:
        msg = f"{ts}.".encode() + raw_body
        return hmac.new(self.secret, msg, hashlib.sha256).hexdigest()

    def verify(self, raw_body: bytes, headers: Dict[str, str]) -> bool:
        ts = headers.get("X-Timestamp", "")
        sig = headers.get("X-Signature", "")
        if not ts or not sig:
            return False
        try:
            ts_i = int(ts)
        except ValueError:
            return False

        # 1) 重放防护
        if abs(time.time() - ts_i) > self.tolerance_sec:
            return False

        expected = self.sign(raw_body, ts_i)
        # 2) 常量时间比较，防时序攻击
        return hmac.compare_digest(expected, sig)

verifier = WebhookVerifier(vault.get_hmac_secret("xianyu"))
if not verifier.verify(body, headers):
    return {"code": 403, "msg": "bad signature / replay"}

• SHA-256 起步

• 带 X-Timestamp，超 5 分钟拒

• msg_id 去重（接前篇 IdempotentConsumer）

• 签名错返回 403，但错误信息不吐 secret / payload 原文

六、脱敏层（二手设备特别吃 IMEI / 序列号 / 买家地址）

# masking.py
import re

class PiiMasker:
    @staticmethod
    def phone(v: str) -> str:
        return re.sub(r"(\d{3})\d{4}(\d{4})", r"\1****\2", v) if v else v

    @staticmethod
    def email(v: str) -> str:
        if not v or "@" not in v:
            return v
        name, domain = v.split("@", 1)
        keep = name[:2]
        return f"{keep}***@{domain}"

    @staticmethod
    def id_card(v: str) -> str:
        return v[:6] + "********" + v[-4:] if len(v) >= 10 else "****"

    @staticmethod
    def imei(v: str) -> str:
        # 设备级 PII：日志里只允许后4位
        return "************" + v[-4:] if v and len(v) >= 4 else "****"

    @staticmethod
    def address(v: str) -> str:
        # 门牌号打码，城市/国家保留（用于仓配分析）
        return re.sub(r"(\d{1,4})\s*号|\b\d{1,4}\b\s*(室|栋|单元)", "***\\1", v) if v else v

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
MASK_FIELD_MAP = {
    "buyer_phone": PiiMasker.phone,
    "buyer_email": PiiMasker.email,
    "buyer_id_card": PiiMasker.id_card,
    "device_imei": PiiMasker.imei,
    "device_serial": PiiMasker.imei,
    "buyer_address": PiiMasker.address,
}

def mask_obj(obj: dict) -> dict:
    out = dict(obj)
    for k, fn in MASK_FIELD_MAP.items():
        if k in out and out[k]:
            out[k] = fn(out[k])
    return out

• 落库可加密存明文（走 KMS / 字段级加密），但日志、消息总线、BI 导出必须脱敏

• IMEI / 序列号：客服界面可查，分析库只留哈希

• 退款原因、买家姓名：姓名保留首尾，其余掩码

七、最小权限：服务账号 + RBAC + 作用域

# rbac.py
from enum import Enum

class Scope(str, Enum):
    ORDER_READ = "order:read"
    ORDER_SHIP = "order:ship"
    LISTING_READ = "listing:read"
    LISTING_WRITE = "listing:write"
    INVENTORY_READ = "inventory:read"
    INVENTORY_WRITE = "inventory:write"
    REFUND_READ = "refund:read"
    PII_READ = "pii:read"
    FINANCE_READ = "finance:read"

# 适配器只允许拿自己那点权限
SERVICE_ACCOUNTS = {
    "inventory-sync": {Scope.INVENTORY_READ, Scope.INVENTORY_WRITE, Scope.LISTING_READ},
    "order-consumer": {Scope.ORDER_READ, Scope.ORDER_SHIP, Scope.REFUND_READ},
    "cs-console": {Scope.ORDER_READ, Scope.REFUND_READ, Scope.PII_READ},
    "finance-recon": {Scope.ORDER_READ, Scope.REFUND_READ, Scope.FINANCE_READ},
}

def can(service: str, scope: Scope) -> bool:
    return scope in SERVICE_ACCOUNTS.get(service, set())

• 库存 worker 不能读 finance

• 客服后台 不能调 shipment:write

• 新平台接入默认 0 scope，按“要先跑通哪条链路”追加

• 所有跨域访问走 audit log：谁、哪个服务、哪个 SKU、为什么

八、把四件事串进消息消费链路

# compliant_consumer.py
def consume_inbound(platform: str, headers: dict, raw: bytes, payload: dict):
    # 1) 传输层：TLS 外面已经保住
    # 2) Webhook 验签
    verifier = WebhookVerifier(vault.get_hmac_secret(platform))
    if platform in ("xianyu", "vinted", "ebay"):
        if not verifier.verify(raw, headers):
            return {"code": 403, "msg": "unauthorized"}

    # 3) 取调用凭据（OAuth2 / opaque token）
    if platform == "backmarket":
        auth = BackMarketAuth(vault.get_opaque_token("backmarket")).auth_header()
    else:
        auth = OAuth2Client(**vault.get_oauth_client(platform)).auth_header()

    # 4) 业务处理前脱敏日志
    safe_payload = mask_obj(payload)
    log.info("inbound", extra={"platform": platform, "payload": safe_payload})

    # 5) 统一幂等消费（前篇）
    return unified_consumer.handle(PlatformMessage(
        msg_id=headers.get("X-Message-Id", payload.get("id")),
        platform=platform,
        payload=payload,
    ))

九、合规检查清单（上线前过一遍）

• [ ] 没有 client_secret / opaque token 进代码仓库

• [ ] OAuth2 token 有过期、有 scope、有自动刷新

• [ ] Back Market token 按 region 隔离、有轮换 runbook

• [ ] 所有 webhook 有 HMAC + timestamp + nonce/msg_id 去重

• [ ] 手机号/邮箱/IMEI/地址在日志和消息总线里是脱敏的

• [ ] 服务账号按“只够跑通业务”授权，不堆 admin

• [ ] 401/403/429 有告警，但不返回内部错误细节

• [ ] PII 字段级加密（KMS/AES-256），分析库用 tokenization

• [ ] 审计日志：msg_id / platform / sku / order_id / actor / scope

十、一句话收口

数据合规前置 =

鉴权按平台能力分层（OAuth2 / opaque token / mTLS）

• 入站消息全部 HMAC 验签防伪造防重放

• PII 在日志/总线/BI 层先脱敏再流动

• 每个适配器只用够用的那几个 scope

二手 ERP 做跨境，合规不是上线后补的工单，是 consume() 第一行就挡人的东西。