| """Reading real P2P captures off disk — all three formats. |
| |
| The debug.log and .btcap fixtures are checked in beside this file. The pcap and |
| pcapng fixtures are *built* here instead: a committed binary blob is a fixture |
| nobody can read in a review, and building the bytes means the test states the |
| wire format it claims to parse rather than trusting a capture somebody took |
| once. |
| |
| The txid check is not self-referential. The parser's hashing convention is |
| pinned to Bitcoin's genesis coinbase transaction — a published vector — and the |
| segwit case computes its expectation from the legacy serialisation this test |
| writes itself, never from the parser. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| import struct |
| from pathlib import Path |
|
|
| import pytest |
|
|
| from p2p import capture_reader as cr |
|
|
| FIXTURES = Path(__file__).resolve().parent / "fixtures" / "capture" |
| TX_GENESIS = ("4a5e1e4baab89f3a32518a88c31bc87f618f76673e2cc77ab2127b7afdeda33b") |
|
|
| |
| |
| |
| GENESIS_RAW = bytes.fromhex( |
| "01000000010000000000000000000000000000000000000000000000000000000000000000" |
| "ffffffff4d04ffff001d0104455468652054696d65732030332f4a616e2f32303039204368" |
| "616e63656c6c6f72206f6e206272696e6b206f66207365636f6e64206261696c6f75742066" |
| "6f722062616e6b73ffffffff0100f2052a01000000434104678afdb0fe5548271967f1a671" |
| "30b7105cd6a828e03909a67962e0ea1f61deb649f6bc3f4cef38c4f35504e51ec112de5c38" |
| "4df7ba0b8d578a4c702b6bf11d5fac00000000") |
|
|
| MAGIC = bytes.fromhex("f9beb4d9") |
| NODE = "198.51.100.2" |
| PEER = "203.0.113.9" |
|
|
|
|
| def double_sha(data: bytes) -> str: |
| return hashlib.sha256(hashlib.sha256(data).digest()).digest()[::-1].hex() |
|
|
|
|
| def segwit_tx() -> tuple[bytes, str, str]: |
| """A one-in one-out segwit transaction, and its two identifiers. |
| |
| Built from the legacy bytes outward, so the expected txid comes from the |
| serialisation this test defines and not from the code under test. |
| """ |
| version = bytes.fromhex("02000000") |
| txin = b"\x11" * 32 + bytes.fromhex("00000000") + b"\x00" + bytes.fromhex("ffffffff") |
| txout = struct.pack("<q", 12_345_678) + bytes([0x19]) + bytes.fromhex( |
| "76a914" + "22" * 20 + "88ac") |
| locktime = bytes.fromhex("00000000") |
| legacy = version + b"\x01" + txin + b"\x01" + txout + locktime |
| witness = b"\x01\x02\xde\xad" |
| wire = version + b"\x00\x01" + b"\x01" + txin + b"\x01" + txout + witness + locktime |
| return wire, double_sha(legacy), double_sha(wire) |
|
|
|
|
| def p2p_message(command: str, payload: bytes) -> bytes: |
| checksum = hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] |
| return (MAGIC + command.encode().ljust(12, b"\x00") |
| + struct.pack("<I", len(payload)) + checksum + payload) |
|
|
|
|
| def inv_message(entries: list[tuple[int, str]]) -> bytes: |
| payload = bytes([len(entries)]) |
| for kind, digest in entries: |
| payload += struct.pack("<I", kind) + bytes.fromhex(digest)[::-1] |
| return p2p_message("inv", payload) |
|
|
|
|
| def tcp_frame(src: str, dst: str, sport: int, dport: int, payload: bytes) -> bytes: |
| """Ethernet + IPv4 + TCP around a payload. Checksums are not verified by |
| anything we wrote, so they are left zero rather than faked.""" |
| tcp = struct.pack("!HHIIBBHHH", sport, dport, 1, 1, 5 << 4, 0x18, 65535, 0, 0) + payload |
| total = 20 + len(tcp) |
| ip = (struct.pack("!BBHHHBBH", 0x45, 0, total, 0, 0, 64, 6, 0) |
| + bytes(int(o) for o in src.split(".")) + bytes(int(o) for o in dst.split("."))) |
| return b"\xaa" * 12 + b"\x08\x00" + ip + tcp |
|
|
|
|
| def write_pcap(path: Path, packets: list[tuple[float, bytes]]) -> Path: |
| out = bytearray(struct.pack("<IHHiIII", 0xA1B2C3D4, 2, 4, 0, 0, 65535, 1)) |
| for ts, frame in packets: |
| out += struct.pack("<IIII", int(ts), int(round((ts % 1) * 1e6)), |
| len(frame), len(frame)) |
| out += frame |
| path.write_bytes(bytes(out)) |
| return path |
|
|
|
|
| def write_pcapng(path: Path, packets: list[tuple[float, bytes]]) -> Path: |
| def block(kind: int, body: bytes) -> bytes: |
| body += b"\x00" * (-len(body) % 4) |
| length = 12 + len(body) |
| return struct.pack("<II", kind, length) + body + struct.pack("<I", length) |
|
|
| shb = block(0x0A0D0D0A, struct.pack("<IHHq", 0x1A2B3C4D, 1, 0, -1)) |
| options = struct.pack("<HH", 9, 1) + b"\x06" + b"\x00" * 3 + struct.pack("<HH", 0, 0) |
| idb = block(0x00000001, struct.pack("<HHI", 1, 0, 65535) + options) |
| out = bytearray(shb + idb) |
| for ts, frame in packets: |
| micros = int(round(ts * 1e6)) |
| out += block(0x00000006, struct.pack("<IIIII", 0, micros >> 32, micros & 0xFFFFFFFF, |
| len(frame), len(frame)) + frame) |
| path.write_bytes(bytes(out)) |
| return path |
|
|
|
|
| @pytest.fixture(scope="module") |
| def wire(): |
| """The stream both pcap fixtures carry: one wtxid inv, then the segwit |
| transaction itself split across two TCP segments, then a txid inv.""" |
| raw, txid, wtxid = segwit_tx() |
| tx_message = p2p_message("tx", raw) |
| cut = len(tx_message) // 2 |
| packets = [ |
| (1_790_244_001.25, tcp_frame(PEER, NODE, 8333, 51000, |
| inv_message([(cr.INV_WTX, wtxid)]))), |
| (1_790_244_001.40, tcp_frame(PEER, NODE, 8333, 51000, tx_message[:cut])), |
| (1_790_244_001.41, tcp_frame(PEER, NODE, 8333, 51000, tx_message[cut:])), |
| (1_790_244_002.00, tcp_frame(NODE, "198.51.100.77", 51001, 8333, |
| inv_message([(cr.INV_TX, TX_GENESIS)]))), |
| (1_790_244_002.10, tcp_frame(PEER, NODE, 443, 9000, b"not bitcoin at all")), |
| ] |
| return {"packets": packets, "txid": txid, "wtxid": wtxid} |
|
|
|
|
| |
| def test_genesis_txid_matches_the_published_vector(): |
| txid, wtxid = cr.txids_of(GENESIS_RAW) |
| assert txid == TX_GENESIS |
| assert wtxid == txid, "a pre-segwit transaction has one identifier, not two" |
|
|
|
|
| def test_segwit_txid_strips_the_witness(): |
| raw, expected_txid, expected_wtxid = segwit_tx() |
| txid, wtxid = cr.txids_of(raw) |
| assert txid == expected_txid |
| assert wtxid == expected_wtxid |
| assert txid != wtxid, "the fixture is pointless if the two ids coincide" |
|
|
|
|
| |
| def test_debug_log_binds_peers_agents_and_directions(): |
| events = cr.read_capture(FIXTURES / "node1.debug.log") |
| assert cr.detect_format(FIXTURES / "node1.debug.log") == "debug_log" |
| kinds = [(e.message_type, e.direction) for e in events] |
| assert kinds == [("inv", "inbound"), ("inv_wtx", "inbound"), |
| ("inv", "outbound"), ("tx", "inbound")] |
|
|
| first = events[0] |
| assert first.txid == TX_GENESIS |
| assert (first.peer_id, first.peer_ip, first.peer_port) == (1, PEER, 8333) |
| assert first.user_agent == "/Satoshi:25.0.0/" |
| assert first.capture_source == "debug.log:node1.debug.log" |
| assert events[1].peer_ip == "198.51.100.77" |
|
|
|
|
| def test_the_version_lines_own_address_is_not_mistaken_for_the_peers(): |
| """`us=198.51.100.2:8333` sits on the same line as `peer=1`. Binding a peer |
| to our own address would make every observation look locally originated, |
| which is the exact error this whole pipeline exists to avoid.""" |
| events = cr.read_capture(FIXTURES / "node1.debug.log") |
| assert NODE not in {e.peer_ip for e in events} |
|
|
|
|
| def test_debug_log_derives_a_clock_from_the_first_event(): |
| events = cr.read_capture(FIXTURES / "node1.debug.log") |
| assert events[0].monotonic_or_derived_ts == 0.0 |
| assert events[-1].monotonic_or_derived_ts == pytest.approx(0.75, abs=1e-6) |
| assert all(e.wall_clock_ts is not None for e in events) |
|
|
|
|
| def test_a_received_tx_line_without_a_txid_is_not_invented(): |
| """debug=net logs `received: tx (223 bytes) peer=1` with no identifier. |
| Emitting a row for it would mean a transaction event with no transaction.""" |
| events = cr.read_capture(FIXTURES / "node1.debug.log") |
| assert all(len(e.txid) == 64 for e in events) |
|
|
|
|
| |
| @pytest.mark.parametrize("writer", [write_pcap, write_pcapng]) |
| def test_pcap_decodes_inv_and_tx_across_segments(tmp_path, wire, writer): |
| path = writer(tmp_path / f"cap{writer is write_pcapng and 'ng' or ''}.pcap", |
| wire["packets"]) |
| events = cr.read_capture(path, local_ips=[NODE]) |
| assert cr.detect_format(path) == "pcap" |
|
|
| |
| txs = [e for e in events if e.message_type == "tx"] |
| assert len(txs) == 1 and txs[0].txid == wire["txid"] |
| assert txs[0].peer_ip == PEER and txs[0].peer_port == 8333 |
| assert txs[0].direction == "inbound" |
|
|
| outbound = [e for e in events if e.direction == "outbound"] |
| assert [(e.txid, e.peer_ip) for e in outbound] == [(TX_GENESIS, "198.51.100.77")] |
|
|
|
|
| @pytest.mark.parametrize("writer", [write_pcap, write_pcapng]) |
| def test_wtxid_inv_is_resolved_by_the_transaction_in_the_same_capture(tmp_path, wire, writer): |
| path = writer(tmp_path / "cap.pcap", wire["packets"]) |
| events = cr.read_capture(path, local_ips=[NODE]) |
| invs = [e for e in events if e.message_type.startswith("inv") and e.direction == "inbound"] |
| assert len(invs) == 1 |
| assert invs[0].message_type == "inv", "the tx was in the capture, so this resolves" |
| assert invs[0].txid == wire["txid"] != wire["wtxid"] |
|
|
|
|
| def test_an_unresolved_wtxid_keeps_its_marker(tmp_path, wire): |
| """Without the `tx` message there is nothing to resolve against, and the |
| row must say the identifier is a wtxid rather than pass it off as a txid.""" |
| inv_only = [wire["packets"][0]] |
| path = write_pcap(tmp_path / "inv_only.pcap", inv_only) |
| (event,) = cr.read_capture(path, local_ips=[NODE]) |
| assert event.message_type == "inv_wtx" |
| assert event.txid == wire["wtxid"] |
|
|
|
|
| def test_non_bitcoin_traffic_and_a_desynchronised_stream_are_survived(tmp_path, wire): |
| """Port 443 traffic is ignored outright; garbage prepended to a P2P flow |
| must resynchronise on the next magic instead of losing the flow.""" |
| raw, txid, _ = segwit_tx() |
| noisy = b"\xff" * 37 + p2p_message("tx", raw) |
| packets = [(1_790_244_003.0, tcp_frame(PEER, NODE, 8333, 51000, noisy))] |
| path = write_pcap(tmp_path / "noisy.pcap", packets) |
| events = cr.read_capture(path, local_ips=[NODE]) |
| assert [e.txid for e in events] == [txid] |
|
|
|
|
| def test_the_capture_host_is_inferred_when_not_given(tmp_path, wire): |
| """Every flow in a host capture has that host at one end.""" |
| path = write_pcap(tmp_path / "cap.pcap", wire["packets"]) |
| inferred = cr.read_capture(path) |
| assert {e.peer_ip for e in inferred} == {PEER, "198.51.100.77"} |
|
|
|
|
| |
| def test_btcap_reads_aliases_and_skips_unusable_lines(): |
| path = FIXTURES / "node1.btcap" |
| events = cr.read_capture(path) |
| assert cr.detect_format(path) == "btcap" |
| assert len(events) == 3, "a malformed line and a line with no message type are dropped" |
| assert events[1].txid == "0000000000000000000000000000000000000000000000000000000000000abc" |
| assert events[1].peer_ip == "198.51.100.77" |
| assert events[1].user_agent == "/bitcoinj:0.16.2/" |
| assert events[1].message_type == "inv_wtx" |
|
|
|
|
| def test_btcap_keeps_a_monotonic_clock_it_was_given(): |
| """A collector that recorded a monotonic reading knows better than an offset |
| derived from wall-clock times that may have been stepped mid-capture.""" |
| events = cr.read_capture(FIXTURES / "node1.btcap") |
| assert events[0].monotonic_or_derived_ts == 12.5 |
| assert events[1].monotonic_or_derived_ts == pytest.approx(0.06, abs=1e-6) |
|
|
|
|
| def test_gzipped_captures_are_read(tmp_path): |
| import gzip |
|
|
| path = tmp_path / "node1.btcap.gz" |
| path.write_bytes(gzip.compress((FIXTURES / "node1.btcap").read_bytes())) |
| assert len(cr.read_capture(path)) == 3 |
|
|
|
|
| |
| def test_read_directory_merges_formats_in_time_order(tmp_path, wire): |
| for name in ("node1.debug.log", "node1.btcap"): |
| (tmp_path / name).write_bytes((FIXTURES / name).read_bytes()) |
| write_pcap(tmp_path / "node1.pcap", wire["packets"]) |
| (tmp_path / "manifest.json").write_text("{}") |
| (tmp_path / "notes.md").write_text("not a capture") |
|
|
| events = cr.read_directory(tmp_path, local_ips=[NODE]) |
| stamps = [e.wall_clock_ts for e in events] |
| assert stamps == sorted(stamps) |
| |
| |
| assert {e.capture_source for e in events} == { |
| "debug.log:node1.debug.log", "btcap:node1.btcap", "pcap:node1.pcap", |
| "collector:node1"} |
|
|
|
|
| def test_to_frame_has_the_canonical_columns(): |
| frame = cr.to_frame(cr.read_capture(FIXTURES / "node1.debug.log")) |
| assert list(frame.columns) == cr.COLUMNS |
| assert str(frame["wall_clock_ts"].dtype).startswith("datetime64") |
|
|
|
|
| def test_detect_format_refuses_to_guess(): |
| with pytest.raises(ValueError): |
| cr.detect_format(FIXTURES.parent / "capture") |
|
|
|
|
| |
| def version_message(agent: str, services: int) -> bytes: |
| """version, services, time, addr_recv, addr_from, nonce, user agent, height.""" |
| payload = (struct.pack("<iQq", 70016, services, 1_790_244_000) + b"\x00" * 26 |
| + b"\x00" * 26 + struct.pack("<Q", 7) |
| + bytes([len(agent)]) + agent.encode() + struct.pack("<i", 850_000)) |
| return p2p_message("version", payload) |
|
|
|
|
| def test_pcap_reads_the_peers_version_message_for_agent_and_services(tmp_path): |
| packets = [ |
| (1_790_244_000.5, tcp_frame(PEER, NODE, 8333, 51000, |
| version_message("/Satoshi:27.0.0/", 0x409))), |
| (1_790_244_001.0, tcp_frame(PEER, NODE, 8333, 51000, |
| inv_message([(cr.INV_TX, TX_GENESIS)]))), |
| |
| (1_790_244_001.1, tcp_frame(NODE, "198.51.100.77", 51001, 8333, |
| version_message("/Satoshi:26.0.0/", 1))), |
| (1_790_244_001.2, tcp_frame(NODE, "198.51.100.77", 51001, 8333, |
| inv_message([(cr.INV_TX, TX_GENESIS)]))), |
| ] |
| events = cr.read_capture(write_pcap(tmp_path / "v.pcap", packets), local_ips=[NODE]) |
| theirs = next(e for e in events if e.peer_ip == PEER) |
| assert (theirs.user_agent, theirs.services) == ("/Satoshi:27.0.0/", 0x409) |
| ours = next(e for e in events if e.peer_ip == "198.51.100.77") |
| assert (ours.user_agent, ours.services) == (None, None) |
|
|
|
|
| def test_btcap_services_accept_an_integer_or_hex_and_debug_log_leaves_them_unknown(tmp_path): |
| path = tmp_path / "s.btcap" |
| path.write_text( |
| '{"txid": "ab", "message_type": "inv", "peer_ip": "203.0.113.1", ' |
| '"wall_clock_ts": 1790244000, "services": 1033}\n' |
| '{"txid": "cd", "message_type": "inv", "peer_ip": "203.0.113.2", ' |
| '"wall_clock_ts": 1790244001, "service_flags": "0x409"}\n') |
| assert [e.services for e in cr.read_capture(path)] == [1033, 0x409] |
| assert all(e.services is None for e in cr.read_capture(FIXTURES / "node1.debug.log")) |
|
|
