From 6af4c208ce38fab58d5f410f12b2d92d95050036 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 2 Sep 2026 22:34:12 +0300 Subject: [PATCH] test(perception): localize transport tails with bounded TLS record witnesses --- .../pilot_clock_route_probe.py | 90 ++++++++++- .../pilot_tls_record_trace.py | 143 ++++++++++++++++++ tests/test_perception_tls_record_trace.py | 96 ++++++++++++ 3 files changed, 326 insertions(+), 3 deletions(-) create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_tls_record_trace.py create mode 100644 tests/test_perception_tls_record_trace.py diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_clock_route_probe.py b/experiments/perception/worker/streaming_profile_stage1/pilot_clock_route_probe.py index 885ce5b..b094c12 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_clock_route_probe.py +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_clock_route_probe.py @@ -11,6 +11,7 @@ from pathlib import Path import grpc from pilot_network_control import read_control, write_control +from pilot_tls_record_trace import TlsRecordRelay from k1link.perception.realtime_contract import StreamStart from k1link.perception.streaming_clock import ClockMappingError, ClockWindow @@ -65,8 +66,14 @@ def assessment(own, peer, now): async def client(args, report): bootstrap = read_control(Path(args.bootstrap)) ticket = ControlTicket(StreamStart.from_dict(bootstrap["activation"]), bootstrap["token"]) + relay = None + target = args.target + if getattr(args, "trace_relay", False): + host, port = target.rsplit(":", 1) + relay = TlsRecordRelay(host, int(port)) + target = f"localhost:{await relay.start()}" client = StreamControlClient( - args.target, + target, Path(args.certificate).read_bytes(), ticket, clock_id="route-source-" + secrets.token_hex(12), @@ -74,6 +81,30 @@ async def client(args, report): window = ClockWindow( client.clock_id, bootstrap["remote_clock_id"], rate_ppm=500, timestamp_error_ns=50_000 ) + calls = report.setdefault("rpc_calls", []) + + def traced(call, method): + async def invoke(raw, **kwargs): + value = json.loads(raw) + nonce = value["nonce"] if method == "poll" else value["probe"]["nonce"] + if len(calls) >= 256: + raise ValueError("bounded RPC trace exceeded") + row = {"method": method, "nonce": nonce, "request_bytes": len(raw)} + calls.append(row) + row["invoked_ns"] = str(time.monotonic_ns()) + try: + response = await call(raw, **kwargs) + row["completed_ns"] = str(time.monotonic_ns()) + row["response_bytes"] = len(response) + return response + except BaseException: + row["failed_ns"] = str(time.monotonic_ns()) + raise + + return invoke + + client.call = traced(client.call, "poll") + client.report_call = traced(client.report_call, "report") try: for index in range(args.samples): probe, grant = await client.poll() @@ -102,6 +133,40 @@ async def client(args, report): await asyncio.sleep(0.05) finally: await client.close() + if relay: + await relay.close() + report["tls_records"] = relay.snapshot() + + +class TracedEndpoint(StreamControlEndpoint): + def __init__(self, *args, rpc_trace, **kwargs): + super().__init__(*args, **kwargs) + self.rpc_trace = rpc_trace + + async def _traced(self, method, handler, raw, context): + entered = time.monotonic_ns() + response = await handler(raw, context) + finished = time.monotonic_ns() + value = json.loads(response) + if len(self.rpc_trace) >= 256: + raise ValueError("bounded handler trace exceeded") + self.rpc_trace.append( + { + "method": method, + "nonce": value["nonce"], + "entered_ns": str(entered), + "returned_ns": str(finished), + "request_bytes": len(raw), + "response_bytes": len(response), + } + ) + return response + + async def poll(self, raw, context): + return await self._traced("poll", super().poll, raw, context) + + async def report_clock(self, raw, context): + return await self._traced("report", super().report_clock, raw, context) async def server(args, report): @@ -137,7 +202,13 @@ async def server(args, report): ) return result - endpoint = StreamControlEndpoint(ticket, lambda: None, clock_id=clock_id, observe=observe) + endpoint = TracedEndpoint( + ticket, + lambda: None, + clock_id=clock_id, + observe=observe, + rpc_trace=report.setdefault("rpc_calls", []), + ) rpc = grpc.aio.server( options=( ("grpc.max_send_message_length", MAX_CONTROL), @@ -149,10 +220,18 @@ async def server(args, report): credentials = grpc.ssl_server_credentials( ((Path(args.private_key).read_bytes(), Path(args.certificate).read_bytes()),) ) - if not rpc.add_secure_port(args.target, credentials): + traced = getattr(args, "trace_relay", False) + port = rpc.add_secure_port("127.0.0.1:0" if traced else args.target, credentials) + if not port: raise ValueError("clock diagnostic bind failed") await rpc.start() + relay = None try: + if traced: + host, external = args.target.rsplit(":", 1) + relay = TlsRecordRelay("127.0.0.1", port) + port = await relay.start(host, int(external)) + report["listen_port"] = port write_control( Path(args.bootstrap), { @@ -169,6 +248,9 @@ async def server(args, report): finally: Path(args.bootstrap).unlink(missing_ok=True) await rpc.stop(1) + if relay: + await relay.close() + report["tls_records"] = relay.snapshot() async def run(args): @@ -188,6 +270,7 @@ async def run(args): "actuation_allowed": False, "samples": [], "error": None, + "trace_relay": getattr(args, "trace_relay", False), } witness = LoopWitness() task = asyncio.create_task(witness.run()) @@ -219,6 +302,7 @@ if __name__ == "__main__": parser.add_argument("--private-key") parser.add_argument("--samples", type=int, default=96) parser.add_argument("--output", required=True) + parser.add_argument("--trace-relay", action="store_true") args = parser.parse_args() if not 8 <= args.samples <= 128 or (args.role == "server" and not args.private_key): parser.error("bounded 8..128 samples and server private key required") diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_tls_record_trace.py b/experiments/perception/worker/streaming_profile_stage1/pilot_tls_record_trace.py new file mode 100644 index 0000000..913fa0c --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_tls_record_trace.py @@ -0,0 +1,143 @@ +"""Bounded diagnostic relay: correlate encrypted TLS records, never decrypt/log payloads. + +For isolated CPU attribution only. This relay is NOT a runtime transport replacement. +Receipt and drain timestamps describe application boundaries, not kernel ACK/wire time. +""" + +import asyncio +import hashlib +import time +from contextlib import suppress + + +class RecordWitness: + def __init__(self, *, max_bytes=8 * 1024 * 1024, max_records=4096): + self.max_bytes, self.max_records = max_bytes, max_records + self.buffer = bytearray() + self.rows = [] + self.total = self.offset = 0 + self.first_ns = None + + def feed(self, raw, received_ns, drained_ns): + if self.total + len(raw) > self.max_bytes: + raise ValueError("bounded encrypted stream exceeded") + self.total += len(raw) + self.buffer.extend(raw) + if self.first_ns is None: + self.first_ns = received_ns + while len(self.buffer) >= 5: + kind, major, minor = self.buffer[:3] + size = int.from_bytes(self.buffer[3:5], "big") + if kind not in (20, 21, 22, 23) or major != 3 or minor > 3 or size > 18432: + raise ValueError("invalid bounded TLS record") + length = size + 5 + if len(self.buffer) < length: + return + if len(self.rows) >= self.max_records: + raise ValueError("bounded TLS record ledger exceeded") + self.rows.append( + { + "offset": self.offset, + "bytes": length, + "type": kind, + "sha256": hashlib.sha256(memoryview(self.buffer)[:length]).hexdigest(), + "first_received_ns": str(self.first_ns), + "received_ns": str(received_ns), + "drained_ns": str(drained_ns), + } + ) + del self.buffer[:length] + self.offset += length + self.first_ns = received_ns if self.buffer else None + + def snapshot(self): + return {"bytes": self.total, "incomplete_bytes": len(self.buffer), "records": self.rows} + + +class TlsRecordRelay: + """At most two local diagnostic connections; no listener outside its owner scope.""" + + def __init__(self, target_host, target_port): + self.target = (target_host, target_port) + self.server = None + self.connections = [] + self.tasks = set() + self.writers = set() + + async def start(self, host="127.0.0.1", port=0): + self.server = await asyncio.start_server(self._accept, host, port, limit=64 * 1024) + return self.server.sockets[0].getsockname()[1] + + async def _accept(self, reader, writer): + task = asyncio.current_task() + self.tasks.add(task) + self.writers.add(writer) + row = {"request": RecordWitness(), "response": RecordWitness(), "error": None} + if len(self.connections) >= 2: + writer.close() + self.tasks.remove(task) + self.writers.remove(writer) + return + self.connections.append(row) + peer = None + jobs = [] + try: + remote, peer = await asyncio.wait_for(asyncio.open_connection(*self.target), 2) + self.writers.add(peer) + + async def pipe(source, sink, witness): + while True: + raw = await asyncio.wait_for(source.read(64 * 1024), 10) + received = time.monotonic_ns() + if not raw: + if sink.can_write_eof(): + sink.write_eof() + return + sink.write(raw) + await asyncio.wait_for(sink.drain(), 0.5) + drained = time.monotonic_ns() + witness.feed(raw, received, drained) + + jobs = [ + asyncio.create_task(pipe(reader, peer, row["request"])), + asyncio.create_task(pipe(remote, writer, row["response"])), + ] + await asyncio.gather(*jobs) + except asyncio.CancelledError: + row["error"] = "cancelled-on-close" + raise + except Exception as exc: + row["error"] = type(exc).__name__ + finally: + for job in jobs: + job.cancel() + await asyncio.gather(*jobs, return_exceptions=True) + for stream in (writer, peer): + if stream is not None: + stream.close() + with suppress(OSError, TimeoutError): + await asyncio.wait_for(stream.wait_closed(), 1) + self.writers.discard(stream) + self.tasks.discard(task) + + async def close(self): + if self.server: + self.server.close() + await self.server.wait_closed() + if self.tasks: + _, pending = await asyncio.wait(self.tasks, timeout=0.3) + for task in pending: + task.cancel() + await asyncio.gather(*pending, return_exceptions=True) + for writer in self.writers: + writer.close() + + def snapshot(self): + return [ + { + "request": row["request"].snapshot(), + "response": row["response"].snapshot(), + "error": row["error"], + } + for row in self.connections + ] diff --git a/tests/test_perception_tls_record_trace.py b/tests/test_perception_tls_record_trace.py new file mode 100644 index 0000000..5764339 --- /dev/null +++ b/tests/test_perception_tls_record_trace.py @@ -0,0 +1,96 @@ +"""Bounded diagnostic attribution; sparse local TLS checks, no source/model workload.""" + +# ruff: noqa: F811 -- imported shared fixtures +import asyncio +import hashlib +import importlib +from types import SimpleNamespace + +import pytest +from test_perception_network_graph_adapter import adapter # noqa: F401 +from test_perception_streaming_grpc import tls # noqa: F401 + + +def record(payload=b"opaque ciphertext"): + return b"\x17\x03\x03" + len(payload).to_bytes(2, "big") + payload + + +@pytest.mark.parametrize("cut", [1, 4, 5, 9, 21]) +def test_record_hash_identity_survives_arbitrary_tcp_chunks(adapter, cut): + module = importlib.import_module("pilot_tls_record_trace") + raw = record() + witness = module.RecordWitness() + witness.feed(raw[:cut], 10, 12) + witness.feed(raw[cut:] + raw, 20, 24) + result = witness.snapshot() + assert result["bytes"] == 2 * len(raw) and result["incomplete_bytes"] == 0 + assert [r["offset"] for r in result["records"]] == [0, len(raw)] + assert all(r["sha256"] == hashlib.sha256(raw).hexdigest() for r in result["records"]) + assert result["records"][0]["first_received_ns"] == "10" + assert result["records"][0]["received_ns"] == "20" + assert result["records"][1]["first_received_ns"] == "20" + assert "ciphertext" not in str(result) + + +@pytest.mark.parametrize("fault", ["bytes", "records", "header", "record_length"]) +def test_trace_limits_fail_explicitly(adapter, fault): + module = importlib.import_module("pilot_tls_record_trace") + witness = module.RecordWitness(max_bytes=5 if fault == "bytes" else 1024, max_records=1) + raw = record() * 2 + if fault == "header": + raw = b"\x00\x03\x03\x00\x00" + elif fault == "record_length": + raw = b"\x17\x03\x03\xff\xff" + with pytest.raises(ValueError, match="bounded"): + witness.feed(raw, 1, 2) + + +def test_two_relays_match_encrypted_records_and_original_clock_calls(tmp_path, tls, adapter): + module = importlib.import_module("pilot_clock_route_probe") + + async def check(): + cert, key = tmp_path / "cert.pem", tmp_path / "key.pem" + cert.write_bytes(tls[0]) + key.write_bytes(tls[1]) + args = SimpleNamespace( + target="127.0.0.1:0", + bootstrap=tmp_path / "bootstrap.json", + certificate=cert, + private_key=key, + samples=8, + trace_relay=True, + ) + server_report, client_report = {"samples": []}, {"samples": []} + task = asyncio.create_task(module.server(args, server_report)) + try: + async with asyncio.timeout(2): + while "listen_port" not in server_report or not args.bootstrap.exists(): + await asyncio.sleep(0.01) + client_args = SimpleNamespace(**vars(args)) + client_args.target = f"localhost:{server_report['listen_port']}" + await module.client(client_args, client_report) + await asyncio.wait_for(task, 2) + assert server_report["accepted"] == 8 and server_report["rejected"] == 0 + assert len(client_report["rpc_calls"]) == len(server_report["rpc_calls"]) == 16 + assert not args.bootstrap.exists() + for client, server in zip( + client_report["rpc_calls"], server_report["rpc_calls"], strict=True + ): + assert (client["method"], client["nonce"]) == (server["method"], server["nonce"]) + assert int(client["invoked_ns"]) < int(server["entered_ns"]) + assert int(server["returned_ns"]) < int(client["completed_ns"]) + a, b = client_report["tls_records"], server_report["tls_records"] + assert len(a) == len(b) == 1 + assert a[0]["error"] is b[0]["error"] is None + for direction in ("request", "response"): + x, y = a[0][direction], b[0][direction] + assert x["incomplete_bytes"] == y["incomplete_bytes"] == 0 + assert [(r["offset"], r["sha256"]) for r in x["records"]] == [ + (r["offset"], r["sha256"]) for r in y["records"] + ] + assert len(x["records"]) >= 16 + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + asyncio.run(check())