diff --git a/blockrun_llm/x402.py b/blockrun_llm/x402.py index 647d545..7da953f 100644 --- a/blockrun_llm/x402.py +++ b/blockrun_llm/x402.py @@ -244,7 +244,7 @@ def extract_payment_details(payment_required: Dict[str, Any]) -> Dict[str, Any]: # SPL program IDs TOKEN_PROGRAM_ID = "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA" -ASSOCIATED_TOKEN_PROGRAM_ID = "ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJe1bRS" +ASSOCIATED_TOKEN_PROGRAM_ID = "ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL" # Compute budget defaults (match @x402/svm) DEFAULT_COMPUTE_UNIT_PRICE_MICROLAMPORTS = 1 @@ -383,7 +383,8 @@ def create_solana_payment_payload( # Partial sign: user signs, fee_payer (CDP) co-signs on server side # fee_payer is always first signer (index 0), owner is second (index 1) - msg_bytes = bytes(message) + # v0 messages must be signed with the 0x80 version prefix included + msg_bytes = b'\x80' + bytes(message) user_sig = keypair.sign_message(msg_bytes) null_sig = Signature.default() # placeholder for fee_payer tx = VersionedTransaction.populate(message, [null_sig, user_sig]) diff --git a/tests/unit/test_x402.py b/tests/unit/test_x402.py index ec6a846..efb5aef 100644 --- a/tests/unit/test_x402.py +++ b/tests/unit/test_x402.py @@ -268,3 +268,56 @@ def test_payload_transaction_is_base64(self): # Should be valid base64 tx_bytes = base64.b64decode(decoded["payload"]["transaction"]) assert len(tx_bytes) > 0 + + + def test_v0_signature_includes_version_prefix(self): + """User signature must be over 0x80 + message_body for v0 transactions.""" + from blockrun_llm.x402 import create_solana_payment_payload + from solders.transaction import VersionedTransaction + from solders.keypair import Keypair + import json + import base64 + import base58 + + payload = create_solana_payment_payload( + private_key=self.TEST_BS58_KEY, + recipient=self.TEST_RECIPIENT, + amount="1000", + fee_payer=self.TEST_FEE_PAYER, + ) + decoded = json.loads(base64.b64decode(payload)) + tx_bytes = base64.b64decode(decoded["payload"]["transaction"]) + tx = VersionedTransaction.from_bytes(tx_bytes) + + # Recover the user keypair + secret = base58.b58decode(self.TEST_BS58_KEY) + keypair = Keypair.from_seed(secret[:32]) + + # The signing data for v0 must include the 0x80 prefix + msg_with_prefix = b'\x80' + bytes(tx.message) + + # Verify the user's signature (index 1) is over the prefixed message + from nacl.signing import VerifyKey + vk = VerifyKey(bytes(keypair.pubkey())) + # Should not raise + vk.verify(msg_with_prefix, bytes(tx.signatures[1])) + + +class TestAssociatedTokenProgramId: + """Verify the Associated Token Program ID is correct.""" + + def test_associated_token_program_id(self): + """ASSOCIATED_TOKEN_PROGRAM_ID must match Solana mainnet.""" + from blockrun_llm.x402 import ASSOCIATED_TOKEN_PROGRAM_ID + + assert ASSOCIATED_TOKEN_PROGRAM_ID == "ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL" + + def test_ata_derivation(self): + """ATA derivation must match on-chain addresses.""" + from blockrun_llm.x402 import _get_ata, USDC_SOLANA + + # Known wallet -> known USDC ATA (verified on-chain) + owner = "CtJTYWPQSL5jw9B2JRHmpQjYCSSgUX3LRvmMBhq55HmQ" + expected_ata = "HZPPxg9ZyoHu4f2pj5uEEXsArLA2rnL9FtDgC8rrAp5Q" + + assert _get_ata(owner, USDC_SOLANA) == expected_ata