BIP352: Add intermediate vector material for silent payments

bip-0352/reference.py

@@ -116,7 +116,7 @@ def decode_silent_payment_address(address: str, hrp: str = "tsp") -> Tuple[ECPub
     return B_scan, B_spend
 
 
-def create_outputs(input_priv_keys: List[Tuple[ECKey, bool]], outpoints: List[COutPoint], recipients: List[str], hrp="tsp") -> List[str]:
+def create_outputs(input_priv_keys: List[Tuple[ECKey, bool]], outpoints: List[COutPoint], recipients: List[str], expected: Dict[str, any] = None, hrp="tsp") -> List[str]:
     G = ECKey().set(1).get_pubkey()
     negated_keys = []
     for key, is_xonly in input_priv_keys:
@@ -129,10 +129,17 @@ def create_outputs(input_priv_keys: List[Tuple[ECKey, bool]], outpoints: List[CO
     if not a_sum.valid:
         # Input privkeys sum is zero -> fail
         return []
+    assert ECKey().set(bytes.fromhex(expected.get("input_private_key_sum"))) == a_sum, "a_sum did not match expected input_private_key_sum"
+
     input_hash = get_input_hash(outpoints, a_sum * G)
     silent_payment_groups: Dict[ECPubKey, List[ECPubKey]] = {}
     for recipient in recipients:
-        B_scan, B_m = decode_silent_payment_address(recipient, hrp=hrp)
+        B_scan, B_m = decode_silent_payment_address(recipient["address"], hrp=hrp)
+        # Verify decoded intermediate keys for recipient
+        expected_B_scan = ECPubKey().set(bytes.fromhex(recipient["scan_pub_key"]))
+        expected_B_m = ECPubKey().set(bytes.fromhex(recipient["spend_pub_key"]))
+        assert expected_B_scan == B_scan, "B_scan did not match expected recipient.scan_pub_key"
+        assert expected_B_m == B_m, "B_m did not match expected recipient.spend_pub_key"
         if B_scan in silent_payment_groups:
             silent_payment_groups[B_scan].append(B_m)
         else:
@@ -141,6 +148,16 @@ def create_outputs(input_priv_keys: List[Tuple[ECKey, bool]], outpoints: List[CO
     outputs = []
     for B_scan, B_m_values in silent_payment_groups.items():
         ecdh_shared_secret = input_hash * a_sum * B_scan
+
+        expected_shared_secrets = expected.get("shared_secrets", {})
+        # Find the recipient address that corresponds to this B_scan and get its index
+        for recipient_idx, recipient in enumerate(recipients):
+            recipient_B_scan = ECPubKey().set(bytes.fromhex(recipient["scan_pub_key"]))
+            if recipient_B_scan == B_scan:
+                expected_shared_secret_hex = expected_shared_secrets[recipient_idx]
+                assert ecdh_shared_secret.get_bytes(False).hex() == expected_shared_secret_hex, f"ecdh_shared_secret did not match expected, recipient {recipient_idx} ({recipient['address']}): expected={expected_shared_secret_hex}"
+                break
+
         k = 0
         for B_m in B_m_values:
             t_k = TaggedHash("BIP0352/SharedSecret", ecdh_shared_secret.get_bytes(False) + ser_uint32(k))
@@ -151,9 +168,15 @@ def create_outputs(input_priv_keys: List[Tuple[ECKey, bool]], outpoints: List[CO
     return list(set(outputs))
 
 
-def scanning(b_scan: ECKey, B_spend: ECPubKey, A_sum: ECPubKey, input_hash: bytes, outputs_to_check: List[ECPubKey], labels: Dict[str, str] = {}) -> List[Dict[str, str]]:
+def scanning(b_scan: ECKey, B_spend: ECPubKey, A_sum: ECPubKey, input_hash: bytes, outputs_to_check: List[ECPubKey], labels: Dict[str, str] = None, expected: Dict[str, any] = None) -> List[Dict[str, str]]:
     G = ECKey().set(1).get_pubkey()
+    input_hash_key = ECKey().set(input_hash)
+    computed_tweak_point = input_hash_key * A_sum
+    assert computed_tweak_point.get_bytes(False).hex() == expected.get("tweak"), "tweak did not match expected"
+
     ecdh_shared_secret = input_hash * b_scan * A_sum
+    assert ecdh_shared_secret.get_bytes(False).hex() == expected.get("shared_secret"), "ecdh_shared_secret did not match expected shared_secret"
+
     k = 0
     wallet = []
     while True:
@@ -236,11 +259,12 @@ def scanning(b_scan: ECKey, B_spend: ECPubKey, A_sum: ECPubKey, input_hash: byte
                     is_p2tr(vin.prevout),
                 ))
                 input_pub_keys.append(pubkey)
+            assert [pk.get_bytes(False).hex() for pk in input_pub_keys] == expected.get("input_pub_keys"), "input_pub_keys did not match expected"
 
             sending_outputs = []
             if (len(input_pub_keys) > 0):
                 outpoints = [vin.outpoint for vin in vins]
-                sending_outputs = create_outputs(input_priv_keys, outpoints, given["recipients"], hrp="sp")
+                sending_outputs = create_outputs(input_priv_keys, outpoints, given["recipients"], expected=expected, hrp="sp")
 
                 # Note: order doesn't matter for creating/finding the outputs. However, different orderings of the recipient addresses
                 # will produce different generated outputs if sending to multiple silent payment addresses belonging to the
@@ -303,6 +327,7 @@ def scanning(b_scan: ECKey, B_spend: ECPubKey, A_sum: ECPubKey, input_hash: byte
                     # Input pubkeys sum is point at infinity -> skip tx
                     assert expected["outputs"] == []
                     continue
+                assert A_sum.get_bytes(False).hex() == expected.get("input_pub_key_sum"), "A_sum did not match expected input_pub_key_sum"
                 input_hash = get_input_hash([vin.outpoint for vin in vins], A_sum)
                 pre_computed_labels = {
                     (generate_label(b_scan, label) * G).get_bytes(False).hex(): generate_label(b_scan, label).hex()
@@ -315,6 +340,7 @@ def scanning(b_scan: ECKey, B_spend: ECPubKey, A_sum: ECPubKey, input_hash: byte
                     input_hash=input_hash,
                     outputs_to_check=outputs_to_check,
                     labels=pre_computed_labels,
+                    expected=expected,
                 )
 
             # Check that the private key is correct for the found output public key