feat(chv): 对齐 Muse P10 支付安全——凭证隔离 + 结账检测 + 每次购买人工审批 + 一次性卡号

- 支付凭证隔离:PaymentCredentialVault(只活内存、白名单
  stripe-link/shop-pay、只存不透明 id + 掩码标签、原始卡号
  PAN 拒绝入库、按次释放绑定商家/金额/货币、单次 + TTL)
- 结账页检测:detect_checkout(URL + 无障碍树信号,确定性
  规则、不读原始 DOM、与 P4 detect_form_risk 同输入形状)
- 每次购买人工审批:PaymentGate 经 P5 对话外通道发审批,
  payload 含确切购买详情;完成时先验签、再验详情摘要
  (篡改即拒且审批保持待处理)、最后单次消费答复
- 一次性卡号:OneTimeCard 绑定商家/金额/货币/有效期,单次
  兑现;SimulatedWalletBackend 沙箱钱包(假 PAN 只在后端
  私有内存、charge 只记账不联网、取消即 void)
- 限额:SpendLimits 三档 minor units 比对,限额不可用时
  不编造上限(limits_checked=False)
- 审计:PaymentAudit 只留允许字段,秘密字段名丢弃 + redacted
- 审批通道最小扩展:ApprovalRequest 增加可选 details 字段,
  不传时 payload 与 P5 完全一致
- 真实支付与扣款不做:全部沙箱验证,真实钱包集成待用户
  另行批准;新测试 79 项,全量 cloud 套件 2037 OK

round-12 集成说明:本提交为 acf98f8 的 cherry-pick(只含 P10
自有 diff,基线 25bcd35 已含的 P3/P5 内容未重复合入;
approval_channel.py 仅加可选 details 字段)。
This commit is contained in:
host committed 2026-11-07 00:09:55 +00:00
1 parent 90c42f6e1d
commit b55b49b358
7 files changed
+1936 -5

No files matched your search

+20 -5
View File
@@ -11,6 +11,9 @@
"reasons": [str, ...], "expires_at": float} "reasons": [str, ...], "expires_at": float}
``target`` 只有目标描述(连接器/方法/scheme/主机/端口/路径), ``target`` 只有目标描述(连接器/方法/scheme/主机/端口/路径),
``reasons`` 只有原因名;载荷里绝无凭证值与用户数据内容。 ``reasons`` 只有原因名;载荷里绝无凭证值与用户数据内容。
P10 扩展:可选 ``details`` 字段(支付审批的确切购买详情,
商家/商品/金额/货币,供客户端 UI 展示;不传时 payload 与
P5 完全一致)。
``nonce`` 是随机 16 字节 hex(32 字符),每次请求唯一。 ``nonce`` 是随机 16 字节 hex(32 字符),每次请求唯一。
``expires_at`` 默认请求后 15 分钟(900 秒),可注入时钟。 ``expires_at`` 默认请求后 15 分钟(900 秒),可注入时钟。
- 答复(``Answer``,客户端答完直接回 Sentinel):: - 答复(``Answer``,客户端答完直接回 Sentinel)::
@@ -72,38 +75,50 @@ class ApprovalRequest:
reasons: Iterable[str] | None, reasons: Iterable[str] | None,
nonce: str | None = None, nonce: str | None = None,
now: float | 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.pending_id = pending_id
self.target = dict(target) self.target = dict(target)
self.reasons = list(reasons or []) self.reasons = list(reasons or [])
self.nonce = nonce or secrets.token_hex(16) self.nonce = nonce or secrets.token_hex(16)
at = now if now is not None else time.time() at = now if now is not None else time.time()
self.expires_at = float(at) + ttl_seconds 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: def payload(self) -> dict:
"""可外发给客户端的请求载荷(只含描述字段 + nonce + """可外发给客户端的请求载荷(只含描述字段 + nonce +
过期,无密钥、无凭证值、无用户数据内容)。""" 过期,无密钥、无凭证值、无用户数据内容)。有 details
return {"pending_id": self.pending_id, "nonce": self.nonce, 时带上(支付审批的确切购买详情)。"""
out: dict = {"pending_id": self.pending_id, "nonce": self.nonce,
"target": dict(self.target), "target": dict(self.target),
"reasons": list(self.reasons), "reasons": list(self.reasons),
"expires_at": self.expires_at} "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, def request_approval(queue_record: dict, secret: str,
now: float | None = None, now: float | None = None,
ttl_seconds: int = DEFAULT_TTL_SECONDS ttl_seconds: int = DEFAULT_TTL_SECONDS,
details: dict | None = None
) -> ApprovalRequest: ) -> ApprovalRequest:
"""为一条队列记录生成通道请求并登记(nonce + 过期)。 """为一条队列记录生成通道请求并登记(nonce + 过期)。
``secret`` 是调用方持有的通道密钥:本模块不保存它、不进 ``secret`` 是调用方持有的通道密钥:本模块不保存它、不进
载荷、不进日志;答复方用同一密钥签名,校验时对上才算数。 载荷、不进日志;答复方用同一密钥签名,校验时对上才算数。
``details`` 可选(P10 支付审批的确切购买详情,进 payload
供客户端 UI 展示;不传时行为与 P5 完全一致)。
返回的 ``ApprovalRequest.payload()`` 即发给客户端的东西。 返回的 ``ApprovalRequest.payload()`` 即发给客户端的东西。
""" """
pending_id = queue_record["pending_id"] pending_id = queue_record["pending_id"]
req = ApprovalRequest( req = ApprovalRequest(
pending_id, _target_from_record(queue_record), pending_id, _target_from_record(queue_record),
queue_record.get("reasons", []), queue_record.get("reasons", []),
now=now, ttl_seconds=ttl_seconds) now=now, ttl_seconds=ttl_seconds, details=details)
_REQUESTS[pending_id] = {"nonce": req.nonce, _REQUESTS[pending_id] = {"nonce": req.nonce,
"expires_at": req.expires_at} "expires_at": req.expires_at}
return req return req
+143
View File
@@ -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)
+769
View File
@@ -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)
+133
View File
@@ -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()
+346
View File
@@ -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()
+403
View File
@@ -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()
+122
View File
@@ -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 虚拟卡;结果未知时不重试
的端到端表现(须真实钱包侧配合)。