pytest: add fixture for checking packet sizes.
What changed, and why it matters
This commit only adds a new test fixture and one test to the project's test suite. It lets developers capture network traffic during tests and check that TCP packet payloads stay a constant size. It does not change the actual Core Lightning software that users run, so it cannot introduce a security vulnerability or fix one in production code.
No security action needed. Treat as a normal testing-infrastructure change. Reviewers may want to confirm the test's xfail status and that it does not accidentally require elevated privileges in CI.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The diff adds pytest infrastructure (have_pcap_tools, tcp_capture fixtures and a TcpCapture helper class) and a single xfail-marked test, test_constant_packet_size. The helper uses dumpcap/tshark to record loopback traffic and asserts that all captured TCP payload lengths are identical. The test is currently expected to fail (pytest.mark.xfail(strict=True)) and is skipped when pcap tools are unavailable. No production code, protocol logic, or cryptographic paths are modified.
Changed components
tests/fixtures.pytests/test_connection.pyInspect captured patch +135 / −0
diff --git a/tests/fixtures.py b/tests/fixtures.py
index d3a4a114..7ff7e26d 100644
--- a/tests/fixtures.py
+++ b/tests/fixtures.py
@@ -7,6 +7,10 @@ from pathlib import Path
import os
import pytest
import re
+import shutil
+import subprocess
+import tempfile
+import time
@pytest.fixture
@@ -82,3 +86,108 @@ def compat():
def is_compat(version):
compat = CompatLevel()
return compat(version)
+
+
+def dumpcap_usable():
+ def have_binary(name):
+ return shutil.which(name) is not None
+
+ if not have_binary("dumpcap") or not have_binary("tshark"):
+ return False
+
+ try:
+ with tempfile.TemporaryDirectory() as td:
+ pcap = Path(td) / "probe.pcap"
+
+ proc = subprocess.Popen(
+ [
+ "dumpcap",
+ "-i", "lo",
+ "-w", str(pcap),
+ "-f", "tcp",
+ ],
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ )
+
+ time.sleep(0.2)
+ proc.terminate()
+ proc.wait(timeout=1)
+
+ return pcap.exists() and pcap.stat().st_size > 0
+ except (PermissionError, subprocess.SubprocessError, OSError):
+ return False
+
+
+@pytest.fixture(scope="session")
+def have_pcap_tools():
+ if not dumpcap_usable():
+ pytest.skip("dumpcap/tshark not available or insufficient privileges")
+
+
+class TcpCapture:
+ def __init__(self, tmpdir):
+ self.tmpdir = Path(tmpdir)
+ self.pcap = self.tmpdir / "traffic.pcap"
+ self.proc = None
+ self.port = None
+
+ def start(self, port):
+ assert self.proc is None, "capture already started"
+ self.port = int(port)
+
+ self.proc = subprocess.Popen(
+ [
+ "dumpcap",
+ "-i", "lo",
+ "-w", str(self.pcap),
+ "-f", f"tcp port {self.port}",
+ ],
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ )
+
+ # allow filter attach
+ time.sleep(0.2)
+
+ def stop(self):
+ if self.proc:
+ self.proc.terminate()
+ self.proc.wait(timeout=2)
+ self.proc = None
+
+ def assert_constant_payload(self):
+ tshark_cmd = [
+ "tshark",
+ "-r", str(self.pcap),
+ "-Y", "tcp.len > 0",
+ "-T", "fields",
+ "-e", "tcp.len",
+ ]
+
+ out = subprocess.check_output(tshark_cmd, text=True)
+ lengths = [int(x) for x in out.splitlines() if x.strip()]
+
+ assert lengths, f"No TCP payload packets captured on port {self.port}"
+
+ uniq = set(lengths)
+ assert len(uniq) == 1, (
+ f"Non-constant TCP payload sizes on port {self.port}: "
+ f"{sorted(uniq)}:"
+ + subprocess.check_output(["tshark", "-r", str(self.pcap)], text=True)
+ )
+
+
+@pytest.fixture
+def tcp_capture(have_pcap_tools, tmp_path):
+ # You will need permissions. Most distributions have a group which has
+ # permissions to use dumpcap:
+ # $ ls -l /usr/bin/dumpcap
+ # -rwxr-xr-- 1 root wireshark 229112 Apr 16 2024 /usr/bin/dumpcap
+ # $ getcap /usr/bin/dumpcap
+ # /usr/bin/dumpcap cap_net_admin,cap_net_raw=eip
+ # So you just need to be in the wireshark group.
+ cap = TcpCapture(tmp_path)
+ yield cap
+ cap.stop()
+ cap.assert_constant_payload()
diff --git a/tests/test_connection.py b/tests/test_connection.py
index 20fa6259..b93179f6 100644
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -4771,3 +4771,29 @@ def test_networkevents(node_factory, executor):
'type': 'connect_fail'},
{'created_index': 7,
'type': 'connect'}]}
+
+
+@pytest.mark.xfail(strict=True)
+def test_constant_packet_size(node_factory, tcp_capture):
+ """
+ Test that TCP packets between nodes are constant size. This will be skipped unless
+ you can run `dumpcap` (usually means you have to be in the `wireshark` group).
+ """
+ l1, l2, l3, l4 = node_factory.get_nodes(4)
+
+ # Encrypted setup BOLT 8 has some short packets.
+ l1.connect(l2)
+ l2.connect(l3)
+
+ tcp_capture.start(l2.port)
+
+ # This gives us gossip send a recv, and channel establishment.
+ node_factory.join_nodes([l1, l2, l3, l4], wait_for_announce=True)
+
+ # Forwarding, incoming and outgoing payments.
+ for src, dest in (l1, l2), (l2, l3), (l1, l4):
+ inv = dest.rpc.invoice(10000000, "test_constant_packet_size", "test_constant_packet_size")
+ src.rpc.xpay(inv['bolt11'])
+
+ # Padding pings don't elicit a response
+ assert not l2.daemon.is_in_log("connectd: Unexpected pong")
Why this scored 15/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.