Pyth-Wormhole VAA Processor

A micro-library that demonstrates, step-by-step, how to unpack and verify a single Pyth price-update VAA transported over the Wormhole network.

Goals

1. Keep the logic atomic – each helper does exactly one thing.
2. Prove every helper with pytest unit-tests exercised on real BTC price-update VAAs (both a frozen fixture and an optionally refreshed one).
3. Remain dependency-light: the core code is pure Python (only pycryptodome for Keccak hashing and eth_keys for ECDSA recovery). Extra features (fetch_vaa_from_wormholescan) pull in requests only when you invoke them, so import-time stays minimal.

Folder layout (key files)

pyth_wormhole_vaa_processor/
├── helpers.py                 # all helper functions
├── scripts/
│   └── fetch_fixture.py       # pull latest VAA from Wormholescan
├── tests/
│   ├── btc_vaa_analysis.json  # frozen fixture
│   ├── latest_btc_vaa.json    # (generated on-demand)
│   ├── test_helpers.py        # body/header helpers
│   └── test_signatures.py     # guardian-set + quorum validation
└── README.md                  # this file

Helpers (public API)

Function	Purpose
decode_base64_to_bytes(b64)	Base-64 → raw bytes (validates)
strip_pnau_wrapper(raw)	Remove the 10-byte PNAU wrapper Wormhole uses in its REST payloads
extract_header(vaa)	(version, guardian_set_index, signature_count)
extract_signatures(vaa, sig_cnt)	List of GuardianSignature objects and offset of the body
extract_body(vaa, offset)	Raw body bytes – exactly what guardians sign
body_keccak(body)	Single Keccak-256 hash (bytes32)
recover_pubkey(hash32, sig)	secp256k1 public key (bytes65) via Xian ecdsa_recover
pubkey_to_eth_address(pub65)	Ethereum address (0x-prefixed, checksum-agnostic)
verify_signature(hash32, sig, expected_addr)	Convenience boolean
verify_vaa_signatures(body, sigs)	Full 2/3-quorum check against Guardian-Set-4
fetch_vaa_from_wormholescan(...)	Fetch raw VAA JSON from Wormholescan

All helpers are pure (no global state) so they can be combined freely.

Running the tests

From the repo root:

# one-off: ensure project root is on PYTHONPATH if you're deep in a subdir
export PYTHONPATH=$PYTHONPATH:"$(pwd)"

pytest -q pyth_wormhole_vaa_processor/tests

You should see all tests pass (currently eight assertions across two test modules) – every helper proven on real BTC VAAs.

Extending

Add new helpers in helpers.py, create a matching test_<name>.py alongside test_helpers.py, import previous helpers in the test to build inputs, and keep each test minimal yet complete.

Wire-format reference

Below is an authoritative crib-sheet for the binary formats that the helpers work with. Keep it close to the code so future contributors don't have to hunt for the spec.

1 Wormhole VAA

A VAA (Verified Action Approval) is what Guardians sign. At the byte level it is:

┌──────────────────┐                          ┌────────────────────────┐
│   Header         │                          │         Body           │
│──────────────────│                          │────────────────────────│
│ u8  version      │                          │ u32  timestamp         │
│ u32 guardian_set │                          │ u32  nonce             │
│ u8  sig_count    │                          │ u16  emitter_chain     │
│                  │ sig_count × signatures → │ bytes32 emitter_addr   │
│                  │                          │ u64  sequence          │
│                  │                          │ u8   consistency_level │
│                  │                          │ bytes  payload         │
└──────────────────┘                          └────────────────────────┘

Signature element (65 bytes each):

┌────────┬──────────────────────────────────────────────┐
│ u8 idx │ r (32) | s (32) | v (1)                      │
└────────┴──────────────────────────────────────────────┘

Signing procedure (what Guardians actually sign):

digest = keccak256( keccak256(body_bytes) )
signature = ecdsa_sign(digest, guardian_private_key)

All fields are big-endian.

2 Pyth AccumulatorUpdateData (payload inside the body)

The Pyth oracle piggy-backs its own TLV structure inside payload.

2.1 Envelope header (9 bytes)

┌──────────┬────────┬────────┬──────────────┬──────────┐
│ u32 magic│  u8 maj│  u8 min│ u16 trail_sz │  u8 type │
└──────────┴────────┴────────┴──────────────┴──────────┘

• magic = 0x41555756 ("AUWV"). (Mnemonic: Accumulator-Update Wormhole Verification.) Do not confuse this with the outer REST wrapper "PNAU" (which is stripped before VAA parsing).
• maj/min = protocol version (currently 1.0).
• trail_sz = length of an optional trailing section immediately after the header.
• type = proof-type ( 0 → Wormhole). The helpers ignore it for now.

Immediately after the 9-byte header (and after any trailing section) comes a series of update records.

2.2 Price-Feed message (variant 0x00)

┌─────────────────────────────┐
│ u8    variant   (=0x00)     │
│ bytes32 feed_id            │
│ i64   price               │
│ u64   confidence          │
│ i32   exponent            │
│ u64   publish_time        │
│ u64   prev_publish_time   │
│ i64   ema_price           │
│ u64   ema_conf            │
└─────────────────────────────┘   (total 1 + 32 + 8×4 + 4 = 84 bytes)

A convenience dataclass PriceUpdate in helpers.py maps one-to-one to this structure.

2.3 TWAP message (variant 0x01)

Not yet parsed by this mini-library, but for completeness:

variant (0x01) | bytes32 feed_id | u128 cumulative_price | u128 cumulative_conf |
                u64 num_down_slots | i32 exponent | u64 publish_time |
                u64 prev_publish_time | u64 publish_slot

3 Putting it together

1. Remove the REST wrapper (PNAU + length) if present.
2. Parse the VAA header & signatures, slice off the body.
3. Use parse_body_header() to get the Wormhole body header + offset.
4. Feed the payload into parse_accumulator_header() to verify the PNAU magic & locate the first update record.
5. Call decode_price_update_payload() (or future TWAP decoder) to obtain the oracle data.

---

The unit tests (tests/test_*) exercise this pipeline end-to-end on a real BTC price-update VAA. Adding new helpers? Create a matching test that proves it on the same fixture. Green tests == guaranteed parsing contract.

Refreshing the fixture (optional)

The repo ships with a frozen BTC price-update fixture (btc_vaa_analysis.json). If you want to test against today's VAA you can pull a fresh one straight from the public Wormholescan REST API:

# 1. Fetch and write a new JSON fixture (uses the new helper)
python -m pyth_wormhole_vaa_processor.scripts.fetch_fixture \
       --emitter-chain 26 \   # Pythnet
       --emitter-address e101faedac5851e32b9b23b5f9411a8c2bac4aae3ed4dd7b811dd1a72ea4aa71 \
       --out pyth_wormhole_vaa_processor/tests/latest_btc_vaa.json

# 2. Re-run the tests – the suite auto-detects the freshest fixture
pytest -q pyth_wormhole_vaa_processor/tests

The signature-validation test (test_signatures.py) now verifies that:

1. Each (r,s,v) pair is a mathematically valid ECDSA signature of the double-Keccak digest of the body.
2. The recovered address for every signature matches the hard-coded list of Guardian-Set-4 addresses.
3. At least 13 of the 19 guardians sign (2/3 quorum) – i.e. verify_vaa_signatures returns valid=True.

Printing the recovered addresses for manual inspection is easy:

from pyth_wormhole_vaa_processor.helpers import *
import base64, json, textwrap

raw = json.load(open('pyth_wormhole_vaa_processor/tests/latest_btc_vaa.json'))['vaa_base64']
vaa = strip_pnau_wrapper(decode_base64_to_bytes(raw))
_,_,cnt = extract_header(vaa)
sigs,off = extract_signatures(vaa,cnt)
body     = extract_body(vaa, off)
digest   = body_digest(body)

print('\n'.join(textwrap.wrap(', '.join(signature_to_address(digest,s) for s in sigs), 78)))

You should see 13 addresses — all present in the Wormhole Guardian Set 4.