diff --git a/cloud/chv/approval_channel.py b/cloud/chv/approval_channel.py index 575b321..0e1645c 100644 --- a/cloud/chv/approval_channel.py +++ b/cloud/chv/approval_channel.py @@ -11,6 +11,9 @@ "reasons": [str, ...], "expires_at": float} ``target`` 只有目标描述（连接器/方法/scheme/主机/端口/路径）， ``reasons`` 只有原因名；载荷里绝无凭证值与用户数据内容。 + P10 扩展：可选 ``details`` 字段（支付审批的确切购买详情， + 商家/商品/金额/货币，供客户端 UI 展示；不传时 payload 与 + P5 完全一致）。 ``nonce`` 是随机 16 字节 hex（32 字符），每次请求唯一。 ``expires_at`` 默认请求后 15 分钟（900 秒），可注入时钟。 - 答复（``Answer``，客户端答完直接回 Sentinel）:: @@ -72,38 +75,50 @@ class ApprovalRequest: reasons: Iterable[str] | None, nonce: str | None = None, now: float | None = None, - ttl_seconds: int = DEFAULT_TTL_SECONDS) -> None: + ttl_seconds: int = DEFAULT_TTL_SECONDS, + details: dict | None = None) -> None: self.pending_id = pending_id self.target = dict(target) self.reasons = list(reasons or []) self.nonce = nonce or secrets.token_hex(16) at = now if now is not None else time.time() self.expires_at = float(at) + ttl_seconds + # 可选的确切详情（P10 支付审批用：商家、商品、金额、 + # 货币——供客户端 UI 展示；无 details 时 payload 与 + # P5 完全一致）。只含描述字段，无凭证值。 + self.details = dict(details) if details is not None else None def payload(self) -> dict: """可外发给客户端的请求载荷（只含描述字段 + nonce + - 过期，无密钥、无凭证值、无用户数据内容）。""" - return {"pending_id": self.pending_id, "nonce": self.nonce, - "target": dict(self.target), - "reasons": list(self.reasons), - "expires_at": self.expires_at} + 过期，无密钥、无凭证值、无用户数据内容）。有 details + 时带上（支付审批的确切购买详情）。""" + out: dict = {"pending_id": self.pending_id, "nonce": self.nonce, + "target": dict(self.target), + "reasons": list(self.reasons), + "expires_at": self.expires_at} + if self.details is not None: + out["details"] = dict(self.details) + return out def request_approval(queue_record: dict, secret: str, now: float | None = None, - ttl_seconds: int = DEFAULT_TTL_SECONDS + ttl_seconds: int = DEFAULT_TTL_SECONDS, + details: dict | None = None ) -> ApprovalRequest: """为一条队列记录生成通道请求并登记（nonce + 过期）。 ``secret`` 是调用方持有的通道密钥：本模块不保存它、不进 载荷、不进日志；答复方用同一密钥签名，校验时对上才算数。 + ``details`` 可选（P10 支付审批的确切购买详情，进 payload + 供客户端 UI 展示；不传时行为与 P5 完全一致）。 返回的 ``ApprovalRequest.payload()`` 即发给客户端的东西。 """ pending_id = queue_record["pending_id"] req = ApprovalRequest( pending_id, _target_from_record(queue_record), queue_record.get("reasons", []), - now=now, ttl_seconds=ttl_seconds) + now=now, ttl_seconds=ttl_seconds, details=details) _REQUESTS[pending_id] = {"nonce": req.nonce, "expires_at": req.expires_at} return req diff --git a/cloud/chv/checkout_detect.py b/cloud/chv/checkout_detect.py new file mode 100644 index 0000000..7eab362 --- /dev/null +++ b/cloud/chv/checkout_detect.py @@ -0,0 +1,143 @@ +"""结账页检测（feat-chv-muse-p10，叶子模块，纯标准库）。 + +对齐 Muse 官方安全文（security.muse.ai 2026-09-08，见 +docs/chv-muse-p10-gap.md）：已保存支付凭证的网站，浏览器内置 +结账页检测，每次都弹出人工审批并展示确切购买详情。 + +本模块是纯函数，不启动浏览器、不联网、不读原始 DOM： +- 输入只有 URL + 无障碍树节点（与 P4 的 ``detect_form_risk`` + 同输入形状——P2 的浏览器子代理只返回无障碍树快照，节点 + 只有 role/name/label/type 等描述字段）； +- 输出 ``CheckoutVerdict``：是否结账页、置信度、命中原因、 + 表单风险、页面金额文本； +- 确定性规则，不冒充模型分类器（见 P6 口径）。 + +诚实边界：启发式检测，不是 100% 检出——商户自定义结账页、 +单页应用内嵌结账可能漏检。漏检时的安全底线在调用方 +（``PaymentGate`` 拒绝任何未经审批的扣款），不在本模块 +宣称“所有结账页都会被拦下”。 + +页面金额文本（``amount_text``）只作展示信号：扣款金额永远 +以 ``PurchaseDetails``（订单数据）为准，页面文本是不受信任 +输入（P6 口径），不得直接拿去扣款。 +""" + +import re +from dataclasses import dataclass, field +from urllib.parse import urlparse + +# URL 路径段信号：结账步骤（高）与购物车（中，结账前一步）。 +_CHECKOUT_SEGMENTS = ("checkout", "check-out", "payment", "pay", + "billing", "order-review", "place-order", + "confirm-order") +_CART_SEGMENTS = ("cart", "basket", "bag") + +# 无障碍树节点文本信号（小写子串匹配）。 +_PAYMENT_HINTS = ("card number", "credit card", "debit card", + "cardholder", "expiry", "expiration", "exp date", + "cvc", "cvv", "security code", "卡号", "信用卡", + "有效期", "安全码") +_ORDER_HINTS = ("place order", "pay now", "confirm purchase", + "complete purchase", "checkout", "下单", "立即支付", + "确认购买", "去支付", "提交订单") +_PASSWORD_HINTS = ("password", "密码") + +_AMOUNT_RE = re.compile( + r"[$€£¥]\s?\d[\d,]*(?:\.\d{1,2})?" + r"|\b\d[\d,]*(?:\.\d{1,2})?\s?(?:USD|EUR|GBP|JPY|CNY)\b") + + +@dataclass +class CheckoutVerdict: + """一次结账检测的结果（只含描述，不含任何凭证值）。""" + + is_checkout: bool + confidence: str # high / medium / low + reasons: list = field(default_factory=list) + form_risk: str | None = None # payment / password / None + amount_text: str | None = None + + +def _node_text(node: dict) -> str: + """一个无障碍树节点的描述文本（role/name/label/type + 拼起来，小写；原始 DOM 从不到这里）。""" + parts = [str(node.get(k, "")) for k in + ("role", "name", "label", "type")] + return " ".join(parts).lower() + + +def _url_level(url: str) -> str: + """URL 信号档位：high（结账路径）/ medium（购物车）/ + none。只看 path，不看 query（query 里可能有订单号等 + 敏感值，不进原因）。""" + try: + path = urlparse(url).path.lower() + except ValueError: + return "none" + segments = [s for s in path.split("/") if s] + if any(s in _CHECKOUT_SEGMENTS for s in segments): + return "high" + if any(s in _CART_SEGMENTS for s in segments): + return "medium" + return "none" + + +def _node_signals(nodes: list) -> tuple: + """无障碍树信号：(支付字段, 密码字段, 下单按钮, 金额文本)。 + 节点文本只作特征匹配，不存原文。""" + payment = False + password = False + order = False + amount: str | None = None + for node in nodes or []: + text = _node_text(node) + if not text.strip(): + continue + if any(h in text for h in _PAYMENT_HINTS): + payment = True + if any(h in text for h in _PASSWORD_HINTS): + password = True + if any(h in text for h in _ORDER_HINTS): + order = True + if amount is None: + hit = _AMOUNT_RE.search(text) + if hit: + amount = hit.group(0) + return payment, password, order, amount + + +def detect_checkout(url: str, nodes: list | None = None + ) -> CheckoutVerdict: + """检测一个页面是否为结账页。 + + 判定（确定性规则）：支付字段在页 → 结账页（高置信）； + 结账路径 URL → 结账页（有下单按钮/金额则高，否则中）； + 下单按钮 + 金额文本 → 结账页（中置信）；只命中购物车 + 路径 → 不是结账页（购物车是结账前一步）；其余 → 不是。 + ``form_risk``：支付字段在页为 ``payment``，否则密码字 + 段在页为 ``password``，与 P4 的 ``detect_form_risk`` + 口径一致。""" + level = _url_level(url) + has_payment, has_password, has_order, amount = _node_signals( + nodes or []) + reasons: list = [] + if level == "high": + reasons.append("checkout_url") + elif level == "medium": + reasons.append("cart_url") + if has_payment: + reasons.append("payment_fields") + if has_order: + reasons.append("order_button") + if amount is not None: + reasons.append("amount_text") + risk = "payment" if has_payment else ( + "password" if has_password else None) + if has_payment: + return CheckoutVerdict(True, "high", reasons, risk, amount) + if level == "high": + conf = "high" if has_order or amount else "medium" + return CheckoutVerdict(True, conf, reasons, risk, amount) + if has_order and amount is not None: + return CheckoutVerdict(True, "medium", reasons, risk, amount) + return CheckoutVerdict(False, "low", reasons, risk, amount) diff --git a/cloud/chv/payments.py b/cloud/chv/payments.py new file mode 100644 index 0000000..2525956 --- /dev/null +++ b/cloud/chv/payments.py @@ -0,0 +1,769 @@ +"""支付安全门（feat-chv-muse-p10，叶子模块，纯标准库）。 + +对齐 Muse 官方安全文（security.muse.ai 2026-09-08，见 +docs/chv-muse-p10-gap.md）：支付时发行一次性卡号给商家而非 +真实卡号，绑定特定商家、特定金额、有限有效期；已保存支付 +凭证的网站每次购买都弹人工审批并展示确切购买详情；未购物 +过的网站走内置合规钱包（Stripe Link 首发、Shop Pay 次之）。 + +本模块只做客内 agent 侧的支付安全门（检测、路由、限额、 +审批、绑定、审计），不产生真实扣款： +- ``PaymentCredentialVault``：已保存支付方式的隔离存储， + 走 P3 authd 模式（只活内存、白名单、按次释放绑定、审计 + 绝无凭证值），但另立库——只存不透明支付方式 id + 掩码 + 标签，原始卡号 PAN 拒绝入库； +- ``OneTimeCard``：一次性卡号的绑定与兑现校验（商家 + + 金额 + 货币 + 有效期，单次有效）； +- ``SimulatedWalletBackend``：沙箱钱包（假 PAN 只在后端 + 私有内存里，调用方只见掩码标签与 card_id；``charge`` + 只记账不联网），供测试与演示；真实钱包集成是待用户 + 另行批准的事项，见差距表 §3； +- ``PaymentGate``：编排——结账检测 → 路由 → 限额 → + P5 对话外审批（确切购买详情进 payload）→ 铸一次性卡 + → 兑现校验 → 记账 → 审计。无审批的支付一律拒绝。 + +真实一次性虚拟卡由 Link 在其侧发行（连接器文档原文）， +本轨的卡是绑定与兑现校验的等价模型，不是发卡行。 +""" + +import hashlib +import json +import secrets +import time +from collections.abc import Callable +from dataclasses import dataclass + +from cloud.chv.approval_channel import ( + Answer, + request_approval, + resolve_via_channel, + verify_answer, +) +from cloud.chv.checkout_detect import CheckoutVerdict +from cloud.chv.sentinel import ApprovalQueue, EgressRequest + +# 钱包提供方白名单（连接器文档现状：stripe-link 首发、 +# shop-pay 已可用；未知提供方一律拒绝）。 +ALLOWED_WALLET_PROVIDERS = frozenset({"stripe-link", "shop-pay"}) + +# 一次性卡默认有效期（秒）与按次释放默认有效期（秒）。 +DEFAULT_CARD_TTL_SECONDS = 900 +DEFAULT_RELEASE_TTL_SECONDS = 300 + +# 审计里绝不出现的字段名（命中即丢弃并标 redacted）。 +_SECRET_FIELD_NAMES = ("card_number", "cardnumber", "pan", "cvv", + "cvc", "token", "secret", "password") + +# 沙箱钱包的假卡号段（Stripe 测试卡号段，永不产生真实扣款）。 +_TEST_PAN_PREFIX = "400000000000" + + +class PaymentRefusal(Exception): + """支付拒绝：kind + 原因，不含任何卡号与凭证值。""" + + def __init__(self, kind: str, message: str) -> None: + super().__init__("payment_refused:%s: %s" % (kind, message)) + self.kind = kind + + +def luhn_valid(number: str) -> bool: + """Luhn 校验（只用于识别“像卡号”的数字串，拒绝入库）。""" + digits = [int(c) for c in number if c.isdigit()] + if len(digits) < 13 or len(digits) > 19: + return False + total = 0 + for i, d in enumerate(reversed(digits)): + if i % 2 == 1: + d *= 2 + if d > 9: + d -= 9 + total += d + return total % 10 == 0 + + +def _looks_like_pan(value: str) -> bool: + """值是否像原始卡号：去空格/横线后 13–19 位数字且过 + Luhn。命中即视为 PAN，拒绝入库、拒绝进审计。""" + compact = value.replace(" ", "").replace("-", "") + return compact.isdigit() and luhn_valid(compact) + + +def _masked_label(pan: str, brand: str = "Visa") -> str: + """假卡号的掩码标签（只露后 4 位，审计与审批里唯一 + 允许出现的卡号派生形式）。""" + return "%s ....%s" % (brand, pan[-4:]) + + +@dataclass +class PaymentMethod: + """一个已保存的支付方式（只含不透明 id + 掩码标签， + 无真实卡号）。``merchant_host`` 是该凭证保存于的商家 + 主机（None = 未限定商家，释放时仍须绑定本次商家）。""" + + provider: str + method_id: str + masked_label: str + merchant_host: str | None = None + + def describe(self) -> dict: + """审计/审批用描述（不含任何凭证值）。""" + return {"provider": self.provider, "method_id": self.method_id, + "masked_label": self.masked_label, + "merchant_host": self.merchant_host} + + +@dataclass +class PaymentRelease: + """一次支付方式按次释放（替身）：绑定到一次具体的购买。 + 兑现后交出的是不透明 method_id + 掩码标签，不是卡号。""" + + release_id: str + provider: str + method_id: str + masked_label: str + merchant: str + amount_minor: int + currency: str + expires_at: float + uses_left: int = 1 + + def describe(self) -> dict: + """审计用描述（只含绑定，不含值）。""" + return {"release_id": self.release_id, + "provider": self.provider, "method_id": self.method_id, + "masked_label": self.masked_label, + "merchant": self.merchant, + "amount_minor": self.amount_minor, + "currency": self.currency, + "expires_at": self.expires_at, + "uses_left": self.uses_left} + + def __repr__(self) -> str: # 绝不含任何凭证值 + return ("PaymentRelease(id=%s provider=%s merchant=%s " + "amount=%d %s)" % (self.release_id, self.provider, + self.merchant, self.amount_minor, + self.currency)) + + +class PaymentCredentialVault: + """只活内存的支付凭证保险库（一个 VM 一个实例）。 + + - 值只存在内存 dict 里，不落盘、不进日志； + - 只存不透明支付方式 id + 掩码标签：原始卡号 PAN、 + 未知提供方、空 id 一律拒绝入库； + - ``release()`` 按次发放：绑定 {商家, 金额, 货币}， + 单次有效、ttl 后过期； + - ``consume()`` 单次兑现：兑现一次即删除，过期也兑现 + 不了；兑现失败不留可重试的残值。 + clock 可注入（测试用假时钟测过期），默认 time.time。""" + + def __init__(self, clock: Callable[[], float] | None = None) -> None: + self._store: dict[tuple[str, str], PaymentMethod] = {} + self._releases: dict[str, PaymentRelease] = {} + self._clock = clock or time.time + + def __repr__(self) -> str: # 绝不含任何凭证值 + return "PaymentCredentialVault(providers=%s)" % sorted( + {p for (p, _m) in self._store}) + + def store(self, provider: str, method_id: str, + masked_label: str, merchant_host: str | None = None + ) -> None: + """入库：白名单校验 + PAN 拒绝。原始卡号、未知提供方、 + 空 id、空标签一律拒绝入库（PaymentRefusal），绝不静默 + 收下。""" + if provider not in ALLOWED_WALLET_PROVIDERS: + raise PaymentRefusal( + "unknown_provider", + "钱包提供方 %s 不在白名单内" % provider) + if not isinstance(method_id, str) or not method_id.strip(): + raise PaymentRefusal("empty_method_id", "支付方式 id 为空") + if _looks_like_pan(method_id) or _looks_like_pan(masked_label): + raise PaymentRefusal( + "raw_pan_refused", + "原始卡号不得入库，只存不透明 id 与掩码标签") + if not isinstance(masked_label, str) or not masked_label.strip(): + raise PaymentRefusal("empty_label", "掩码标签为空") + key = (provider, method_id) + self._store[key] = PaymentMethod( + provider, method_id, masked_label, merchant_host) + + def has(self, provider: str, method_id: str) -> bool: + """该（提供方，方式 id）是否已入库（不返回值）。""" + return (provider, method_id) in self._store + + def methods_for_merchant(self, merchant_host: str) -> list: + """该商家主机上已保存的支付方式（只返回描述对象， + 不含任何凭证值）。""" + return [m for m in self._store.values() + if m.merchant_host is None + or m.merchant_host == merchant_host] + + def all_methods(self) -> list: + """全部已保存支付方式（描述对象，不含任何凭证值）。""" + return list(self._store.values()) + + def forget(self, provider: str, method_id: str) -> bool: + """删一条入库支付方式；同时作废其全部未兑现释放。 + 未知键返回 False，不抛错。""" + gone = self._store.pop((provider, method_id), None) is not None + if gone: + doomed = [rid for rid, rel in self._releases.items() + if rel.provider == provider + and rel.method_id == method_id] + for rid in doomed: + self._releases.pop(rid, None) + return gone + + def release(self, provider: str, method_id: str, merchant: str, + amount_minor: int, currency: str, + ttl_seconds: int = DEFAULT_RELEASE_TTL_SECONDS, + max_uses: int = 1) -> PaymentRelease: + """按次发放：绑定须与入库作用域一致（merchant_host + 有记录时须一致），金额须 > 0。发放的释放单次有效、 + ttl 后过期；须经 consume() 兑现一次才交出方式引用。""" + method = self._store.get((provider, method_id)) + if method is None: + raise PaymentRefusal( + "method_not_stored", + "支付方式 %s 未入库" % method_id) + if method.merchant_host is not None \ + and method.merchant_host != merchant: + raise PaymentRefusal( + "scope_mismatch", "支付方式作用域商家不一致，拒绝释放") + if amount_minor <= 0: + raise PaymentRefusal("bad_amount", "金额须大于 0") + release = PaymentRelease( + release_id=secrets.token_hex(8), + provider=provider, method_id=method_id, + masked_label=method.masked_label, merchant=merchant, + amount_minor=amount_minor, currency=currency, + expires_at=self._clock() + ttl_seconds, + uses_left=max_uses) + self._releases[release.release_id] = release + return release + + def consume(self, release_id: str) -> PaymentMethod: + """兑现一次释放：过期/用完/未知 id 一律拒绝 + （PaymentRefusal，不含值）。兑现成功返回支付方式 + 引用（不透明 id + 掩码标签），并把该释放删除。""" + release = self._releases.get(release_id) + if release is None: + raise PaymentRefusal( + "release_unknown", "释放记录不存在或已兑现") + if self._clock() >= release.expires_at: + self._releases.pop(release_id, None) + raise PaymentRefusal("release_expired", "释放已过期") + if release.uses_left <= 0: + self._releases.pop(release_id, None) + raise PaymentRefusal("release_exhausted", "释放已用完") + release.uses_left -= 1 + method = self._store.get((release.provider, release.method_id)) + self._releases.pop(release_id, None) + if method is None: + raise PaymentRefusal( + "method_gone", "支付方式已删除，释放作废") + return method + + +@dataclass +class PurchaseDetails: + """一次购买的确切详情（审批单上展示给用户的全部内容， + 也是防篡改摘要的输入）。金额以 minor units 计（分/日元 + 等无小数货币按其最小单位），货币为三位字母码，不换算。""" + + merchant: str + merchant_host: str + items: tuple + amount_minor: int + currency: str + wallet_provider: str + + def total_minor(self) -> int: + """商品小计（数量 × 单价之和），须等于 amount_minor。""" + return sum(qty * unit for (_n, qty, unit) in self.items) + + def validate(self) -> None: + """校验详情自洽：金额 > 0、货币三位码、商品非空、 + 小计等于总额、提供方在白名单内。任一失败即 + ValueError（调用方 bug，不走支付拒绝通道）。""" + if self.amount_minor <= 0: + raise ValueError("amount_minor must be > 0") + if len(self.currency) != 3 or not self.currency.isalpha(): + raise ValueError("currency must be a 3-letter code") + if not self.items: + raise ValueError("items must not be empty") + if self.total_minor() != self.amount_minor: + raise ValueError("items total != amount_minor") + if self.wallet_provider not in ALLOWED_WALLET_PROVIDERS: + raise ValueError("unknown wallet_provider") + + def describe(self) -> dict: + """审批 payload 里的确切购买详情（商家、商品、数量、 + 金额、货币、提供方；无任何卡号与凭证值）。""" + return {"merchant": self.merchant, + "merchant_host": self.merchant_host, + "items": [{"name": n, "qty": q, "unit_minor": u} + for (n, q, u) in self.items], + "amount_minor": self.amount_minor, + "currency": self.currency, + "wallet_provider": self.wallet_provider} + + def digest(self) -> str: + """购买详情摘要（SHA-256，规范 JSON）：审批时记下， + 扣款时重算——详情被篡改（换商家/改金额/换商品）即 + 摘要对不上，支付拒绝。""" + canonical = json.dumps(self.describe(), sort_keys=True, + separators=(",", ":"), + ensure_ascii=False) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +@dataclass +class OneTimeCard: + """一张一次性卡号（绑定与兑现校验的等价模型）。 + + 绑定 {商家, 金额, 货币, 有效期}，单次有效。真实一次 + 性虚拟卡由 Link 在其侧发行（连接器文档原文），本类 + 只做绑定校验——调用方只见 card_id 与掩码标签，卡号 + 本体只活在钱包后端私有内存里。""" + + card_id: str + merchant: str + amount_minor: int + currency: str + expires_at: float + masked_label: str + uses_left: int = 1 + + def describe(self) -> dict: + """审计用描述（只含绑定与掩码标签，不含卡号）。""" + return {"card_id": self.card_id, "merchant": self.merchant, + "amount_minor": self.amount_minor, + "currency": self.currency, + "expires_at": self.expires_at, + "masked_label": self.masked_label, + "uses_left": self.uses_left} + + def __repr__(self) -> str: # 绝不含卡号 + return ("OneTimeCard(id=%s merchant=%s amount=%d %s)" + % (self.card_id, self.merchant, self.amount_minor, + self.currency)) + + def authorize(self, merchant: str, amount_minor: int, + currency: str, now: float) -> tuple: + """兑现校验：用完/过期/换商家/超额/换货币任一命中 + 即 (False, 原因)；全过则消耗一次并返回 (True, + "ok")。单次有效，用过即废。""" + if self.uses_left <= 0: + return False, "card_exhausted" + if now >= self.expires_at: + return False, "card_expired" + if merchant != self.merchant: + return False, "wrong_merchant" + if currency != self.currency: + return False, "wrong_currency" + if amount_minor > self.amount_minor: + return False, "amount_exceeded" + self.uses_left -= 1 + return True, "ok" + + +@dataclass +class SpendLimits: + """钱包侧限额（minor units，无货币、不换算——按 + stripe-link 连接器文档口径）。None = 该档未设置。 + ``daily_spent_minor`` / ``thirty_day_spent_minor`` 是 + 调用方传入的已用额度（本模块不记账）。""" + + per_purchase_minor: int | None = None + daily_minor: int | None = None + thirty_day_minor: int | None = None + daily_spent_minor: int = 0 + thirty_day_spent_minor: int = 0 + + def check(self, amount_minor: int) -> tuple: + """逐项比 minor units：超任一档即 (False, 原因)； + 三档全为 None（限额不可用）→ (True, + "limits_unavailable")，绝不编造上限。""" + if self.per_purchase_minor is not None \ + and amount_minor > self.per_purchase_minor: + return False, "over_per_purchase_limit" + if self.daily_minor is not None \ + and self.daily_spent_minor + amount_minor \ + > self.daily_minor: + return False, "over_daily_limit" + if self.thirty_day_minor is not None \ + and self.thirty_day_spent_minor + amount_minor \ + > self.thirty_day_minor: + return False, "over_thirty_day_limit" + if self.per_purchase_minor is None \ + and self.daily_minor is None \ + and self.thirty_day_minor is None: + return True, "limits_unavailable" + return True, "ok" + + +class PaymentAudit: + """支付审计（内存态记录表）。 + + 事件只含允许字段（商家、金额、货币、掩码标签、 + pending_id、原因、路由、提供方）；字段名命中 + ``_SECRET_FIELD_NAMES`` 一律丢弃并标 ``redacted``—— + 卡号与凭证值绝不进审计。""" + + ALLOWED_FIELDS = frozenset({ + "merchant", "merchant_host", "amount_minor", "currency", + "masked_label", "pending_id", "reason", "route", + "wallet_provider", "card_id", "limits_checked", + }) + + def __init__(self, clock: Callable[[], float] | None = None) -> None: + self._records: list = [] + self._clock = clock or time.time + + def record(self, event: str, fields: dict | None = None, + now: float | None = None) -> dict: + """记一条审计（只留允许字段；秘密字段名丢弃并标 + redacted）。返回记下的记录副本。""" + at = now if now is not None else self._clock() + kept: dict = {"event": event, "ts": at} + redacted = False + for key, value in (fields or {}).items(): + if key in _SECRET_FIELD_NAMES: + redacted = True + continue + if key in self.ALLOWED_FIELDS: + kept[key] = value + if redacted: + kept["redacted"] = True + self._records.append(dict(kept)) + return dict(kept) + + def list_records(self) -> list: + """全部审计记录（副本，按记录时间排序）。""" + return [dict(r) for r in self._records] + + +class SimulatedWalletBackend: + """沙箱钱包（测试与演示用，绝不产生真实扣款）。 + + - 假 PAN 只存在 ``self._pans`` 私有内存里，调用方只见 + card_id 与掩码标签；``__repr__`` 与账本里绝无卡号； + - ``mint_card`` 按购买详情铸一次性卡（绑定商家 + 金额 + + 货币 + 900 秒有效期）； + - ``charge`` 只记账不联网：绑定校验通过即记一笔 + ``simulated_paid``，否则记拒绝原因； + - ``void_card`` 作废一张卡（取消待支付时调用）。 + 真实钱包集成是待用户另行批准的事项（见差距表 §3）。""" + + def __init__(self, providers: tuple = ("stripe-link", "shop-pay"), + clock: Callable[[], float] | None = None) -> None: + self._providers = tuple(providers) + self._vault = PaymentCredentialVault(clock=clock) + self._pans: dict[str, str] = {} + self._cards: dict[str, OneTimeCard] = {} + self._voided: set = set() + self._ledger: list = [] + self._clock = clock or time.time + self.fail_next_charge = False + + def __repr__(self) -> str: # 绝不含任何卡号 + return "SimulatedWalletBackend(providers=%s)" % list( + self._providers) + + def supports(self, provider: str) -> bool: + """该钱包提供方是否可用（沙箱里由构造参数决定， + 真实侧由钱包连接状态决定）。""" + return provider in self._providers + + def add_method(self, provider: str, method_id: str, + masked_label: str, merchant_host: str | None = None + ) -> None: + """往沙箱钱包里加一个已保存支付方式（只存不透明 + id + 掩码标签，走同一入库校验）。""" + self._vault.store(provider, method_id, masked_label, + merchant_host) + + def list_methods(self, merchant_host: str | None = None) -> list: + """已保存支付方式（描述对象；merchant_host 给了 + 则只返回该商家的）。""" + if merchant_host is None: + return [m.describe() + for m in self._vault.all_methods()] + return [m.describe() for m in + self._vault.methods_for_merchant(merchant_host)] + + def mint_card(self, details: PurchaseDetails, + now: float | None = None) -> OneTimeCard: + """按购买详情铸一张一次性卡：假 PAN 只进私有内存， + 返回的卡对象只有 card_id 与掩码标签。""" + at = now if now is not None else self._clock() + card_id = "otc_" + secrets.token_hex(8) + pan = _TEST_PAN_PREFIX + "%04d" % secrets.randbelow(10000) + self._pans[card_id] = pan + card = OneTimeCard( + card_id=card_id, merchant=details.merchant, + amount_minor=details.amount_minor, + currency=details.currency, + expires_at=at + DEFAULT_CARD_TTL_SECONDS, + masked_label=_masked_label(pan)) + self._cards[card_id] = card + return card + + def charge(self, card: OneTimeCard, details: PurchaseDetails, + now: float | None = None) -> dict: + """模拟扣款：只记账不联网。绑定校验通过即记一笔 + simulated_paid；``fail_next_charge`` 为真时模拟 + 钱包侧失败（扣款失败即 void 该卡）。""" + at = now if now is not None else self._clock() + entry: dict + if self.fail_next_charge: + self.fail_next_charge = False + self.void_card(card.card_id) + entry = {"status": "simulated_failed", + "reason": "wallet_declined", + "card_id": card.card_id} + self._ledger.append(dict(entry)) + return dict(entry) + ok, reason = card.authorize( + details.merchant, details.amount_minor, + details.currency, at) + if not ok: + entry = {"status": "refused", "reason": reason, + "card_id": card.card_id} + self._ledger.append(dict(entry)) + return dict(entry) + entry = {"status": "simulated_paid", + "card_id": card.card_id, + "merchant": details.merchant, + "amount_minor": details.amount_minor, + "currency": details.currency, + "masked_label": card.masked_label} + self._ledger.append(dict(entry)) + return dict(entry) + + def void_card(self, card_id: str) -> bool: + """作废一张一次性卡（取消待支付时调用；已作废或 + 未知 id 返回 False，不抛错）。""" + if card_id in self._voided or card_id not in self._cards: + return False + self._voided.add(card_id) + card = self._cards[card_id] + card.uses_left = 0 + return True + + def ledger(self) -> list: + """模拟账本（副本；只含记账条目，无任何卡号）。""" + return [dict(e) for e in self._ledger] + + +def _verdict_of(answer: Answer | dict) -> str: + """答复对象或同款 dict 里的 verdict（verify_answer 已 + 校验过合法性，这里只取值；取不到即空串）。""" + if isinstance(answer, dict): + value = answer.get("verdict", "") + return value if isinstance(value, str) else "" + return getattr(answer, "verdict", "") or "" + + +class PaymentGate: + """支付安全门：一次购买的完整编排。 + + ``begin_purchase``：结账检测 → 路由（已保存凭证 / + 合规钱包 stripe-link 优先、shop-pay 次之）→ 限额 → + P5 对话外审批（确切购买详情进 payload）。无审批不支付。 + ``complete_purchase``：验签（不消费）→ 验详情摘要 + （篡改即拒，审批保持待处理）→ 单次消费答复 → 铸一次 + 性卡 → 兑现校验 → 记账 → 审计。审批被拒即停止，不扣款。 + """ + + def __init__(self, queue: ApprovalQueue, + wallet: SimulatedWalletBackend, + vault: PaymentCredentialVault, + channel_secret: str, + clock: Callable[[], float] | None = None, + audit: PaymentAudit | None = None) -> None: + self._queue = queue + self._wallet = wallet + self._vault = vault + self._secret = channel_secret + self._clock = clock or time.time + self._audit = audit or PaymentAudit(clock=clock) + self._pending: dict[str, dict] = {} + + def _route(self, details: PurchaseDetails) -> tuple: + """路由：商家有已保存凭证 → saved；无 → 合规钱包 + （stripe-link 优先、shop-pay 次之）；两家都不支持 + → (None, "no_wallet_route")。""" + saved = self._vault.methods_for_merchant( + details.merchant_host) + if saved: + return "saved", saved[0].provider + if self._wallet.supports("stripe-link"): + return "wallet", "stripe-link" + if self._wallet.supports("shop-pay"): + return "wallet", "shop-pay" + return "", "no_wallet_route" + + def _refuse(self, details: PurchaseDetails, reason: str, + at: float, limits_checked: bool | None = None) -> dict: + """记一条 payment_refused 审计并返回统一的拒绝结果。""" + fields: dict = {"merchant": details.merchant, + "merchant_host": details.merchant_host, + "amount_minor": details.amount_minor, + "currency": details.currency, + "wallet_provider": details.wallet_provider, + "reason": reason} + if limits_checked is not None: + fields["limits_checked"] = limits_checked + self._audit.record("payment_refused", fields, now=at) + out: dict = {"ok": False, "reason": reason} + if limits_checked is not None: + out["limits_checked"] = limits_checked + return out + + def _request_payment_approval(self, details: PurchaseDetails, + route: str, provider: str, + limits_checked: bool, + at: float) -> dict: + """记入 Sentinel 待审批队列并经 P5 通道发审批请求 + （确切购买详情进 payload）。返回 begin 的成功结果。""" + request = EgressRequest( + connector="payment", method="purchase", scheme="https", + host=details.merchant_host, port=443, path="/checkout") + pending_id = self._queue.record(request, ["payment_approval"]) + record = next(r for r in self._queue.list_pending() + if r["pending_id"] == pending_id) + approval = request_approval( + record, self._secret, now=at, + details=details.describe()) + self._pending[pending_id] = { + "details": details, "digest": details.digest(), + "route": route, "provider": provider, + "card_id": None, "limits_checked": limits_checked} + self._audit.record( + "payment_approval_requested", + {"merchant": details.merchant, + "merchant_host": details.merchant_host, + "amount_minor": details.amount_minor, + "currency": details.currency, + "wallet_provider": details.wallet_provider, + "pending_id": pending_id, "route": route, + "limits_checked": limits_checked}, now=at) + return {"ok": True, "pending_id": pending_id, + "payload": approval.payload(), "route": route, + "provider": provider} + + def begin_purchase(self, details: PurchaseDetails, + checkout: CheckoutVerdict, + limits: SpendLimits | None = None, + now: float | None = None) -> dict: + """开始一次购买：返回 {"ok": True, "pending_id", + "payload", "route", "provider"}，或 {"ok": False, + "reason"}。审批请求发给客户端 UI，payload 的 details + 就是用户看到的确切购买详情。""" + details.validate() + at = now if now is not None else self._clock() + if not checkout.is_checkout: + return self._refuse(details, "not_checkout", at) + route, provider = self._route(details) + if not route: + return self._refuse(details, provider, at) + limits_checked = False + if limits is not None: + ok, reason = limits.check(details.amount_minor) + if not ok: + return self._refuse(details, reason, at, True) + limits_checked = reason != "limits_unavailable" + return self._request_payment_approval( + details, route, provider, limits_checked, at) + + def _charge_after_approval(self, pending_id: str, + details: PurchaseDetails, + base: dict, at: float) -> dict: + """审批通过后：铸一次性卡 → 兑现校验 → 记账。扣款 + 失败即 {"ok": False, "reason"}，不扣款；成功即 + {"ok": True, "charge", "card_id", "masked_label"}。""" + card = self._wallet.mint_card(details, now=at) + self._pending[pending_id]["card_id"] = card.card_id + result = self._wallet.charge(card, details, now=at) + if result.get("status") != "simulated_paid": + self._audit.record( + "payment_refused", + dict(base, reason=result.get("reason", "charge_failed"), + card_id=card.card_id), now=at) + self._pending.pop(pending_id, None) + return {"ok": False, + "reason": result.get("reason", "charge_failed")} + self._audit.record( + "payment_completed", + dict(base, card_id=card.card_id, + masked_label=card.masked_label), now=at) + self._pending.pop(pending_id, None) + return {"ok": True, "charge": result, + "card_id": card.card_id, + "masked_label": card.masked_label} + + def complete_purchase(self, pending_id: str, + answer: Answer | dict, + actual_details: PurchaseDetails, + now: float | None = None) -> dict: + """完成一次购买：验签 → 验详情摘要 → 单次消费答复 + → 铸一次性卡 → 兑现校验 → 记账。任一步失败即 + {"ok": False, "reason"}，不扣款。""" + pending = self._pending.get(pending_id) + if pending is None: + return {"ok": False, "reason": "unknown_pending"} + at = now if now is not None else self._clock() + details = pending["details"] + base = {"merchant": details.merchant, + "amount_minor": details.amount_minor, + "currency": details.currency, + "pending_id": pending_id} + if not verify_answer(answer, self._secret, now=at): + self._audit.record("payment_refused", + dict(base, reason="invalid_answer"), + now=at) + return {"ok": False, "reason": "invalid_answer"} + if actual_details.digest() != pending["digest"]: + self._audit.record("payment_refused", + dict(base, reason="details_tampered"), + now=at) + return {"ok": False, "reason": "details_tampered"} + if not resolve_via_channel(self._queue, answer, self._secret, + now=at): + return {"ok": False, "reason": "resolve_failed"} + if _verdict_of(answer) == "deny": + self._audit.record("payment_denied", base, now=at) + self._pending.pop(pending_id, None) + return {"ok": False, "reason": "denied"} + return self._charge_after_approval(pending_id, details, + base, at) + + def cancel_purchase(self, pending_id: str, + now: float | None = None) -> dict: + """取消一笔待支付：审批保持未消费（用户可重新发起）， + 已铸的卡即 void（取消待支付 void 虚拟卡，连接器文档 + 口径）。未知 pending 返回 False，不抛错。""" + pending = self._pending.pop(pending_id, None) + if pending is None: + return {"ok": False, "reason": "unknown_pending"} + at = now if now is not None else self._clock() + card_id = pending.get("card_id") + if card_id is not None: + self._wallet.void_card(card_id) + self._audit.record("payment_card_voided", + {"pending_id": pending_id, + "card_id": card_id}, now=at) + self._audit.record( + "payment_cancelled", + {"merchant": pending["details"].merchant, + "amount_minor": pending["details"].amount_minor, + "currency": pending["details"].currency, + "pending_id": pending_id}, now=at) + return {"ok": True, "reason": "cancelled"} + + def pending_ids(self) -> list: + """全部待处理购买的 pending_id（副本）。""" + return list(self._pending) diff --git a/cloud/tests/test_chv_muse_p10_checkout.py b/cloud/tests/test_chv_muse_p10_checkout.py new file mode 100644 index 0000000..454491c --- /dev/null +++ b/cloud/tests/test_chv_muse_p10_checkout.py @@ -0,0 +1,133 @@ +"""对齐 Muse P10 结账检测测试：detect_checkout 纯函数。 + +只用标准库，不启动浏览器、不联网。覆盖： +- URL 信号（结账路径 / 购物车 / 普通页 / 坏 URL）； +- 无障碍树信号（支付字段 / 密码字段 / 下单按钮 / 金额文本）； +- 中英文特征；节点为空；页面金额只作展示信号。 +""" + +import sys +import unittest +from pathlib import Path + +try: + from cloud.chv.checkout_detect import detect_checkout +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.chv.checkout_detect import detect_checkout + + +def _node(role: str, name: str) -> dict: + return {"role": role, "name": name} + + +class CheckoutUrlTest(unittest.TestCase): + def test_checkout_path_is_checkout_high(self) -> None: + verdict = detect_checkout("https://shop.example.com/checkout") + self.assertTrue(verdict.is_checkout) + self.assertEqual(verdict.confidence, "medium") + self.assertIn("checkout_url", verdict.reasons) + + def test_payment_path_is_checkout(self) -> None: + verdict = detect_checkout( + "https://shop.example.com/order/payment?step=2") + self.assertTrue(verdict.is_checkout) + self.assertIn("checkout_url", verdict.reasons) + + def test_cart_path_is_not_checkout(self) -> None: + verdict = detect_checkout("https://shop.example.com/cart") + self.assertFalse(verdict.is_checkout) + self.assertEqual(verdict.confidence, "low") + self.assertIn("cart_url", verdict.reasons) + + def test_plain_page_is_not_checkout(self) -> None: + verdict = detect_checkout("https://shop.example.com/products/1") + self.assertFalse(verdict.is_checkout) + self.assertEqual(verdict.confidence, "low") + self.assertEqual(verdict.reasons, []) + + def test_bad_url_does_not_crash(self) -> None: + verdict = detect_checkout("::::not a url::::") + self.assertFalse(verdict.is_checkout) + + def test_no_nodes_url_only_decision(self) -> None: + verdict = detect_checkout( + "https://shop.example.com/checkout", None) + self.assertTrue(verdict.is_checkout) + + +class CheckoutNodesTest(unittest.TestCase): + def test_payment_fields_make_checkout_high(self) -> None: + nodes = [_node("textbox", "Card number"), + _node("textbox", "Expiry date"), + _node("textbox", "Security code (CVC)")] + verdict = detect_checkout("https://shop.example.com/buy", + nodes) + self.assertTrue(verdict.is_checkout) + self.assertEqual(verdict.confidence, "high") + self.assertEqual(verdict.form_risk, "payment") + self.assertIn("payment_fields", verdict.reasons) + + def test_password_field_marks_form_risk(self) -> None: + nodes = [_node("textbox", "Password")] + verdict = detect_checkout("https://shop.example.com/login", + nodes) + self.assertFalse(verdict.is_checkout) + self.assertEqual(verdict.form_risk, "password") + + def test_order_button_plus_amount_is_checkout(self) -> None: + nodes = [_node("button", "Place order"), + _node("text", "Total: $50.00")] + verdict = detect_checkout("https://shop.example.com/review", + nodes) + self.assertTrue(verdict.is_checkout) + self.assertEqual(verdict.confidence, "medium") + self.assertEqual(verdict.amount_text, "$50.00") + + def test_order_button_alone_is_not_checkout(self) -> None: + nodes = [_node("button", "Place order")] + verdict = detect_checkout("https://shop.example.com/review", + nodes) + self.assertFalse(verdict.is_checkout) + + def test_amount_only_is_not_checkout(self) -> None: + nodes = [_node("text", "Total: $50.00")] + verdict = detect_checkout("https://shop.example.com/review", + nodes) + self.assertFalse(verdict.is_checkout) + self.assertEqual(verdict.amount_text, "$50.00") + + def test_chinese_payment_hints(self) -> None: + nodes = [_node("textbox", "卡号"), + _node("button", "立即支付")] + verdict = detect_checkout("https://shop.example.com/buy", + nodes) + self.assertTrue(verdict.is_checkout) + self.assertEqual(verdict.form_risk, "payment") + + def test_checkout_url_plus_payment_fields_high(self) -> None: + nodes = [_node("textbox", "Card number"), + _node("button", "Pay now")] + verdict = detect_checkout("https://shop.example.com/checkout", + nodes) + self.assertTrue(verdict.is_checkout) + self.assertEqual(verdict.confidence, "high") + self.assertIn("checkout_url", verdict.reasons) + self.assertIn("payment_fields", verdict.reasons) + + def test_yen_and_currency_code_amounts(self) -> None: + nodes = [_node("text", "合計：¥1,200"), + _node("text", "Total 25.00 USD")] + verdict = detect_checkout("https://shop.example.com/review", + nodes) + self.assertEqual(verdict.amount_text, "¥1,200") + + def test_empty_nodes_plain_page(self) -> None: + verdict = detect_checkout("https://shop.example.com/", []) + self.assertFalse(verdict.is_checkout) + self.assertIsNone(verdict.form_risk) + self.assertIsNone(verdict.amount_text) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_chv_muse_p10_gate.py b/cloud/tests/test_chv_muse_p10_gate.py new file mode 100644 index 0000000..5a76a94 --- /dev/null +++ b/cloud/tests/test_chv_muse_p10_gate.py @@ -0,0 +1,346 @@ +"""对齐 Muse P10 支付安全门端到端测试：PaymentGate 编排。 + +只用标准库、假时钟与沙箱钱包，绝不启动真实 VM、不联网、 +不产生真实扣款。覆盖： +- 完整购买链（检测 → 路由 → 限额 → 审批 → 铸卡 → 记账）； +- 审批 payload 含确切购买详情；无 details 时 payload 与 P5 一致； +- 无审批不支付、对话内伪造批准无效、答复过期/重放无效； +- 详情被篡改即拒且审批保持待处理（同一答复换正确详情仍可用）； +- 审批拒绝即停、不扣款；超限额审批前即拒；限额不可用时不编造； +- 无钱包路由即拒；已保存凭证路由；取消待支付；扣款失败 void 卡； +- 审计里绝无卡号。 +""" + +import sys +import unittest +from pathlib import Path + +try: + from cloud.chv.approval_channel import ( + ApprovalRequest, + make_answer, + request_approval, + ) + from cloud.chv.checkout_detect import CheckoutVerdict + from cloud.chv.payments import ( + PaymentAudit, + PaymentCredentialVault, + PaymentGate, + PurchaseDetails, + SimulatedWalletBackend, + SpendLimits, + ) + from cloud.chv.sentinel import ApprovalQueue +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.chv.approval_channel import ( + ApprovalRequest, + make_answer, + request_approval, + ) + from cloud.chv.checkout_detect import CheckoutVerdict + from cloud.chv.payments import ( + PaymentAudit, + PaymentCredentialVault, + PaymentGate, + PurchaseDetails, + SimulatedWalletBackend, + SpendLimits, + ) + from cloud.chv.sentinel import ApprovalQueue + +_NOW = 1_700_000_000.0 +_SECRET = "chan-secret" + + +class _Clock: + """可推进的假时钟（测过期用）。""" + + def __init__(self, now: float = _NOW) -> None: + self.now = now + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +def _details(amount: int = 5000, + merchant: str = "Example Store") -> PurchaseDetails: + return PurchaseDetails( + merchant=merchant, merchant_host="shop.example.com", + items=(("Widget", 2, 2500),), amount_minor=amount, + currency="USD", wallet_provider="stripe-link") + + +def _checkout() -> CheckoutVerdict: + return CheckoutVerdict(True, "high", ["checkout_url"], "payment", + "$50.00") + + +def _not_checkout() -> CheckoutVerdict: + return CheckoutVerdict(False, "low", ["cart_url"], None, None) + + +def _answer_for(payload: dict, verdict: str, secret: str = _SECRET): + """按 payload 重建等价请求对象并签发答复（expires_at + 按默认 900 秒 TTL 反推，nonce 与 pending_id 取自 payload）。""" + req = ApprovalRequest( + payload["pending_id"], payload["target"], + payload["reasons"], nonce=payload["nonce"], + now=payload["expires_at"] - 900) + return make_answer(req, verdict, secret) + + +class GateTest(unittest.TestCase): + def setUp(self) -> None: + self.clock = _Clock() + self.queue = ApprovalQueue(clock=self.clock) + self.wallet = SimulatedWalletBackend(clock=self.clock) + self.vault = PaymentCredentialVault(clock=self.clock) + self.audit = PaymentAudit(clock=self.clock) + self.gate = PaymentGate( + self.queue, self.wallet, self.vault, _SECRET, + clock=self.clock, audit=self.audit) + + def _begin(self, details: PurchaseDetails | None = None, + limits: SpendLimits | None = None) -> dict: + return self.gate.begin_purchase( + details or _details(), _checkout(), limits=limits, + now=self.clock.now) + + def test_begin_ok_payload_has_exact_details(self) -> None: + result = self._begin() + self.assertTrue(result["ok"]) + self.assertEqual(result["route"], "wallet") + self.assertEqual(result["provider"], "stripe-link") + payload = result["payload"] + details = payload["details"] + self.assertEqual(details["merchant"], "Example Store") + self.assertEqual(details["amount_minor"], 5000) + self.assertEqual(details["currency"], "USD") + self.assertEqual(details["items"][0]["name"], "Widget") + self.assertEqual(details["items"][0]["qty"], 2) + self.assertEqual(len(self.queue.list_pending()), 1) + self.assertIn(result["pending_id"], self.gate.pending_ids()) + + def test_begin_not_checkout_refused(self) -> None: + result = self.gate.begin_purchase( + _details(), _not_checkout(), now=self.clock.now) + self.assertFalse(result["ok"]) + self.assertEqual(result["reason"], "not_checkout") + self.assertEqual(self.queue.list_pending(), []) + + def test_begin_no_wallet_route_refused(self) -> None: + wallet = SimulatedWalletBackend(providers=(), clock=self.clock) + gate = PaymentGate(self.queue, wallet, self.vault, _SECRET, + clock=self.clock, audit=self.audit) + result = gate.begin_purchase(_details(), _checkout(), + now=self.clock.now) + self.assertFalse(result["ok"]) + self.assertEqual(result["reason"], "no_wallet_route") + + def test_begin_saved_credential_route(self) -> None: + self.vault.store("shop-pay", "pm_saved", "MC ....5555", + "shop.example.com") + result = self._begin() + self.assertTrue(result["ok"]) + self.assertEqual(result["route"], "saved") + self.assertEqual(result["provider"], "shop-pay") + + def test_begin_over_limit_refused_before_approval(self) -> None: + limits = SpendLimits(per_purchase_minor=4999) + result = self._begin(limits=limits) + self.assertFalse(result["ok"]) + self.assertEqual(result["reason"], "over_per_purchase_limit") + self.assertEqual(self.queue.list_pending(), []) + + def test_begin_limits_unavailable_still_approves(self) -> None: + result = self._begin(limits=SpendLimits()) + self.assertTrue(result["ok"]) + events = [r["event"] for r in self.audit.list_records()] + self.assertIn("payment_approval_requested", events) + requested = [r for r in self.audit.list_records() + if r["event"] == "payment_approval_requested"][0] + self.assertFalse(requested["limits_checked"]) + + def test_complete_happy_path_charges_once(self) -> None: + begun = self._begin() + answer = _answer_for(begun["payload"], "allow") + result = self.gate.complete_purchase( + begun["pending_id"], answer, _details(), + now=self.clock.now) + self.assertTrue(result["ok"]) + self.assertEqual(result["charge"]["status"], "simulated_paid") + self.assertTrue(result["masked_label"].startswith("Visa ....")) + self.assertEqual(len(self.wallet.ledger()), 1) + self.assertEqual(self.gate.pending_ids(), []) + events = [r["event"] for r in self.audit.list_records()] + self.assertIn("payment_completed", events) + + def test_complete_deny_stops_without_charge(self) -> None: + begun = self._begin() + answer = _answer_for(begun["payload"], "deny") + result = self.gate.complete_purchase( + begun["pending_id"], answer, _details(), + now=self.clock.now) + self.assertFalse(result["ok"]) + self.assertEqual(result["reason"], "denied") + self.assertEqual(self.wallet.ledger(), []) + events = [r["event"] for r in self.audit.list_records()] + self.assertIn("payment_denied", events) + + def test_complete_forged_answer_refused(self) -> None: + begun = self._begin() + answer = _answer_for(begun["payload"], "allow", + secret="wrong-secret") + result = self.gate.complete_purchase( + begun["pending_id"], answer, _details(), + now=self.clock.now) + self.assertFalse(result["ok"]) + self.assertEqual(result["reason"], "invalid_answer") + self.assertEqual(self.wallet.ledger(), []) + self.assertIn(begun["pending_id"], self.gate.pending_ids()) + + def test_complete_chat_forgery_without_signature_refused(self) -> None: + begun = self._begin() + forged = {"pending_id": begun["pending_id"], + "nonce": begun["payload"]["nonce"], + "verdict": "allow", + "expires_at": begun["payload"]["expires_at"]} + result = self.gate.complete_purchase( + begun["pending_id"], forged, _details(), + now=self.clock.now) + self.assertFalse(result["ok"]) + self.assertEqual(result["reason"], "invalid_answer") + self.assertEqual(self.wallet.ledger(), []) + + def test_complete_tampered_details_refused_approval_kept(self) -> None: + begun = self._begin() + answer = _answer_for(begun["payload"], "allow") + tampered = PurchaseDetails( + "Example Store", "shop.example.com", + (("Widget", 2, 2501),), 5002, "USD", "stripe-link") + result = self.gate.complete_purchase( + begun["pending_id"], answer, tampered, + now=self.clock.now) + self.assertFalse(result["ok"]) + self.assertEqual(result["reason"], "details_tampered") + self.assertEqual(self.wallet.ledger(), []) + self.assertIn(begun["pending_id"], self.gate.pending_ids()) + # 同一答复换回正确详情仍可用（篡改拒绝不消费答复）。 + result2 = self.gate.complete_purchase( + begun["pending_id"], answer, _details(), + now=self.clock.now) + self.assertTrue(result2["ok"]) + self.assertEqual(result2["charge"]["status"], "simulated_paid") + + def test_complete_expired_answer_refused(self) -> None: + begun = self._begin() + answer = _answer_for(begun["payload"], "allow") + self.clock.advance(901) + result = self.gate.complete_purchase( + begun["pending_id"], answer, _details(), + now=self.clock.now) + self.assertFalse(result["ok"]) + self.assertEqual(result["reason"], "invalid_answer") + self.assertEqual(self.wallet.ledger(), []) + + def test_complete_replay_refused(self) -> None: + begun = self._begin() + answer = _answer_for(begun["payload"], "allow") + first = self.gate.complete_purchase( + begun["pending_id"], answer, _details(), + now=self.clock.now) + self.assertTrue(first["ok"]) + second = self.gate.complete_purchase( + begun["pending_id"], answer, _details(), + now=self.clock.now) + self.assertFalse(second["ok"]) + self.assertEqual(second["reason"], "unknown_pending") + self.assertEqual(len(self.wallet.ledger()), 1) + + def test_complete_unknown_pending_refused(self) -> None: + begun = self._begin() + answer = _answer_for(begun["payload"], "allow") + result = self.gate.complete_purchase( + "pending_nope", answer, _details(), now=self.clock.now) + self.assertFalse(result["ok"]) + self.assertEqual(result["reason"], "unknown_pending") + + def test_cancel_pending_then_complete_fails(self) -> None: + begun = self._begin() + cancelled = self.gate.cancel_purchase( + begun["pending_id"], now=self.clock.now) + self.assertTrue(cancelled["ok"]) + answer = _answer_for(begun["payload"], "allow") + result = self.gate.complete_purchase( + begun["pending_id"], answer, _details(), + now=self.clock.now) + self.assertFalse(result["ok"]) + self.assertEqual(result["reason"], "unknown_pending") + events = [r["event"] for r in self.audit.list_records()] + self.assertIn("payment_cancelled", events) + + def test_cancel_unknown_pending_refused(self) -> None: + result = self.gate.cancel_purchase("pending_nope", + now=self.clock.now) + self.assertFalse(result["ok"]) + self.assertEqual(result["reason"], "unknown_pending") + + def test_charge_failure_voids_card_no_retry(self) -> None: + begun = self._begin() + self.wallet.fail_next_charge = True + answer = _answer_for(begun["payload"], "allow") + result = self.gate.complete_purchase( + begun["pending_id"], answer, _details(), + now=self.clock.now) + self.assertFalse(result["ok"]) + self.assertEqual(result["reason"], "wallet_declined") + ledger = self.wallet.ledger() + self.assertEqual(len(ledger), 1) + self.assertEqual(ledger[0]["status"], "simulated_failed") + + def test_audit_never_contains_card_numbers(self) -> None: + begun = self._begin() + answer = _answer_for(begun["payload"], "allow") + self.gate.complete_purchase( + begun["pending_id"], answer, _details(), + now=self.clock.now) + blob = str(self.audit.list_records()) + self.assertNotIn("400000000000", blob) + self.assertNotIn("4242424242424242", blob) + + +class ApprovalChannelDetailsTest(unittest.TestCase): + def test_payload_without_details_unchanged_from_p5(self) -> None: + queue = ApprovalQueue(clock=_Clock()) + record = {"pending_id": "p1", + "connector": "payment", "method": "purchase", + "scheme": "https", "host": "shop.example.com", + "port": 443, "path": "/checkout", + "reasons": ["payment_approval"]} + req = request_approval(record, _SECRET, now=_NOW) + self.assertNotIn("details", req.payload()) + self.assertEqual(req.payload()["target"]["host"], + "shop.example.com") + + def test_payload_with_details_carries_exact_purchase(self) -> None: + queue = ApprovalQueue(clock=_Clock()) + record = {"pending_id": "p2", + "connector": "payment", "method": "purchase", + "scheme": "https", "host": "shop.example.com", + "port": 443, "path": "/checkout", + "reasons": ["payment_approval"]} + req = request_approval(record, _SECRET, now=_NOW, + details=_details().describe()) + payload = req.payload() + self.assertEqual(payload["details"]["amount_minor"], 5000) + self.assertEqual(payload["details"]["merchant"], + "Example Store") + self.assertNotIn("card_number", str(payload)) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_chv_muse_p10_payments.py b/cloud/tests/test_chv_muse_p10_payments.py new file mode 100644 index 0000000..a5e3ed7 --- /dev/null +++ b/cloud/tests/test_chv_muse_p10_payments.py @@ -0,0 +1,403 @@ +"""对齐 Muse P10 支付单元测试：凭证保险库、一次性卡号、 +购买详情摘要、限额、审计、沙箱钱包。 + +只用标准库与假时钟，绝不启动真实 VM、不联网、不产生真实 +扣款。覆盖： +- 凭证隔离（PAN 拒绝入库、未知提供方拒绝、按次释放绑定、 + 单次兑现、过期、忘记作废、repr 无凭证值）； +- 一次性卡号（换商家/超额/换货币/过期/复用一律拒）； +- 购买详情（自洽校验、摘要防篡改）； +- 限额（三档、边界、限额不可用时不编造上限）； +- 审计（秘密字段名丢弃 + redacted）； +- 沙箱钱包（铸卡、记账、失败 void、账本无卡号）。 +""" + +import sys +import unittest +from pathlib import Path + +try: + from cloud.chv.payments import ( + OneTimeCard, + PaymentAudit, + PaymentCredentialVault, + PaymentRefusal, + PurchaseDetails, + SimulatedWalletBackend, + SpendLimits, + luhn_valid, + ) +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.chv.payments import ( + OneTimeCard, + PaymentAudit, + PaymentCredentialVault, + PaymentRefusal, + PurchaseDetails, + SimulatedWalletBackend, + SpendLimits, + luhn_valid, + ) + +_NOW = 1_700_000_000.0 +_TEST_PAN = "4242424242424242" # Luhn 有效的测试卡号，永不真扣款 + + +class _Clock: + """可推进的假时钟（测过期用）。""" + + def __init__(self, now: float = _NOW) -> None: + self.now = now + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +def _details(amount: int = 5000) -> PurchaseDetails: + return PurchaseDetails( + merchant="Example Store", merchant_host="shop.example.com", + items=(("Widget", 2, 2500),), amount_minor=amount, + currency="USD", wallet_provider="stripe-link") + + +class LuhnTest(unittest.TestCase): + def test_valid_test_pan_passes_luhn(self) -> None: + self.assertTrue(luhn_valid(_TEST_PAN)) + + def test_short_or_bad_number_fails(self) -> None: + self.assertFalse(luhn_valid("123")) + self.assertFalse(luhn_valid("4242424242424241")) + + +class VaultTest(unittest.TestCase): + def setUp(self) -> None: + self.clock = _Clock() + self.vault = PaymentCredentialVault(clock=self.clock) + + def test_store_release_consume_roundtrip(self) -> None: + self.vault.store("stripe-link", "pm_abc", "Visa ....4242", + "shop.example.com") + self.assertTrue(self.vault.has("stripe-link", "pm_abc")) + rel = self.vault.release("stripe-link", "pm_abc", + "shop.example.com", 5000, "USD") + method = self.vault.consume(rel.release_id) + self.assertEqual(method.method_id, "pm_abc") + self.assertEqual(method.masked_label, "Visa ....4242") + + def test_raw_pan_as_method_id_refused(self) -> None: + with self.assertRaises(PaymentRefusal) as ctx: + self.vault.store("stripe-link", _TEST_PAN, "Visa ....4242") + self.assertEqual(ctx.exception.kind, "raw_pan_refused") + + def test_raw_pan_as_label_refused(self) -> None: + with self.assertRaises(PaymentRefusal): + self.vault.store("stripe-link", "pm_abc", _TEST_PAN) + + def test_unknown_provider_refused(self) -> None: + with self.assertRaises(PaymentRefusal) as ctx: + self.vault.store("sketchy-pay", "pm_abc", "Visa ....4242") + self.assertEqual(ctx.exception.kind, "unknown_provider") + + def test_empty_method_id_refused(self) -> None: + with self.assertRaises(PaymentRefusal): + self.vault.store("stripe-link", " ", "Visa ....4242") + + def test_release_scope_mismatch_refused(self) -> None: + self.vault.store("stripe-link", "pm_abc", "Visa ....4242", + "shop.example.com") + with self.assertRaises(PaymentRefusal) as ctx: + self.vault.release("stripe-link", "pm_abc", + "evil.example.com", 5000, "USD") + self.assertEqual(ctx.exception.kind, "scope_mismatch") + + def test_release_unknown_method_refused(self) -> None: + with self.assertRaises(PaymentRefusal): + self.vault.release("stripe-link", "pm_nope", + "shop.example.com", 5000, "USD") + + def test_release_bad_amount_refused(self) -> None: + self.vault.store("stripe-link", "pm_abc", "Visa ....4242") + with self.assertRaises(PaymentRefusal): + self.vault.release("stripe-link", "pm_abc", + "shop.example.com", 0, "USD") + + def test_consume_twice_second_fails(self) -> None: + self.vault.store("stripe-link", "pm_abc", "Visa ....4242") + rel = self.vault.release("stripe-link", "pm_abc", + "shop.example.com", 5000, "USD") + self.vault.consume(rel.release_id) + with self.assertRaises(PaymentRefusal) as ctx: + self.vault.consume(rel.release_id) + self.assertEqual(ctx.exception.kind, "release_unknown") + + def test_consume_expired_release_fails(self) -> None: + self.vault.store("stripe-link", "pm_abc", "Visa ....4242") + rel = self.vault.release("stripe-link", "pm_abc", + "shop.example.com", 5000, "USD", + ttl_seconds=300) + self.clock.advance(301) + with self.assertRaises(PaymentRefusal) as ctx: + self.vault.consume(rel.release_id) + self.assertEqual(ctx.exception.kind, "release_expired") + + def test_forget_removes_and_voids_releases(self) -> None: + self.vault.store("stripe-link", "pm_abc", "Visa ....4242") + rel = self.vault.release("stripe-link", "pm_abc", + "shop.example.com", 5000, "USD") + self.assertTrue(self.vault.forget("stripe-link", "pm_abc")) + self.assertFalse(self.vault.has("stripe-link", "pm_abc")) + with self.assertRaises(PaymentRefusal): + self.vault.consume(rel.release_id) + self.assertFalse(self.vault.forget("stripe-link", "pm_abc")) + + def test_repr_contains_no_credential_value(self) -> None: + self.vault.store("stripe-link", "pm_abc", "Visa ....4242") + self.assertNotIn(_TEST_PAN, repr(self.vault)) + self.assertNotIn("pm_abc", repr(self.vault)) + + def test_methods_for_merchant_scoped(self) -> None: + self.vault.store("stripe-link", "pm_a", "Visa ....4242", + "shop.example.com") + self.vault.store("shop-pay", "pm_b", "MC ....5555", + "other.example.com") + found = self.vault.methods_for_merchant("shop.example.com") + self.assertEqual([m.method_id for m in found], ["pm_a"]) + + +class OneTimeCardTest(unittest.TestCase): + def _card(self) -> OneTimeCard: + return OneTimeCard( + card_id="otc_test", merchant="Example Store", + amount_minor=5000, currency="USD", + expires_at=_NOW + 900, masked_label="Visa ....4242") + + def test_authorize_ok_consumes_once(self) -> None: + card = self._card() + ok, reason = card.authorize("Example Store", 5000, "USD", _NOW) + self.assertTrue(ok) + self.assertEqual(reason, "ok") + ok2, reason2 = card.authorize("Example Store", 5000, "USD", + _NOW) + self.assertFalse(ok2) + self.assertEqual(reason2, "card_exhausted") + + def test_partial_amount_ok(self) -> None: + card = self._card() + ok, _ = card.authorize("Example Store", 4999, "USD", _NOW) + self.assertTrue(ok) + + def test_wrong_merchant_refused(self) -> None: + card = self._card() + ok, reason = card.authorize("Evil Store", 5000, "USD", _NOW) + self.assertFalse(ok) + self.assertEqual(reason, "wrong_merchant") + + def test_amount_exceeded_refused(self) -> None: + card = self._card() + ok, reason = card.authorize("Example Store", 5001, "USD", _NOW) + self.assertFalse(ok) + self.assertEqual(reason, "amount_exceeded") + + def test_wrong_currency_refused(self) -> None: + card = self._card() + ok, reason = card.authorize("Example Store", 5000, "EUR", _NOW) + self.assertFalse(ok) + self.assertEqual(reason, "wrong_currency") + + def test_expired_refused(self) -> None: + card = self._card() + ok, reason = card.authorize("Example Store", 5000, "USD", + _NOW + 901) + self.assertFalse(ok) + self.assertEqual(reason, "card_expired") + + def test_describe_and_repr_have_no_pan(self) -> None: + card = self._card() + self.assertNotIn(_TEST_PAN, str(card.describe())) + self.assertNotIn(_TEST_PAN, repr(card)) + + +class PurchaseDetailsTest(unittest.TestCase): + def test_validate_ok_and_total(self) -> None: + details = _details() + details.validate() + self.assertEqual(details.total_minor(), 5000) + + def test_total_mismatch_rejected(self) -> None: + details = _details(amount=5001) + with self.assertRaises(ValueError): + details.validate() + + def test_bad_currency_rejected(self) -> None: + details = PurchaseDetails( + "Example Store", "shop.example.com", + (("Widget", 1, 100),), 100, "US", "stripe-link") + with self.assertRaises(ValueError): + details.validate() + + def test_empty_items_rejected(self) -> None: + details = PurchaseDetails( + "Example Store", "shop.example.com", (), 100, "USD", + "stripe-link") + with self.assertRaises(ValueError): + details.validate() + + def test_unknown_provider_rejected(self) -> None: + details = PurchaseDetails( + "Example Store", "shop.example.com", + (("Widget", 1, 100),), 100, "USD", "sketchy-pay") + with self.assertRaises(ValueError): + details.validate() + + def test_digest_stable_and_tamper_sensitive(self) -> None: + first = _details().digest() + self.assertEqual(first, _details().digest()) + tampered_price = PurchaseDetails( + "Example Store", "shop.example.com", + (("Widget", 2, 2501),), 5002, "USD", "stripe-link") + self.assertNotEqual(first, tampered_price.digest()) + other_merchant = PurchaseDetails( + "Other Store", "shop.example.com", + (("Widget", 2, 2500),), 5000, "USD", "stripe-link") + self.assertNotEqual(first, other_merchant.digest()) + + +class SpendLimitsTest(unittest.TestCase): + def test_within_limits_ok(self) -> None: + limits = SpendLimits(per_purchase_minor=10000, + daily_minor=20000, + thirty_day_minor=100000, + daily_spent_minor=5000) + ok, reason = limits.check(5000) + self.assertTrue(ok) + self.assertEqual(reason, "ok") + + def test_over_per_purchase_limit(self) -> None: + limits = SpendLimits(per_purchase_minor=4999) + ok, reason = limits.check(5000) + self.assertFalse(ok) + self.assertEqual(reason, "over_per_purchase_limit") + + def test_over_daily_limit(self) -> None: + limits = SpendLimits(daily_minor=9000, daily_spent_minor=5000) + ok, reason = limits.check(5000) + self.assertFalse(ok) + self.assertEqual(reason, "over_daily_limit") + + def test_over_thirty_day_limit(self) -> None: + limits = SpendLimits(thirty_day_minor=9000, + thirty_day_spent_minor=5000) + ok, reason = limits.check(5000) + self.assertFalse(ok) + self.assertEqual(reason, "over_thirty_day_limit") + + def test_exact_limit_boundary_ok(self) -> None: + limits = SpendLimits(per_purchase_minor=5000) + ok, _ = limits.check(5000) + self.assertTrue(ok) + + def test_limits_unavailable_never_invents_cap(self) -> None: + limits = SpendLimits() + ok, reason = limits.check(10_000_000) + self.assertTrue(ok) + self.assertEqual(reason, "limits_unavailable") + + +class PaymentAuditTest(unittest.TestCase): + def test_record_keeps_allowed_fields(self) -> None: + audit = PaymentAudit(clock=_Clock()) + rec = audit.record("payment_completed", + {"merchant": "Example Store", + "amount_minor": 5000, + "masked_label": "Visa ....4242", + "pending_id": "abc123"}) + self.assertEqual(rec["event"], "payment_completed") + self.assertEqual(rec["amount_minor"], 5000) + self.assertEqual(rec["masked_label"], "Visa ....4242") + + def test_secret_field_names_dropped_and_flagged(self) -> None: + audit = PaymentAudit(clock=_Clock()) + rec = audit.record("payment_completed", + {"merchant": "Example Store", + "card_number": _TEST_PAN, + "token": "tok_secret", + "amount_minor": 5000}) + self.assertNotIn("card_number", rec) + self.assertNotIn("token", rec) + self.assertTrue(rec["redacted"]) + self.assertNotIn(_TEST_PAN, str(rec)) + + def test_unknown_fields_dropped_silently(self) -> None: + audit = PaymentAudit(clock=_Clock()) + rec = audit.record("payment_completed", + {"merchant": "Example Store", + "buyer_email": "a@example.com"}) + self.assertNotIn("buyer_email", rec) + self.assertEqual(len(audit.list_records()), 1) + + +class SimulatedWalletTest(unittest.TestCase): + def setUp(self) -> None: + self.clock = _Clock() + self.wallet = SimulatedWalletBackend(clock=self.clock) + + def test_supports_known_providers_only(self) -> None: + self.assertTrue(self.wallet.supports("stripe-link")) + self.assertTrue(self.wallet.supports("shop-pay")) + self.assertFalse(self.wallet.supports("sketchy-pay")) + + def test_mint_and_charge_roundtrip(self) -> None: + card = self.wallet.mint_card(_details(), now=_NOW) + self.assertTrue(card.masked_label.startswith("Visa ....")) + result = self.wallet.charge(card, _details(), now=_NOW) + self.assertEqual(result["status"], "simulated_paid") + self.assertEqual(result["amount_minor"], 5000) + self.assertEqual(len(self.wallet.ledger()), 1) + + def test_no_pan_leaks_into_ledger_or_repr(self) -> None: + card = self.wallet.mint_card(_details(), now=_NOW) + self.wallet.charge(card, _details(), now=_NOW) + blob = str(self.wallet.ledger()) + repr(self.wallet) + self.assertNotIn("400000000000", blob) + + def test_charge_wrong_merchant_refused(self) -> None: + card = self.wallet.mint_card(_details(), now=_NOW) + other = PurchaseDetails( + "Other Store", "shop.example.com", + (("Widget", 2, 2500),), 5000, "USD", "stripe-link") + result = self.wallet.charge(card, other, now=_NOW) + self.assertEqual(result["status"], "refused") + self.assertEqual(result["reason"], "wrong_merchant") + + def test_fail_next_charge_voids_card(self) -> None: + card = self.wallet.mint_card(_details(), now=_NOW) + self.wallet.fail_next_charge = True + result = self.wallet.charge(card, _details(), now=_NOW) + self.assertEqual(result["status"], "simulated_failed") + self.assertEqual(result["reason"], "wallet_declined") + ok, reason = card.authorize("Example Store", 5000, "USD", + _NOW) + self.assertFalse(ok) + self.assertEqual(reason, "card_exhausted") + + def test_void_card_unknown_and_twice(self) -> None: + self.assertFalse(self.wallet.void_card("otc_nope")) + card = self.wallet.mint_card(_details(), now=_NOW) + self.assertTrue(self.wallet.void_card(card.card_id)) + self.assertFalse(self.wallet.void_card(card.card_id)) + + def test_add_and_list_methods(self) -> None: + self.wallet.add_method("stripe-link", "pm_abc", + "Visa ....4242", "shop.example.com") + methods = self.wallet.list_methods("shop.example.com") + self.assertEqual(len(methods), 1) + self.assertEqual(methods[0]["method_id"], "pm_abc") + self.assertNotIn(_TEST_PAN, str(methods)) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/chv-muse-p10-gap.md b/docs/chv-muse-p10-gap.md new file mode 100644 index 0000000..6f413ec --- /dev/null +++ b/docs/chv-muse-p10-gap.md @@ -0,0 +1,122 @@ +# 对齐 Muse P10 差距表：支付安全（凭证隔离 + 结账检测 + 每次购买人工审批 + 一次性卡号） + +依据：`~/workspace/research_notes/meta-muse-ai-architecture-20261101-0755/report.md` +（官方以 Meta security.muse.ai 2026-09-08 为准；第三方转述要标注 +来源性质，不得把转述写成官方承诺）。Muse 官方支付安全要求 +（报告 §3.3，）：购物是最常见浏览器用例之一，Meta 对此 +格外谨慎——已保存支付凭证的网站，浏览器内置结账页检测，每次都 +弹出人工审批并展示确切购买详情；未购物过的网站，Muse 内置合规 +钱包，首发合作 Stripe Link，Shop Pay 即将来临；支付时发行一次性 +卡号给商家，而非真实卡号；该凭证绑定特定商家、特定金额、有限 +有效期，即使经提示注入被盗也“不太有用”；Muse 是首个被 Link 购买 +保障覆盖的 AI 代理（损坏/丢失商品免费保障、降价保障、免费退货、 +符合条件购买的退货保证——保障条款以 Link 官方为准，本轨不编 +细节、不实现理赔流程）。 + +连接器文档补充（wallet tools 的实际口径，见 +`/opt/hatch/connectors/stripe-link/CONNECTOR.md`、 +`shop-pay/CONNECTOR.md`、`shopify-ucp/CONNECTOR.md`）： +Link 在一次购买获批后注资一张一次性虚拟卡，商家永远收不到真实 +卡号；Shop Pay 原生路由用一次性 token、卡片路由在审批后才由浏览 +器填卡字段；每次购买走独立 browser task 与独立审批；审批被拒即 +停止、不扣款；结果未知时不重试、不换卡、不换路；取消待支付会 +void 虚拟卡；已完成扣款无钱包退款、走商家退款流程；直接结账须 +先查限额（每笔/每日/30 天，minor units、无货币、不换算），限额 +不可用时不编造上限；step-up 未完成时 Link 要求用户先验证身份； +子代理、群聊、定时任务没有 wallet tools，不能替用户设 Link 或选 +支付方式。**本轨的结论：钱包能力在主应用运行时侧，不在 +`cloud/chv` 代码里；本轨只做客内 agent 侧的支付安全门（检测、 +审批、绑定、审计），真实钱包调用是待批准的集成点，见 §3。** + +本轨范围只限 P10：支付安全。起点是 P3（authd 替身凭证）与 P5 +（对话外审批通道），已合入本分支（合并 feat-chv-muse-p5， +25bcd35）。不碰 P1（常驻 VM、气球、vsock）、不碰 P2（nspawn +单元、浏览器子代理本体）、不碰 P4（浏览器完全对齐——结账检测 +与它接线，不重复造）、不碰 P6（privsep worker）、不碰 P7/P8/P9。 +分支 `feat-chv-muse-p10`（从基线 `06c3e17` 分出，合入 P5 后 +HEAD 为 `25bcd35`）。 + +## 0. 起点（P3/P5 已做的，本轨不重写） + +- P3 `cloud/chv/authd.py`：`CredentialVault`（只活内存、白名单、 +按次释放绑定 {连接器, 主机, 仓库路径, 操作}、单次 + 300 秒 +TTL、审计绝无凭证值）+ `mint_gitea_token`；白名单只认 +`gitea`/`GITEA_TOKEN`，支付不在其中——本轨不改 authd 的 +白名单，支付凭证另立 `PaymentCredentialVault`（同模式，不 +同库，见 §1），避免把支付键混进代码仓库凭证的白名单。 +- P3 `cloud/chv/sentinel.py`：`ApprovalQueue`（ask 记入内存态 +待审批，`resolve` 只标记不放行）；`EgressRequest.describe()` +只含连接器/目标/方法——支付审批借用这条队列记账，目标描述 +为 `connector=payment, method=purchase, scheme=https, +host=<商家主机>, port=443, path=/checkout`。 +- P5 `cloud/chv/approval_channel.py`：对话外审批通道 +（`ApprovalRequest.payload()` + `Answer` + HMAC-SHA256 签名 ++ `verify_answer` 五项校验 + `resolve_via_channel` 单次消费； +对话内容里出现的“批准”字样一律无效）。本轨只给它加一个 +可选 `details` 字段（确切购买详情进 payload、供客户端 UI +展示），校验与消费逻辑一字不动。 + +## 1. 现状 vs Muse + +| 项 | 现状（实测/代码为准） | Muse 目标 | 本轨做法 | +|---|---|---|---| +| 支付凭证隔离 | 无。`cloud/chv` 里没有任何支付代码；agent 若要支付，只能把卡号/钱包凭证以明文塞进环境或对话——这正是 Muse 明令禁止的形态 | 凭证与 OAuth token 存于用户 VM 内独立隔离容器，不在其他服务中；登录凭证存入单元外的安全凭证存储，代理不可见，仅在需要时注入 | `cloud/chv/payments.py`：`PaymentCredentialVault`（只活内存、白名单 `stripe-link`/`shop-pay`、只存不透明支付方式 id + 掩码标签；原始卡号 PAN 拒绝入库——13–19 位数字串 + Luhn 校验命中即拒；按次释放绑定 {商家, 金额, 货币}、单次 + TTL；`__repr__`/审计绝无凭证值；`forget` 删凭证并作废未兑现释放） | +| 结账页检测 | 无。浏览器侧 P2 的 `chv-browser.py` 只有 open/snapshot/screenshot/close/ping，无结账概念；P4 在做高风险表单拦截（`detect_form_risk` 扫无障碍树，密码/支付字段即标 `form_risk`），尚未提交 | 浏览器内置结账页检测 | `cloud/chv/checkout_detect.py`（纯函数，标准库）：`detect_checkout(url, a11y_nodes)`——URL 信号（/checkout、/payment、/pay、/order-review 等）+ 无障碍树信号（卡号/有效期/安全码字段、“下单/支付”按钮、金额文本）；返回 `is_checkout`/`confidence`/`reasons`/`form_risk`/`amount_text`；确定性规则，不冒充模型分类器；只读 a11y 节点，绝不读原始 DOM；与 P4 的 `detect_form_risk` 同输入形状，P4 的 broker 在高风险表单提交前调本检测（接线点见 §4，P4 未提交前先留接口 + 测试） | +| 每次购买人工审批 | 无。没有任何“购买前停下”的机制 | 已保存支付凭证的网站，每次都弹出人工审批并展示确切购买详情 | `PaymentGate.begin_purchase`：结账检测未命中即拒；路由（见下行）；限额检查（见下行）；经 P5 通道发审批，payload 的 `details` 含确切购买详情（商家、商品、数量、金额、货币、钱包提供方）；审批只走通道答复，对话内“批准”无效；`complete_purchase` 先验签（`verify_answer`，不消费）、再验购买详情摘要（`PurchaseDetails.digest()`，篡改即拒且审批保持待处理）、最后 `resolve_via_channel` 单次消费；审批被拒即停止，不扣款 | +| 一次性卡号 | 无 | 支付时发行一次性卡号给商家而非真实卡号；绑定特定商家、特定金额、有限有效期；被盗也“不太有用” | `OneTimeCard`（绑定 {商家, 金额, 货币, 有效期}，单次兑现；换商家/超额/过期/复用一律拒 + 审计）；`SimulatedWalletBackend`（沙箱钱包：假 PAN 只在后端私有内存里，调用方只见掩码标签与 card_id；`charge` 只记账不联网；取消即 void）；真实一次性虚拟卡由 Link 在其侧发行（连接器文档原文），本轨的卡是绑定与兑现校验的等价模型，不是发卡行——见 §3 | +| 钱包路由 | 无。连接器文档有 stripe-link / shop-pay / shopify-ucp 三条路由，但 `cloud/chv` 里无路由逻辑 | 未购物过的网站走内置合规钱包，首发 Stripe Link，Shop Pay 即将来临 | `PaymentGate` 路由：商家有已保存凭证 → 已保存凭证路由；无 → 合规钱包路由，stripe-link 优先、shop-pay 次之；两家都不支持 → 拒绝（`no_wallet_route`），不编造路由。**来源时间差**：报告写于 2026-11-01 时 Shop Pay 是“即将来临”，连接器文档（当前）显示 Shop Pay 已可用；本轨按连接器文档现状写优先级，并在此标注时间差 | +| 限额 | 无 | （连接器文档）直接结账须先查限额：每笔/每日/30 天，minor units、无货币、不换算；限额不可用时不编造上限 | `SpendLimits.check(amount_minor)`：三档逐项比 minor units；超任一档即拒（`over_limit`，审批前就拒）；三档全为 None（不可用）→ 通过但记 `limits_checked=False`，绝不编造上限 | +| 支付审计 | 无 | 审计轨迹展示已做与计划做的事 | `PaymentAudit.record`：事件（`payment_approval_requested` / `payment_approved` / `payment_denied` / `payment_completed` / `payment_refused` / `payment_cancelled` / `payment_card_voided`）+ 允许字段（商家、金额、货币、掩码标签、pending_id、原因）；字段名命中 `card_number/pan/cvv/token/secret/password` 一律丢弃并标 `redacted=True`；卡号与凭证值绝不进审计 | + +## 2. 改了什么（本轨交付物） + +- `cloud/chv/checkout_detect.py`（新，标准库）：`CheckoutVerdict` + `detect_checkout(url, a11y_nodes)`（URL 信号 + 无障碍树信号 + 金额文本抽取；支付字段即 `form_risk=payment`、密码字段即 `form_risk=password`；页面文本只作展示信号，扣款金额永远以 `PurchaseDetails` 为准——页面文本是不受信任输入，见 P6 口径）。 +- `cloud/chv/payments.py`（新，标准库）：`PaymentCredentialVault` + `PaymentRelease` + `OneTimeCard` + `SimulatedWalletBackend` + `PurchaseDetails`（`digest()` 防篡改）+ `SpendLimits` + `PaymentAudit` + `PaymentGate`（编排：检测 → 路由 → 限额 → 审批 → 铸卡 → 兑现校验 → 记账 → 审计）。 +- 接线（`cloud/chv/approval_channel.py`，最小扩展）：`ApprovalRequest` 增加可选 `details` 字段，`payload()` 有 details 时才带上；`request_approval` 增加可选 `details` 参数；`verify_answer`/`resolve_via_channel` 一字不动；P5 既有测试不受影响。 +- `cloud/tests/test_chv_muse_p10_checkout.py`、`test_chv_muse_p10_payments.py`、`test_chv_muse_p10_gate.py`（新测试，含否定测试：无审批不支付、审批详情被篡改即拒、一次性卡号换商家/超额/过期/复用被拒、原始卡号拒绝入库、凭证泄露面为零、对话内伪造批准无效、答复重放无效、超限额审批前即拒、限额不可用时不编造上限、取消即 void 卡）。 + +## 3. 诚实边界（本轮不宣称的东西） + +1. **真实支付不做，零真实扣款**：全部用沙箱/模拟钱包与假卡号 +（Stripe 测试卡号段 `400000000000xxxx`）验证，不联网、不调 +任何真实钱包接口。真实 Link / Shop Pay 集成（wallet tools +是主应用运行时能力，不在 `cloud/chv` 代码里）列为**待用户 +另行批准**的事项，本轨不产生任何真实扣款、不持有任何真实 +卡号。 +2. **一次性卡号的“发行”在本轨是模拟**：真实一次性虚拟卡由 +Link 在其侧发行（连接器文档原文：“Link funds a one-time +virtual card”）；本轨的 `OneTimeCard` 是绑定（商家 + 金额 ++ 货币 + 有效期）与兑现校验的等价模型，不是发卡行。不要把 +“模拟钱包铸了测试卡号”写成“已接通 Link 发卡”。 +3. **购买保障只作口径引用**：Link 购买保障（损坏/丢失、降价、 +免费退货、退货保证）的 eligibility 由 Link 判定、理赔走 +Link；本轨不实现理赔流程、不承诺任何一笔购买受保障，保障 +条款以 Link 官方页面为准。 +4. **结账检测是启发式，不是 100% 检出**：URL 特征 + 无障碍树 +特征；商户自定义结账页、单页应用内嵌结账可能漏检。漏检时 +的安全底线是：无审批的支付一律拒绝（`PaymentGate` 拒绝 +任何未经审批的扣款），不宣称“所有结账页都会被拦下”。 +5. **限额检查依赖钱包侧返回的限额**：限额不可用时不编造上限， +只记 `limits_checked=False`；审批仍须过。不要把“限额未知” +写成“限额已校验通过”。 +6. **钱包路由优先级按连接器文档现状**：报告（2026-11-01）写 +Shop Pay“即将来临”，连接器文档（当前）显示 Shop Pay 已可 +用；本轨按连接器文档现状写 stripe-link 优先、shop-pay 次之， +并标注了这一时间差；若连接器文档滞后于实际，以实际钱包 +能力为准。 + +## 4. 真机待验证清单（未做，排队在 round-11 与 P4～P9 之后） + +1. P4 的 broker 接线：高风险表单提交前调 `detect_checkout`， +结账页命中时 `open` 导航返回 `ask` 并暂停（与 P4 差距表 +§4 的 `form_submit` ask 种类对齐）。 +2. 真实钱包集成（待用户另行批准）：wallet tools 的 +`list_payment_methods` / 审批单 / `approved_card` 回包与 +`PaymentGate` 的对接；`payment_method_changed`（用户在 +审批单里换卡）时详情摘要重算与重新审批。 +3. 真实商户结账页上的检测命中率抽查（Shopify / Stripe +Checkout / 常见电商各一例），漏检样例记入回归。 +4. 限额读取：`wallet.get_user_info` 的三档限额与本轨 +`SpendLimits` 的对接；限额不可用时的表现。 +5. 取消与未知结果：取消待支付 void 虚拟卡；结果未知时不重试 +的端到端表现（须真实钱包侧配合）。