host
b55b49b358
- 支付凭证隔离：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 字段）。
347 lines
14 KiB
Python
347 lines
14 KiB
Python
"""对齐 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()
|