From 3e344716e0d3da3e7e914e10da1979ad0c38efe0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=83=AD=E5=90=89=E6=B5=A9?= <1625567290@qq.com> Date: Fri, 14 Aug 2026 23:11:26 +0800 Subject: [PATCH 01/12] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E4=BB=85?= =?UTF-8?q?=E9=80=9A=E8=BF=87=20next=5Flayer=20=E5=A5=91=E7=BA=A6=E5=85=B3?= =?UTF-8?q?=E9=97=AD=E5=8E=9F=E5=A7=8B=E9=9A=A7=E9=81=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/eval/harbor/egress_filter.py | 38 ++++++--------- packages/eval/harbor/test_egress_filter.py | 57 +++++++++++----------- 2 files changed, 42 insertions(+), 53 deletions(-) diff --git a/packages/eval/harbor/egress_filter.py b/packages/eval/harbor/egress_filter.py index d86ae896f8..d1779c44c1 100644 --- a/packages/eval/harbor/egress_filter.py +++ b/packages/eval/harbor/egress_filter.py @@ -120,8 +120,12 @@ def public_trajectory_repository(host: str, path_query: str) -> bool: try: from mitmproxy.proxy import commands as proxy_commands + from mitmproxy.proxy.layer import Layer + from mitmproxy.proxy.layers.tcp import TCPLayer except ImportError: proxy_commands = None + Layer = object + TCPLayer = None def request(flow: object) -> None: @@ -136,21 +140,9 @@ def http_connect(flow: object) -> None: apply_http_policy(flow, raw_url) -def tcp_start(flow: object) -> None: - record_raw_tunnel(flow) - kill_flow(flow) - - -def tcp_message(flow: object) -> None: - messages = getattr(flow, "messages", None) - if messages: - messages[-1].content = b"" - kill_flow(flow) - - def next_layer(nextlayer: object) -> None: current = getattr(nextlayer, "layer", None) - if current is None or type(current).__name__ != "TCPLayer": + if TCPLayer is None or not isinstance(current, TCPLayer): return context = getattr(nextlayer, "context", None) record_raw_tunnel(context) @@ -215,18 +207,16 @@ def record_raw_tunnel(flow: object) -> None: pass -def kill_flow(flow: object) -> None: - kill = getattr(flow, "kill", None) - if callable(kill) and getattr(flow, "killable", True): - try: - kill() - except Exception: - pass - - -class CloseRawLayer: +class CloseRawLayer(Layer): def __init__(self, context: object) -> None: - self.context = context + if Layer is object: + self.context = context + return + if getattr(context, "layers", None) is None: + context.layers = [] + if getattr(context, "options", None) is None: + context.options = type("Options", (), {"proxy_debug": False})() + super().__init__(context) def handle_event(self, event: object): if proxy_commands is None: diff --git a/packages/eval/harbor/test_egress_filter.py b/packages/eval/harbor/test_egress_filter.py index 548f9cec79..4f0486fa73 100644 --- a/packages/eval/harbor/test_egress_filter.py +++ b/packages/eval/harbor/test_egress_filter.py @@ -175,46 +175,45 @@ def make(status, body, headers): MODULE.http_connect(allowed) self.assertIsNone(allowed.response, host) - def test_tcp_start_kills_raw_tunnels_and_records_them(self) -> None: - with tempfile.TemporaryDirectory() as directory: - MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl" - killed: list[str] = [] - flow = type( - "Flow", - (), - {"server_conn": SimpleNamespace(address=("ssh.github.com", 443))}, - )() - flow.kill = lambda: killed.append("killed") - MODULE.tcp_start(flow) - self.assertEqual(killed, ["killed"]) - record = json.loads(MODULE.AUDIT_PATH.read_text().splitlines()[0]) - self.assertEqual(record["ruleId"], "raw_tunnel") - self.assertEqual(record["host"], "ssh.github.com") - self.assertEqual(record["normalizedPath"], ":443") - - def test_tcp_message_drops_raw_payloads(self) -> None: - message = SimpleNamespace(content=b"SSH-2.0-test\r\n") - killed: list[str] = [] - flow = SimpleNamespace(messages=[message], killable=True) - flow.kill = lambda: killed.append("killed") - MODULE.tcp_message(flow) - self.assertEqual(message.content, b"") - self.assertEqual(killed, ["killed"]) - def test_next_layer_replaces_raw_tcp_with_a_closer(self) -> None: - class TCPLayer: + class FakeTCPLayer: pass + class CloseConnection: + def __init__(self, connection: object) -> None: + self.connection = connection + with tempfile.TemporaryDirectory() as directory: MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl" - context = SimpleNamespace(server=SimpleNamespace(address=("ssh.github.com", 443))) - nextlayer = SimpleNamespace(layer=TCPLayer(), context=context) + previous_tcp = MODULE.TCPLayer + previous_commands = MODULE.proxy_commands + self.addCleanup(setattr, MODULE, "TCPLayer", previous_tcp) + self.addCleanup(setattr, MODULE, "proxy_commands", previous_commands) + MODULE.TCPLayer = FakeTCPLayer + MODULE.proxy_commands = SimpleNamespace(CloseConnection=CloseConnection) + client = object() + server = SimpleNamespace(address=("ssh.github.com", 443)) + context = SimpleNamespace(client=client, server=server, layers=[], options=None) + nextlayer = SimpleNamespace(layer=FakeTCPLayer(), context=context) MODULE.next_layer(nextlayer) self.assertIsInstance(nextlayer.layer, MODULE.CloseRawLayer) + self.assertIsInstance(nextlayer.layer, MODULE.Layer) + commands = list(nextlayer.layer.handle_event(object())) + self.assertEqual([command.connection for command in commands], [client, server]) + self.assertTrue(all(isinstance(command, CloseConnection) for command in commands)) record = json.loads(MODULE.AUDIT_PATH.read_text().splitlines()[0]) self.assertEqual(record["ruleId"], "raw_tunnel") self.assertEqual(record["host"], "ssh.github.com") + def test_next_layer_leaves_non_tcp_layers_alone(self) -> None: + class HTTPLayer: + pass + + original = HTTPLayer() + nextlayer = SimpleNamespace(layer=original, context=SimpleNamespace()) + MODULE.next_layer(nextlayer) + self.assertIs(nextlayer.layer, original) + def test_audit_escapes_line_separators_so_python_and_typescript_agree(self) -> None: with tempfile.TemporaryDirectory() as directory: MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl" From 6dfaea44def680f022dc2cdcb6198978b4a5b5f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=83=AD=E5=90=89=E6=B5=A9?= <1625567290@qq.com> Date: Sat, 15 Aug 2026 11:14:51 +0800 Subject: [PATCH 02/12] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E4=BF=9D?= =?UTF-8?q?=E7=95=99=20TCP=20=E9=92=A9=E5=AD=90=E7=9B=B4=E5=88=B0=E5=B1=82?= =?UTF-8?q?=E5=88=86=E7=B1=BB=E5=AE=8C=E6=88=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/eval/harbor/egress_filter.py | 43 +++++++++++++++++++- packages/eval/harbor/test_egress_filter.py | 47 ++++++++++++++++++++-- 2 files changed, 86 insertions(+), 4 deletions(-) diff --git a/packages/eval/harbor/egress_filter.py b/packages/eval/harbor/egress_filter.py index d1779c44c1..caa37ad792 100644 --- a/packages/eval/harbor/egress_filter.py +++ b/packages/eval/harbor/egress_filter.py @@ -140,13 +140,30 @@ def http_connect(flow: object) -> None: apply_http_policy(flow, raw_url) +def tcp_start(flow: object) -> None: + # Script next_layer runs before the built-in classifier assigns + # nextlayer.layer, so CONNECT raw tunnels and HTTP 101 upgrades are + # closed here — not by replacing a TCPLayer that does not exist yet. + record_raw_tunnel(flow) + kill_flow(flow) + + +def tcp_message(flow: object) -> None: + messages = getattr(flow, "messages", None) + if messages: + messages[-1].content = b"" + kill_flow(flow) + + def next_layer(nextlayer: object) -> None: current = getattr(nextlayer, "layer", None) if TCPLayer is None or not isinstance(current, TCPLayer): return context = getattr(nextlayer, "context", None) record_raw_tunnel(context) - nextlayer.layer = CloseRawLayer(context) + closer = CloseRawLayer(context) + replace_layer(context, current, closer) + nextlayer.layer = closer def apply_http_policy(flow: object, raw_url: str) -> None: @@ -207,6 +224,30 @@ def record_raw_tunnel(flow: object) -> None: pass +def kill_flow(flow: object) -> None: + kill = getattr(flow, "kill", None) + if callable(kill) and getattr(flow, "killable", True): + try: + kill() + except Exception: + pass + + +def replace_layer(context: object, current: object, closer: object) -> None: + layers = getattr(context, "layers", None) + if not isinstance(layers, list): + return + try: + index = layers.index(current) + except ValueError: + index = len(layers) + if closer in layers: + layers.remove(closer) + if current in layers: + layers.remove(current) + layers.insert(min(index, len(layers)), closer) + + class CloseRawLayer(Layer): def __init__(self, context: object) -> None: if Layer is object: diff --git a/packages/eval/harbor/test_egress_filter.py b/packages/eval/harbor/test_egress_filter.py index 4f0486fa73..357e56e996 100644 --- a/packages/eval/harbor/test_egress_filter.py +++ b/packages/eval/harbor/test_egress_filter.py @@ -175,9 +175,37 @@ def make(status, body, headers): MODULE.http_connect(allowed) self.assertIsNone(allowed.response, host) + def test_tcp_start_kills_raw_tunnels_and_records_them(self) -> None: + with tempfile.TemporaryDirectory() as directory: + MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl" + killed: list[str] = [] + flow = type( + "Flow", + (), + {"server_conn": SimpleNamespace(address=("ssh.github.com", 443))}, + )() + flow.kill = lambda: killed.append("killed") + MODULE.tcp_start(flow) + self.assertEqual(killed, ["killed"]) + record = json.loads(MODULE.AUDIT_PATH.read_text().splitlines()[0]) + self.assertEqual(record["ruleId"], "raw_tunnel") + self.assertEqual(record["host"], "ssh.github.com") + self.assertEqual(record["normalizedPath"], ":443") + + def test_tcp_message_drops_raw_payloads(self) -> None: + message = SimpleNamespace(content=b"SSH-2.0-test\r\n") + killed: list[str] = [] + flow = SimpleNamespace(messages=[message], killable=True) + flow.kill = lambda: killed.append("killed") + MODULE.tcp_message(flow) + self.assertEqual(message.content, b"") + self.assertEqual(killed, ["killed"]) + def test_next_layer_replaces_raw_tcp_with_a_closer(self) -> None: class FakeTCPLayer: - pass + def __init__(self, context: object) -> None: + self.context = context + context.layers.append(self) class CloseConnection: def __init__(self, connection: object) -> None: @@ -193,11 +221,17 @@ def __init__(self, connection: object) -> None: MODULE.proxy_commands = SimpleNamespace(CloseConnection=CloseConnection) client = object() server = SimpleNamespace(address=("ssh.github.com", 443)) - context = SimpleNamespace(client=client, server=server, layers=[], options=None) - nextlayer = SimpleNamespace(layer=FakeTCPLayer(), context=context) + sibling = object() + context = SimpleNamespace( + client=client, server=server, layers=[sibling], options=None + ) + current = FakeTCPLayer(context) + nextlayer = SimpleNamespace(layer=current, context=context) MODULE.next_layer(nextlayer) self.assertIsInstance(nextlayer.layer, MODULE.CloseRawLayer) self.assertIsInstance(nextlayer.layer, MODULE.Layer) + self.assertEqual(context.layers, [sibling, nextlayer.layer]) + self.assertNotIn(current, context.layers) commands = list(nextlayer.layer.handle_event(object())) self.assertEqual([command.connection for command in commands], [client, server]) self.assertTrue(all(isinstance(command, CloseConnection) for command in commands)) @@ -205,6 +239,13 @@ def __init__(self, connection: object) -> None: self.assertEqual(record["ruleId"], "raw_tunnel") self.assertEqual(record["host"], "ssh.github.com") + def test_next_layer_leaves_an_unclassified_layer_alone(self) -> None: + context = SimpleNamespace(layers=[]) + nextlayer = SimpleNamespace(layer=None, context=context) + MODULE.next_layer(nextlayer) + self.assertIsNone(nextlayer.layer) + self.assertEqual(context.layers, []) + def test_next_layer_leaves_non_tcp_layers_alone(self) -> None: class HTTPLayer: pass From 64e305c3c996976c60609b4a3a0f80b65142acdd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=83=AD=E5=90=89=E6=B5=A9?= <1625567290@qq.com> Date: Sat, 15 Aug 2026 11:44:01 +0800 Subject: [PATCH 03/12] =?UTF-8?q?=E6=B5=8B=E8=AF=95=EF=BC=9A=E4=BD=BF?= =?UTF-8?q?=E7=94=A8=E7=9C=9F=E5=AE=9E=20mitmproxy=20=E8=A6=86=E7=9B=96?= =?UTF-8?q?=E5=8E=9F=E5=A7=8B=E9=9A=A7=E9=81=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../eval/harbor/test_egress_filter_live.py | 388 ++++++++++++++++++ packages/eval/package.json | 3 +- 2 files changed, 390 insertions(+), 1 deletion(-) create mode 100644 packages/eval/harbor/test_egress_filter_live.py diff --git a/packages/eval/harbor/test_egress_filter_live.py b/packages/eval/harbor/test_egress_filter_live.py new file mode 100644 index 0000000000..6e61d4e520 --- /dev/null +++ b/packages/eval/harbor/test_egress_filter_live.py @@ -0,0 +1,388 @@ +"""Prove raw tunnels die on a real mitmproxy 12.2.3 without breaking HTTPS or WebSocket. + +Unit tests cannot see addon order: script `next_layer` runs before the built-in +classifier assigns `TCPLayer`. This test starts the pinned proxy image and a +local origin, then asserts what a live cell would observe. + +It needs Docker and the pinned image, so it is opt-in: + + MAKA_EVAL_EGRESS_PROXY_TEST=1 python3 harbor/test_egress_filter_live.py +""" + +from __future__ import annotations + +import base64 +import json +import os +import shutil +import socket +import subprocess +import tempfile +import time +import unittest +from pathlib import Path + +HARBOR_DIR = Path(__file__).parent +PROXY_IMAGE = "maka-eval-egress-proxy:12.2.3" +PROJECT = "maka-eval-egress-proxy-live" +NETWORK = f"{PROJECT}_net" +PROXY = f"{PROJECT}-proxy" +ORIGIN = f"{PROJECT}-origin" +ORIGIN_IMAGE = "python:3.12-slim" +PROXY_PORT = 18081 +COMMAND_TIMEOUT_S = 60 + +ORIGIN_SCRIPT = r""" +import base64, hashlib, json, socket, threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +stats = {"raw_recv": 0, "upgrade_recv": 0} + +class HttpHandler(BaseHTTPRequestHandler): + def do_GET(self): + self.send_response(200) + self.send_header("Content-Type", "text/plain") + self.end_headers() + self.wfile.write(b"http-ok") + def log_message(self, format, *args): + return + +class WsHandler(BaseHTTPRequestHandler): + def do_GET(self): + if self.headers.get("Upgrade", "").lower() != "websocket": + self.send_error(400) + return + key = self.headers.get("Sec-WebSocket-Key", "") + accept = base64.b64encode( + hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()).digest() + ).decode() + self.send_response(101, "Switching Protocols") + self.send_header("Upgrade", "websocket") + self.send_header("Connection", "Upgrade") + self.send_header("Sec-WebSocket-Accept", accept) + self.end_headers() + payload = b"ws-ok" + self.wfile.write(bytes([0x81, len(payload)]) + payload) + self.wfile.flush() + def log_message(self, format, *args): + return + +class UpgradeHandler(BaseHTTPRequestHandler): + def do_GET(self): + self.send_response(101, "Switching Protocols") + self.send_header("Upgrade", "raw") + self.send_header("Connection", "Upgrade") + self.end_headers() + self.wfile.write(b"UPGRADE-BANNER\n") + self.wfile.flush() + self.connection.settimeout(2) + try: + stats["upgrade_recv"] += len(self.connection.recv(64) or b"") + except OSError: + pass + def log_message(self, format, *args): + return + +def serve_raw(): + sock = socket.socket() + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind(("0.0.0.0", 19081)) + sock.listen(8) + while True: + conn, _ = sock.accept() + conn.settimeout(3) + try: + conn.sendall(b"RAW-BANNER\n") + stats["raw_recv"] += len(conn.recv(64) or b"") + except OSError: + pass + finally: + conn.close() + +class StatsHandler(BaseHTTPRequestHandler): + def do_GET(self): + body = json.dumps(stats).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(body) + def log_message(self, format, *args): + return + +threading.Thread(target=serve_raw, daemon=True).start() +threading.Thread(target=lambda: ThreadingHTTPServer(("0.0.0.0", 19080), HttpHandler).serve_forever(), daemon=True).start() +threading.Thread(target=lambda: ThreadingHTTPServer(("0.0.0.0", 19082), WsHandler).serve_forever(), daemon=True).start() +threading.Thread(target=lambda: ThreadingHTTPServer(("0.0.0.0", 19083), UpgradeHandler).serve_forever(), daemon=True).start() +ThreadingHTTPServer(("0.0.0.0", 19084), StatsHandler).serve_forever() +""" + + +def docker_available() -> bool: + if shutil.which("docker") is None: + return False + probe = subprocess.run( + ["docker", "version", "--format", "{{.Server.Version}}"], + capture_output=True, + timeout=COMMAND_TIMEOUT_S, + ) + return probe.returncode == 0 + + +@unittest.skipUnless( + os.environ.get("MAKA_EVAL_EGRESS_PROXY_TEST") == "1", + "set MAKA_EVAL_EGRESS_PROXY_TEST=1 to run the live mitmproxy proxy test", +) +class LiveEgressFilterTest(unittest.TestCase): + workdir: Path | None = None + + @classmethod + def setUpClass(cls) -> None: + if not docker_available(): + raise unittest.SkipTest("Docker daemon is unavailable") + images = subprocess.run( + ["docker", "image", "inspect", PROXY_IMAGE], + capture_output=True, + timeout=COMMAND_TIMEOUT_S, + ) + if images.returncode != 0: + raise unittest.SkipTest(f"{PROXY_IMAGE} is not present") + cls.workdir = Path(tempfile.mkdtemp(prefix="maka-eval-egress-proxy-live-")) + (cls.workdir / "origin.py").write_text(ORIGIN_SCRIPT) + cls.addClassCleanup(shutil.rmtree, cls.workdir, ignore_errors=True) + cls.addClassCleanup(cls._down) + cls._down() + subprocess.run(["docker", "network", "create", NETWORK], check=True, timeout=COMMAND_TIMEOUT_S) + subprocess.run( + [ + "docker", + "run", + "-d", + "--name", + ORIGIN, + "--network", + NETWORK, + "--network-alias", + "origin", + "-v", + f"{cls.workdir / 'origin.py'}:/origin.py:ro", + ORIGIN_IMAGE, + "python", + "/origin.py", + ], + check=True, + timeout=COMMAND_TIMEOUT_S, + ) + subprocess.run( + [ + "docker", + "run", + "-d", + "--name", + PROXY, + "--network", + NETWORK, + "-p", + f"127.0.0.1:{PROXY_PORT}:8080", + "-v", + f"{HARBOR_DIR / 'egress_filter.py'}:/opt/maka-eval/egress_filter.py:ro", + PROXY_IMAGE, + ], + check=True, + timeout=COMMAND_TIMEOUT_S, + ) + cls._wait_for_proxy() + cls._wait_for_origin_via_proxy() + + @classmethod + def _down(cls) -> None: + for name in (PROXY, ORIGIN): + subprocess.run( + ["docker", "rm", "-f", name], + capture_output=True, + timeout=COMMAND_TIMEOUT_S, + ) + subprocess.run( + ["docker", "network", "rm", NETWORK], + capture_output=True, + timeout=COMMAND_TIMEOUT_S, + ) + + @classmethod + def _wait_for_proxy(cls) -> None: + deadline = time.time() + 30 + last_error = "proxy did not listen" + while time.time() < deadline: + try: + with socket.create_connection(("127.0.0.1", PROXY_PORT), 1): + return + except OSError as error: + last_error = str(error) + time.sleep(0.2) + raise AssertionError(last_error) + + @classmethod + def _wait_for_origin_via_proxy(cls) -> None: + deadline = time.time() + 30 + last_error = "origin was not reachable through the proxy" + while time.time() < deadline: + try: + body = cls.http_via_proxy(19080) + except OSError as error: + last_error = str(error) + time.sleep(0.2) + continue + if b"http-ok" in body: + return + last_error = body[:200].decode("latin1", errors="replace") + time.sleep(0.2) + raise AssertionError(last_error) + + @classmethod + def _recv_all(cls, sock: socket.socket, limit: int = 8192) -> bytes: + sock.settimeout(6) + data = b"" + try: + while len(data) < limit: + chunk = sock.recv(4096) + if not chunk: + break + data += chunk + except (TimeoutError, ConnectionResetError, BrokenPipeError): + pass + return data + + @classmethod + def http_via_proxy(cls, port: int, extra_headers: str = "") -> bytes: + request = ( + f"GET http://origin:{port}/ HTTP/1.1\r\n" + f"Host: origin:{port}\r\n" + f"{extra_headers}" + "Connection: close\r\n\r\n" + ).encode() + with socket.create_connection(("127.0.0.1", PROXY_PORT), 5) as sock: + sock.sendall(request) + return cls._recv_all(sock) + + @classmethod + def connect_via_proxy(cls, host: str, port: int, payload: bytes = b"CLIENT\n") -> tuple[bytes, bytes]: + with socket.create_connection(("127.0.0.1", PROXY_PORT), 5) as sock: + sock.sendall(f"CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\n\r\n".encode()) + header = b"" + sock.settimeout(6) + try: + while b"\r\n\r\n" not in header: + chunk = sock.recv(4096) + if not chunk: + break + header += chunk + except (TimeoutError, ConnectionResetError): + return header, b"" + try: + sock.sendall(payload) + except OSError: + pass + return header, cls._recv_all(sock) + + @classmethod + def audit_records(cls) -> list[dict[str, object]]: + listed = subprocess.run( + ["docker", "exec", PROXY, "cat", "/opt/maka-egress-state/hits.jsonl"], + capture_output=True, + text=True, + timeout=COMMAND_TIMEOUT_S, + ) + return [json.loads(line) for line in listed.stdout.splitlines() if line.strip()] + + @classmethod + def origin_stats(cls) -> dict[str, int]: + # Read stats from inside the origin container so the probe does not + # depend on the proxy remaining willing to forward that port. + listed = subprocess.run( + [ + "docker", + "exec", + ORIGIN, + "python", + "-c", + "import urllib.request; print(urllib.request.urlopen('http://127.0.0.1:19084/').read().decode())", + ], + capture_output=True, + text=True, + timeout=COMMAND_TIMEOUT_S, + ) + if listed.returncode != 0: + raise AssertionError(listed.stderr) + return json.loads(listed.stdout) + + def test_https_and_plain_http_still_forward(self) -> None: + http = self.http_via_proxy(19080) + self.assertIn(b"http-ok", http) + curl = subprocess.run( + [ + "curl", + "--silent", + "--show-error", + "--max-time", + "20", + "--proxy", + f"http://127.0.0.1:{PROXY_PORT}", + "--insecure", + "--output", + "/dev/null", + "--write-out", + "%{http_code}", + "https://example.com/", + ], + capture_output=True, + text=True, + timeout=COMMAND_TIMEOUT_S, + ) + self.assertEqual(curl.stdout, "200", curl.stderr) + + def test_connect_to_a_blocklisted_host_is_451(self) -> None: + header, _ = self.connect_via_proxy("tbench.ai", 443, b"") + self.assertIn(b"451", header.split(b"\r\n", 1)[0]) + self.assertIn(b"tbench_domain", header) + + def test_raw_connect_relays_no_bytes_and_is_audited(self) -> None: + header, body = self.connect_via_proxy("origin", 19081) + self.assertIn(b"200", header.split(b"\r\n", 1)[0]) + self.assertNotIn(b"RAW-BANNER", body) + self.assertEqual(body, b"") + self.assertEqual(self.origin_stats()["raw_recv"], 0) + self.assertIn( + {"host": "origin", "normalizedPath": ":19081", "ruleId": "raw_tunnel"}, + [{key: record.get(key) for key in ("host", "normalizedPath", "ruleId")} for record in self.audit_records()], + ) + + def test_http_101_raw_upgrade_is_audited_without_relaying_the_banner(self) -> None: + data = self.http_via_proxy(19083) + self.assertIn(b"101", data.split(b"\r\n", 1)[0]) + self.assertNotIn(b"UPGRADE-BANNER", data) + self.assertEqual(self.origin_stats()["upgrade_recv"], 0) + self.assertIn( + {"host": "origin", "normalizedPath": ":19083", "ruleId": "raw_tunnel"}, + [{key: record.get(key) for key in ("host", "normalizedPath", "ruleId")} for record in self.audit_records()], + ) + + def test_websocket_upgrade_still_completes(self) -> None: + key = base64.b64encode(b"0123456789abcdef").decode() + data = self.http_via_proxy( + 19082, + extra_headers=( + "Upgrade: websocket\r\n" + "Connection: Upgrade\r\n" + f"Sec-WebSocket-Key: {key}\r\n" + "Sec-WebSocket-Version: 13\r\n" + ), + ) + self.assertIn(b"101", data.split(b"\r\n", 1)[0]) + self.assertIn(b"ws-ok", data) + self.assertNotIn( + ":19082", + [str(record.get("normalizedPath")) for record in self.audit_records()], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/eval/package.json b/packages/eval/package.json index 67e03a827a..fd9ff81490 100644 --- a/packages/eval/package.json +++ b/packages/eval/package.json @@ -34,7 +34,8 @@ "clean": "node ../../scripts/clean-paths.mjs dist tsconfig.tsbuildinfo", "build": "tsc -p tsconfig.json", "typecheck": "tsc -p tsconfig.json --noEmit", - "test:dist": "node --test \"dist/**/*.test.js\" && python3 harbor/test_eval_framework.py && python3 harbor/test_relay_contract.py && python3 harbor/test_relay_lifecycle.py && python3 harbor/test_egress_filter.py && python3 harbor/test_run_trial_policy.py && python3 harbor/test_relay_artifacts.py && python3 harbor/test_cell_egress_namespace.py" +<<<<<<< HEAD + "test:dist": "node --test \"dist/**/*.test.js\" && python3 harbor/test_eval_framework.py && python3 harbor/test_relay_contract.py && python3 harbor/test_relay_lifecycle.py && python3 harbor/test_egress_filter.py && python3 harbor/test_run_trial_policy.py && python3 harbor/test_relay_artifacts.py && python3 harbor/test_cell_egress_namespace.py && python3 harbor/test_egress_filter_live.py" }, "dependencies": { "@maka/core": "0.1.0", From f84af61b81866c6b4dd42f468bd4f8135ab71ad2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=83=AD=E5=90=89=E6=B5=A9?= <1625567290@qq.com> Date: Sun, 16 Aug 2026 12:25:17 +0800 Subject: [PATCH 04/12] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E7=94=B1?= =?UTF-8?q?=E5=A5=97=E6=8E=A5=E5=AD=97=E6=89=80=E6=9C=89=E5=B1=82=E5=85=B3?= =?UTF-8?q?=E9=97=AD=E5=8E=9F=E5=A7=8B=E9=9A=A7=E9=81=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../eval/harbor/egress-proxy/entrypoint.sh | 1 + packages/eval/harbor/egress_filter.py | 81 ++++++++++- packages/eval/harbor/test_egress_filter.py | 87 ++++++++++++ .../eval/harbor/test_egress_filter_live.py | 130 +++++++++++------- 4 files changed, 246 insertions(+), 53 deletions(-) diff --git a/packages/eval/harbor/egress-proxy/entrypoint.sh b/packages/eval/harbor/egress-proxy/entrypoint.sh index 13edc0873c..21dde6ccec 100755 --- a/packages/eval/harbor/egress-proxy/entrypoint.sh +++ b/packages/eval/harbor/egress-proxy/entrypoint.sh @@ -57,5 +57,6 @@ exec mitmdump \ --listen-host 0.0.0.0 \ --listen-port 8080 \ --set block_global=false \ + --set rawtcp=false \ --set confdir="$STATE_DIR" \ --scripts /opt/maka-eval/egress_filter.py diff --git a/packages/eval/harbor/egress_filter.py b/packages/eval/harbor/egress_filter.py index caa37ad792..b61d26d4a0 100644 --- a/packages/eval/harbor/egress_filter.py +++ b/packages/eval/harbor/egress_filter.py @@ -127,11 +127,45 @@ def public_trajectory_repository(host: str, path_query: str) -> bool: Layer = object TCPLayer = None +try: + from mitmproxy.net.tls import starts_like_tls_record +except ImportError: + def starts_like_tls_record(data: bytes) -> bool: + return len(data) >= 3 and data[0] == 0x16 and data[1] == 0x03 + + +def configure(updated: object) -> None: + # HTTP 101 upgrades construct TCPLayer inside the HTTP layer without + # another next_layer hook. rawtcp=false makes that path CloseConnection + # itself; WebSocket upgrades stay on the websocket layer. + try: + from mitmproxy import ctx + except ImportError: + return + if getattr(ctx.options, "rawtcp", False): + ctx.options.rawtcp = False + def request(flow: object) -> None: apply_http_policy(flow, flow.request.pretty_url) +def response(flow: object) -> None: + response = getattr(flow, "response", None) + if getattr(response, "status_code", None) != 101: + return + headers = getattr(response, "headers", None) + upgrade = "" + if headers is not None: + try: + upgrade = str(headers.get("upgrade", "") or "") + except Exception: + upgrade = "" + if upgrade.lower() == "websocket": + return + record_raw_tunnel(flow) + + def http_connect(flow: object) -> None: try: raw_url = connect_target_url(flow) @@ -141,9 +175,9 @@ def http_connect(flow: object) -> None: def tcp_start(flow: object) -> None: - # Script next_layer runs before the built-in classifier assigns - # nextlayer.layer, so CONNECT raw tunnels and HTTP 101 upgrades are - # closed here — not by replacing a TCPLayer that does not exist yet. + # Last resort if a TCPLayer is still admitted (tcp_hosts / ignore). + # CONNECT raw is closed by next_layer → CloseRawLayer; HTTP 101 raw is + # closed by rawtcp=false. Neither of those paths starts a TCPLayer. record_raw_tunnel(flow) kill_flow(flow) @@ -156,10 +190,19 @@ def tcp_message(flow: object) -> None: def next_layer(nextlayer: object) -> None: + # Script addons run before the built-in classifier assigns layer. If we + # set CloseRawLayer here, NextLayer leaves it in place. Waiting for + # isinstance(..., TCPLayer) never fires on the production CONNECT path. current = getattr(nextlayer, "layer", None) + context = getattr(nextlayer, "context", None) + if current is None: + if not looks_like_raw_tcp(nextlayer): + return + record_raw_tunnel(context) + nextlayer.layer = CloseRawLayer(context) + return if TCPLayer is None or not isinstance(current, TCPLayer): return - context = getattr(nextlayer, "context", None) record_raw_tunnel(context) closer = CloseRawLayer(context) replace_layer(context, current, closer) @@ -206,6 +249,36 @@ def connect_target_url(flow: object) -> str: return f"https://{host}:{port}/" +def looks_like_raw_tcp(nextlayer: object) -> bool: + """Match mitmproxy 12.2.3's raw-TCP classifier without waiting for TCPLayer.""" + data_client = _next_layer_bytes(nextlayer, "data_client") + data_server = _next_layer_bytes(nextlayer, "data_server") + if starts_like_tls_record(data_client): + return False + if not data_client and not data_server: + return False + probably_no_http = ( + len(data_client) < 3 + or b" " not in data_client + or (data_client.find(b" ") > data_client.find(b"\n")) + or not data_client[:3].isalpha() + or bool(data_server) + or data_client.startswith(b"SSH") + ) + return probably_no_http + + +def _next_layer_bytes(nextlayer: object, name: str) -> bytes: + getter = getattr(nextlayer, name, None) + if not callable(getter): + return b"" + try: + data = getter() + except Exception: + return b"" + return bytes(data) if isinstance(data, (bytes, bytearray)) else b"" + + def tcp_peer(flow: object) -> tuple[str, str]: server = getattr(flow, "server_conn", None) or getattr(flow, "server", None) address = getattr(server, "address", None) if server is not None else None diff --git a/packages/eval/harbor/test_egress_filter.py b/packages/eval/harbor/test_egress_filter.py index 357e56e996..5ac21173a8 100644 --- a/packages/eval/harbor/test_egress_filter.py +++ b/packages/eval/harbor/test_egress_filter.py @@ -201,6 +201,93 @@ def test_tcp_message_drops_raw_payloads(self) -> None: self.assertEqual(message.content, b"") self.assertEqual(killed, ["killed"]) + def test_next_layer_closes_raw_tcp_before_the_builtin_classifier(self) -> None: + class CloseConnection: + def __init__(self, connection: object) -> None: + self.connection = connection + + with tempfile.TemporaryDirectory() as directory: + MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl" + previous_commands = MODULE.proxy_commands + self.addCleanup(setattr, MODULE, "proxy_commands", previous_commands) + MODULE.proxy_commands = SimpleNamespace(CloseConnection=CloseConnection) + client = object() + server = SimpleNamespace(address=("ssh.github.com", 443)) + context = SimpleNamespace(client=client, server=server, layers=[], options=None) + + def data_client() -> bytes: + return b"SSH-2.0-OpenSSH_9.0" + + nextlayer = SimpleNamespace( + layer=None, context=context, data_client=data_client, data_server=lambda: b"" + ) + MODULE.next_layer(nextlayer) + self.assertIsInstance(nextlayer.layer, MODULE.CloseRawLayer) + commands = list(nextlayer.layer.handle_event(object())) + self.assertEqual([command.connection for command in commands], [client, server]) + record = json.loads(MODULE.AUDIT_PATH.read_text().splitlines()[0]) + self.assertEqual(record["ruleId"], "raw_tunnel") + self.assertEqual(record["host"], "ssh.github.com") + + def test_next_layer_leaves_tls_and_http_for_the_builtin_classifier(self) -> None: + context = SimpleNamespace(layers=[]) + tls = SimpleNamespace( + layer=None, + context=context, + data_client=lambda: b"\x16\x03\x01\x00\x00", + data_server=lambda: b"", + ) + MODULE.next_layer(tls) + self.assertIsNone(tls.layer) + + http = SimpleNamespace( + layer=None, + context=context, + data_client=lambda: b"GET / HTTP/1.1\r\n", + data_server=lambda: b"", + ) + MODULE.next_layer(http) + self.assertIsNone(http.layer) + + empty = SimpleNamespace( + layer=None, + context=context, + data_client=lambda: b"", + data_server=lambda: b"", + ) + MODULE.next_layer(empty) + self.assertIsNone(empty.layer) + + def test_response_audits_a_non_websocket_101_upgrade(self) -> None: + with tempfile.TemporaryDirectory() as directory: + MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl" + raw = type( + "Flow", + (), + { + "response": SimpleNamespace( + status_code=101, headers={"upgrade": "raw"} + ), + "server_conn": SimpleNamespace(address=("origin", 19083)), + }, + )() + MODULE.response(raw) + record = json.loads(MODULE.AUDIT_PATH.read_text().splitlines()[0]) + self.assertEqual(record["ruleId"], "raw_tunnel") + + websocket = type( + "Flow", + (), + { + "response": SimpleNamespace( + status_code=101, headers={"upgrade": "websocket"} + ), + "server_conn": SimpleNamespace(address=("origin", 19082)), + }, + )() + MODULE.response(websocket) + self.assertEqual(len(MODULE.AUDIT_PATH.read_text().splitlines()), 1) + def test_next_layer_replaces_raw_tcp_with_a_closer(self) -> None: class FakeTCPLayer: def __init__(self, context: object) -> None: diff --git a/packages/eval/harbor/test_egress_filter_live.py b/packages/eval/harbor/test_egress_filter_live.py index 6e61d4e520..e3ebbf3b29 100644 --- a/packages/eval/harbor/test_egress_filter_live.py +++ b/packages/eval/harbor/test_egress_filter_live.py @@ -20,17 +20,14 @@ import tempfile import time import unittest +import uuid from pathlib import Path HARBOR_DIR = Path(__file__).parent PROXY_IMAGE = "maka-eval-egress-proxy:12.2.3" -PROJECT = "maka-eval-egress-proxy-live" -NETWORK = f"{PROJECT}_net" -PROXY = f"{PROJECT}-proxy" -ORIGIN = f"{PROJECT}-origin" ORIGIN_IMAGE = "python:3.12-slim" -PROXY_PORT = 18081 COMMAND_TIMEOUT_S = 60 +CLOSE_TIMEOUT_S = 2.0 ORIGIN_SCRIPT = r""" import base64, hashlib, json, socket, threading @@ -75,9 +72,13 @@ def do_GET(self): self.end_headers() self.wfile.write(b"UPGRADE-BANNER\n") self.wfile.flush() - self.connection.settimeout(2) + self.connection.settimeout(60) try: - stats["upgrade_recv"] += len(self.connection.recv(64) or b"") + while True: + chunk = self.connection.recv(64) + if not chunk: + break + stats["upgrade_recv"] += len(chunk) except OSError: pass def log_message(self, format, *args): @@ -90,10 +91,14 @@ def serve_raw(): sock.listen(8) while True: conn, _ = sock.accept() - conn.settimeout(3) try: conn.sendall(b"RAW-BANNER\n") - stats["raw_recv"] += len(conn.recv(64) or b"") + conn.settimeout(60) + while True: + chunk = conn.recv(64) + if not chunk: + break + stats["raw_recv"] += len(chunk) except OSError: pass finally: @@ -134,6 +139,10 @@ def docker_available() -> bool: ) class LiveEgressFilterTest(unittest.TestCase): workdir: Path | None = None + network: str = "" + proxy: str = "" + origin: str = "" + proxy_port: int = 0 @classmethod def setUpClass(cls) -> None: @@ -146,21 +155,24 @@ def setUpClass(cls) -> None: ) if images.returncode != 0: raise unittest.SkipTest(f"{PROXY_IMAGE} is not present") + run_id = f"{os.getpid()}-{uuid.uuid4().hex[:8]}" + cls.network = f"maka-eval-egress-live-{run_id}-net" + cls.proxy = f"maka-eval-egress-live-{run_id}-proxy" + cls.origin = f"maka-eval-egress-live-{run_id}-origin" cls.workdir = Path(tempfile.mkdtemp(prefix="maka-eval-egress-proxy-live-")) (cls.workdir / "origin.py").write_text(ORIGIN_SCRIPT) cls.addClassCleanup(shutil.rmtree, cls.workdir, ignore_errors=True) cls.addClassCleanup(cls._down) - cls._down() - subprocess.run(["docker", "network", "create", NETWORK], check=True, timeout=COMMAND_TIMEOUT_S) + subprocess.run(["docker", "network", "create", cls.network], check=True, timeout=COMMAND_TIMEOUT_S) subprocess.run( [ "docker", "run", "-d", "--name", - ORIGIN, + cls.origin, "--network", - NETWORK, + cls.network, "--network-alias", "origin", "-v", @@ -178,11 +190,11 @@ def setUpClass(cls) -> None: "run", "-d", "--name", - PROXY, + cls.proxy, "--network", - NETWORK, + cls.network, "-p", - f"127.0.0.1:{PROXY_PORT}:8080", + "127.0.0.1::8080", "-v", f"{HARBOR_DIR / 'egress_filter.py'}:/opt/maka-eval/egress_filter.py:ro", PROXY_IMAGE, @@ -190,22 +202,40 @@ def setUpClass(cls) -> None: check=True, timeout=COMMAND_TIMEOUT_S, ) + cls.proxy_port = cls._published_port() cls._wait_for_proxy() cls._wait_for_origin_via_proxy() + @classmethod + def _published_port(cls) -> int: + listed = subprocess.run( + ["docker", "port", cls.proxy, "8080/tcp"], + capture_output=True, + text=True, + timeout=COMMAND_TIMEOUT_S, + ) + if listed.returncode != 0: + raise AssertionError(listed.stderr) + # "127.0.0.1:49152" + hostport = listed.stdout.strip().splitlines()[0] + return int(hostport.rsplit(":", 1)[1]) + @classmethod def _down(cls) -> None: - for name in (PROXY, ORIGIN): + for name in (cls.proxy, cls.origin): + if not name: + continue subprocess.run( ["docker", "rm", "-f", name], capture_output=True, timeout=COMMAND_TIMEOUT_S, ) - subprocess.run( - ["docker", "network", "rm", NETWORK], - capture_output=True, - timeout=COMMAND_TIMEOUT_S, - ) + if cls.network: + subprocess.run( + ["docker", "network", "rm", cls.network], + capture_output=True, + timeout=COMMAND_TIMEOUT_S, + ) @classmethod def _wait_for_proxy(cls) -> None: @@ -213,7 +243,7 @@ def _wait_for_proxy(cls) -> None: last_error = "proxy did not listen" while time.time() < deadline: try: - with socket.create_connection(("127.0.0.1", PROXY_PORT), 1): + with socket.create_connection(("127.0.0.1", cls.proxy_port), 1): return except OSError as error: last_error = str(error) @@ -238,17 +268,17 @@ def _wait_for_origin_via_proxy(cls) -> None: raise AssertionError(last_error) @classmethod - def _recv_all(cls, sock: socket.socket, limit: int = 8192) -> bytes: - sock.settimeout(6) + def _recv_until_close(cls, sock: socket.socket, limit: int = 8192) -> bytes: + sock.settimeout(CLOSE_TIMEOUT_S) data = b"" - try: - while len(data) < limit: + while len(data) < limit: + try: chunk = sock.recv(4096) - if not chunk: - break - data += chunk - except (TimeoutError, ConnectionResetError, BrokenPipeError): - pass + except ConnectionResetError: + return data + if not chunk: + return data + data += chunk return data @classmethod @@ -259,34 +289,35 @@ def http_via_proxy(cls, port: int, extra_headers: str = "") -> bytes: f"{extra_headers}" "Connection: close\r\n\r\n" ).encode() - with socket.create_connection(("127.0.0.1", PROXY_PORT), 5) as sock: + with socket.create_connection(("127.0.0.1", cls.proxy_port), 5) as sock: sock.sendall(request) - return cls._recv_all(sock) + return cls._recv_until_close(sock) @classmethod def connect_via_proxy(cls, host: str, port: int, payload: bytes = b"CLIENT\n") -> tuple[bytes, bytes]: - with socket.create_connection(("127.0.0.1", PROXY_PORT), 5) as sock: + with socket.create_connection(("127.0.0.1", cls.proxy_port), 5) as sock: sock.sendall(f"CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\n\r\n".encode()) header = b"" - sock.settimeout(6) - try: - while b"\r\n\r\n" not in header: - chunk = sock.recv(4096) - if not chunk: - break - header += chunk - except (TimeoutError, ConnectionResetError): - return header, b"" + sock.settimeout(CLOSE_TIMEOUT_S) + while b"\r\n\r\n" not in header: + chunk = sock.recv(4096) + if not chunk: + return header, b"" + header += chunk + leftover = header.split(b"\r\n\r\n", 1)[1] + header = header[: header.index(b"\r\n\r\n") + 4] + if not payload: + return header, leftover try: sock.sendall(payload) except OSError: - pass - return header, cls._recv_all(sock) + return header, leftover + return header, leftover + cls._recv_until_close(sock) @classmethod def audit_records(cls) -> list[dict[str, object]]: listed = subprocess.run( - ["docker", "exec", PROXY, "cat", "/opt/maka-egress-state/hits.jsonl"], + ["docker", "exec", cls.proxy, "cat", "/opt/maka-egress-state/hits.jsonl"], capture_output=True, text=True, timeout=COMMAND_TIMEOUT_S, @@ -301,7 +332,7 @@ def origin_stats(cls) -> dict[str, int]: [ "docker", "exec", - ORIGIN, + cls.origin, "python", "-c", "import urllib.request; print(urllib.request.urlopen('http://127.0.0.1:19084/').read().decode())", @@ -322,10 +353,11 @@ def test_https_and_plain_http_still_forward(self) -> None: "curl", "--silent", "--show-error", + "--http1.1", "--max-time", "20", "--proxy", - f"http://127.0.0.1:{PROXY_PORT}", + f"http://127.0.0.1:{self.proxy_port}", "--insecure", "--output", "/dev/null", From 830151c24fc54e03e6154b8427183a0c57a8341e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=83=AD=E5=90=89=E6=B5=A9?= <1625567290@qq.com> Date: Mon, 17 Aug 2026 17:16:17 +0800 Subject: [PATCH 05/12] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E6=8E=92?= =?UTF-8?q?=E9=99=A4=20HTTP=20=E5=89=8D=E7=BC=80=E5=90=8E=E5=86=8D?= =?UTF-8?q?=E5=85=B3=E9=97=AD=E5=8E=9F=E5=A7=8B=20TCP?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/eval/harbor/egress_filter.py | 31 +++++++++++++++------- packages/eval/harbor/test_egress_filter.py | 26 ++++++++++++++++++ packages/eval/package.json | 1 - 3 files changed, 47 insertions(+), 11 deletions(-) diff --git a/packages/eval/harbor/egress_filter.py b/packages/eval/harbor/egress_filter.py index b61d26d4a0..307fc08d0d 100644 --- a/packages/eval/harbor/egress_filter.py +++ b/packages/eval/harbor/egress_filter.py @@ -250,22 +250,33 @@ def connect_target_url(flow: object) -> str: def looks_like_raw_tcp(nextlayer: object) -> bool: - """Match mitmproxy 12.2.3's raw-TCP classifier without waiting for TCPLayer.""" + """Close only when the bytes cannot still become HTTP or TLS. + + Script next_layer runs before mitmproxy 12.2.3's classifier. Copying its + `probably_no_http` test here would treat `GET` / `GET ` as raw and assign + CloseRawLayer while the request line is still arriving. With rawtcp=false + the built-in path would have kept waiting for HttpLayer. + """ data_client = _next_layer_bytes(nextlayer, "data_client") data_server = _next_layer_bytes(nextlayer, "data_server") if starts_like_tls_record(data_client): return False if not data_client and not data_server: return False - probably_no_http = ( - len(data_client) < 3 - or b" " not in data_client - or (data_client.find(b" ") > data_client.find(b"\n")) - or not data_client[:3].isalpha() - or bool(data_server) - or data_client.startswith(b"SSH") - ) - return probably_no_http + if data_server or data_client.startswith(b"SSH"): + return True + return not _still_could_be_http(data_client) + + +def _still_could_be_http(data: bytes) -> bool: + first_line, newline, _rest = data.partition(b"\n") + line = first_line.rstrip(b"\r") + method, space, _remainder = line.partition(b" ") + if not method.isascii() or not method.isalpha(): + return False + if newline and not space: + return False + return True def _next_layer_bytes(nextlayer: object, name: str) -> bytes: diff --git a/packages/eval/harbor/test_egress_filter.py b/packages/eval/harbor/test_egress_filter.py index 5ac21173a8..6749f9d810 100644 --- a/packages/eval/harbor/test_egress_filter.py +++ b/packages/eval/harbor/test_egress_filter.py @@ -258,6 +258,32 @@ def test_next_layer_leaves_tls_and_http_for_the_builtin_classifier(self) -> None MODULE.next_layer(empty) self.assertIsNone(empty.layer) + for prefix in (b"G", b"GE", b"GET", b"GET ", b"GET / HTTP/1.1"): + with self.subTest(prefix=prefix): + incomplete = SimpleNamespace( + layer=None, + context=context, + data_client=lambda prefix=prefix: prefix, + data_server=lambda: b"", + ) + MODULE.next_layer(incomplete) + self.assertIsNone(incomplete.layer) + + def test_next_layer_closes_bytes_that_cannot_become_http(self) -> None: + with tempfile.TemporaryDirectory() as directory: + MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl" + context = SimpleNamespace( + layers=[], options=None, client=object(), server=None + ) + binary = SimpleNamespace( + layer=None, + context=context, + data_client=lambda: b"\x00\x01\x02\x03", + data_server=lambda: b"", + ) + MODULE.next_layer(binary) + self.assertIsInstance(binary.layer, MODULE.CloseRawLayer) + def test_response_audits_a_non_websocket_101_upgrade(self) -> None: with tempfile.TemporaryDirectory() as directory: MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl" diff --git a/packages/eval/package.json b/packages/eval/package.json index fd9ff81490..362404f120 100644 --- a/packages/eval/package.json +++ b/packages/eval/package.json @@ -34,7 +34,6 @@ "clean": "node ../../scripts/clean-paths.mjs dist tsconfig.tsbuildinfo", "build": "tsc -p tsconfig.json", "typecheck": "tsc -p tsconfig.json --noEmit", -<<<<<<< HEAD "test:dist": "node --test \"dist/**/*.test.js\" && python3 harbor/test_eval_framework.py && python3 harbor/test_relay_contract.py && python3 harbor/test_relay_lifecycle.py && python3 harbor/test_egress_filter.py && python3 harbor/test_run_trial_policy.py && python3 harbor/test_relay_artifacts.py && python3 harbor/test_cell_egress_namespace.py && python3 harbor/test_egress_filter_live.py" }, "dependencies": { From c15d66ab9f76d3cd2031d4d228137609b929680c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=83=AD=E5=90=89=E6=B5=A9?= <1625567290@qq.com> Date: Mon, 17 Aug 2026 18:05:34 +0800 Subject: [PATCH 06/12] =?UTF-8?q?=E6=B5=8B=E8=AF=95=EF=BC=9A=E7=BC=BA?= =?UTF-8?q?=E5=B0=91=E6=BA=90=E9=95=9C=E5=83=8F=E6=97=B6=E8=AE=A9=E4=BB=A3?= =?UTF-8?q?=E7=90=86=E5=AE=9E=E6=B5=8B=E5=A4=B1=E8=B4=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../eval/harbor/test_egress_filter_live.py | 22 +++++++++++++------ 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/packages/eval/harbor/test_egress_filter_live.py b/packages/eval/harbor/test_egress_filter_live.py index e3ebbf3b29..64b1063658 100644 --- a/packages/eval/harbor/test_egress_filter_live.py +++ b/packages/eval/harbor/test_egress_filter_live.py @@ -4,7 +4,7 @@ classifier assigns `TCPLayer`. This test starts the pinned proxy image and a local origin, then asserts what a live cell would observe. -It needs Docker and the pinned image, so it is opt-in: +It needs Docker, the pinned proxy image, and `python:3.12-slim`, so it is opt-in: MAKA_EVAL_EGRESS_PROXY_TEST=1 python3 harbor/test_egress_filter_live.py """ @@ -133,6 +133,15 @@ def docker_available() -> bool: return probe.returncode == 0 +def docker_image_present(image: str) -> bool: + inspect = subprocess.run( + ["docker", "image", "inspect", image], + capture_output=True, + timeout=COMMAND_TIMEOUT_S, + ) + return inspect.returncode == 0 + + @unittest.skipUnless( os.environ.get("MAKA_EVAL_EGRESS_PROXY_TEST") == "1", "set MAKA_EVAL_EGRESS_PROXY_TEST=1 to run the live mitmproxy proxy test", @@ -148,13 +157,10 @@ class LiveEgressFilterTest(unittest.TestCase): def setUpClass(cls) -> None: if not docker_available(): raise unittest.SkipTest("Docker daemon is unavailable") - images = subprocess.run( - ["docker", "image", "inspect", PROXY_IMAGE], - capture_output=True, - timeout=COMMAND_TIMEOUT_S, - ) - if images.returncode != 0: + if not docker_image_present(PROXY_IMAGE): raise unittest.SkipTest(f"{PROXY_IMAGE} is not present") + if not docker_image_present(ORIGIN_IMAGE): + raise unittest.SkipTest(f"{ORIGIN_IMAGE} is not present") run_id = f"{os.getpid()}-{uuid.uuid4().hex[:8]}" cls.network = f"maka-eval-egress-live-{run_id}-net" cls.proxy = f"maka-eval-egress-live-{run_id}-proxy" @@ -322,6 +328,8 @@ def audit_records(cls) -> list[dict[str, object]]: text=True, timeout=COMMAND_TIMEOUT_S, ) + if listed.returncode != 0: + raise AssertionError(listed.stderr) return [json.loads(line) for line in listed.stdout.splitlines() if line.strip()] @classmethod From 6435e0c8e7b082e48583972aab6c9d3f44c145b7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=83=AD=E5=90=89=E6=B5=A9?= <1625567290@qq.com> Date: Mon, 17 Aug 2026 18:14:11 +0800 Subject: [PATCH 07/12] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E5=AE=A1?= =?UTF-8?q?=E8=AE=A1=E6=9C=AA=E4=BF=9D=E6=8C=81=20WebSocket=20=E7=9A=84=20?= =?UTF-8?q?101=20=E5=8D=87=E7=BA=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/eval/harbor/egress_filter.py | 13 +++++-------- packages/eval/harbor/test_egress_filter.py | 18 ++++++++++++++++++ .../eval/harbor/test_egress_filter_live.py | 9 ++++++++- 3 files changed, 31 insertions(+), 9 deletions(-) diff --git a/packages/eval/harbor/egress_filter.py b/packages/eval/harbor/egress_filter.py index 307fc08d0d..288b76d3d5 100644 --- a/packages/eval/harbor/egress_filter.py +++ b/packages/eval/harbor/egress_filter.py @@ -154,14 +154,11 @@ def response(flow: object) -> None: response = getattr(flow, "response", None) if getattr(response, "status_code", None) != 101: return - headers = getattr(response, "headers", None) - upgrade = "" - if headers is not None: - try: - upgrade = str(headers.get("upgrade", "") or "") - except Exception: - upgrade = "" - if upgrade.lower() == "websocket": + # mitmproxy 12.2.3 sets flow.websocket before HttpResponseHook only when + # the 101 is a real WebSocket upgrade (Upgrade + version 13 + option on). + # A websocket Upgrade header alone still falls through to CloseConnection + # under rawtcp=false and must be audited. + if getattr(flow, "websocket", None) is not None: return record_raw_tunnel(flow) diff --git a/packages/eval/harbor/test_egress_filter.py b/packages/eval/harbor/test_egress_filter.py index 6749f9d810..5d580f935d 100644 --- a/packages/eval/harbor/test_egress_filter.py +++ b/packages/eval/harbor/test_egress_filter.py @@ -308,12 +308,30 @@ def test_response_audits_a_non_websocket_101_upgrade(self) -> None: "response": SimpleNamespace( status_code=101, headers={"upgrade": "websocket"} ), + "websocket": object(), "server_conn": SimpleNamespace(address=("origin", 19082)), }, )() MODULE.response(websocket) self.assertEqual(len(MODULE.AUDIT_PATH.read_text().splitlines()), 1) + invalid = type( + "Flow", + (), + { + "response": SimpleNamespace( + status_code=101, headers={"upgrade": "websocket"} + ), + "server_conn": SimpleNamespace(address=("origin", 19082)), + }, + )() + MODULE.response(invalid) + self.assertEqual(len(MODULE.AUDIT_PATH.read_text().splitlines()), 2) + self.assertEqual( + json.loads(MODULE.AUDIT_PATH.read_text().splitlines()[1])["ruleId"], + "raw_tunnel", + ) + def test_next_layer_replaces_raw_tcp_with_a_closer(self) -> None: class FakeTCPLayer: def __init__(self, context: object) -> None: diff --git a/packages/eval/harbor/test_egress_filter_live.py b/packages/eval/harbor/test_egress_filter_live.py index 64b1063658..cb975bf643 100644 --- a/packages/eval/harbor/test_egress_filter_live.py +++ b/packages/eval/harbor/test_egress_filter_live.py @@ -323,7 +323,14 @@ def connect_via_proxy(cls, host: str, port: int, payload: bytes = b"CLIENT\n") - @classmethod def audit_records(cls) -> list[dict[str, object]]: listed = subprocess.run( - ["docker", "exec", cls.proxy, "cat", "/opt/maka-egress-state/hits.jsonl"], + [ + "docker", + "exec", + cls.proxy, + "sh", + "-c", + "if [ -f /opt/maka-egress-state/hits.jsonl ]; then cat /opt/maka-egress-state/hits.jsonl; fi", + ], capture_output=True, text=True, timeout=COMMAND_TIMEOUT_S, From 2d8bb38573b8313deb49330e0245e440c684c8f8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=83=AD=E5=90=89=E6=B5=A9?= <1625567290@qq.com> Date: Sun, 23 Aug 2026 14:12:21 +0800 Subject: [PATCH 08/12] =?UTF-8?q?=E8=AE=A9=E5=87=BA=E5=8F=A3=E4=BB=A3?= =?UTF-8?q?=E7=90=86=E5=AE=9E=E6=B5=8B=E8=BF=9B=E5=85=A5=E6=8C=81=E7=BB=AD?= =?UTF-8?q?=E9=9B=86=E6=88=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/ci.yml | 12 ++++++++++++ packages/eval/package.json | 3 ++- scripts/ci-test-plan.test.mjs | 20 ++++++++++++++++++++ 3 files changed, 34 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2c7129c0de..f3367e20fe 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -192,6 +192,18 @@ jobs: fi node scripts/run-workspace-tests-parallel.mjs --concurrency=3 --workspaces="$WORKSPACES" + - name: Run live Eval egress proxy test + if: contains(steps.plan.outputs.standard_workspaces, 'packages/eval') + env: + MAKA_EVAL_EGRESS_PROXY_TEST: '1' + run: | + docker pull python:3.12-slim + docker build \ + --tag maka-eval-egress-proxy:12.2.3 \ + --file packages/eval/harbor/egress-proxy/Dockerfile \ + packages/eval/harbor + npm --workspace @maka/eval run test:egress-proxy:live + - name: Run Runtime Host tests if: steps.plan.outputs.runtime_host == 'true' run: npm --workspace @maka/runtime-host run test:dist diff --git a/packages/eval/package.json b/packages/eval/package.json index 362404f120..f23d7fce60 100644 --- a/packages/eval/package.json +++ b/packages/eval/package.json @@ -34,7 +34,8 @@ "clean": "node ../../scripts/clean-paths.mjs dist tsconfig.tsbuildinfo", "build": "tsc -p tsconfig.json", "typecheck": "tsc -p tsconfig.json --noEmit", - "test:dist": "node --test \"dist/**/*.test.js\" && python3 harbor/test_eval_framework.py && python3 harbor/test_relay_contract.py && python3 harbor/test_relay_lifecycle.py && python3 harbor/test_egress_filter.py && python3 harbor/test_run_trial_policy.py && python3 harbor/test_relay_artifacts.py && python3 harbor/test_cell_egress_namespace.py && python3 harbor/test_egress_filter_live.py" + "test:dist": "node --test \"dist/**/*.test.js\" && python3 harbor/test_eval_framework.py && python3 harbor/test_relay_contract.py && python3 harbor/test_relay_lifecycle.py && python3 harbor/test_egress_filter.py && python3 harbor/test_run_trial_policy.py && python3 harbor/test_relay_artifacts.py && python3 harbor/test_cell_egress_namespace.py", + "test:egress-proxy:live": "python3 harbor/test_egress_filter_live.py" }, "dependencies": { "@maka/core": "0.1.0", diff --git a/scripts/ci-test-plan.test.mjs b/scripts/ci-test-plan.test.mjs index 33f2eda812..5324bebe6d 100644 --- a/scripts/ci-test-plan.test.mjs +++ b/scripts/ci-test-plan.test.mjs @@ -383,6 +383,26 @@ test('workflows never persist the job credential into the checkout', () => { } }); +test('core CI runs the live Eval proxy lifecycle when Eval is selected', () => { + const workflow = readWorkflow('ci.yml'); + const evalPackage = JSON.parse( + readFileSync(new URL('../packages/eval/package.json', import.meta.url), 'utf8'), + ); + + assert.match( + workflow, + /if: contains\(steps\.plan\.outputs\.standard_workspaces, 'packages\/eval'\)/u, + ); + assert.match(workflow, /MAKA_EVAL_EGRESS_PROXY_TEST: '1'/u); + assert.match(workflow, /docker build[\s\S]*maka-eval-egress-proxy:12\.2\.3/u); + assert.match(workflow, /npm --workspace @maka\/eval run test:egress-proxy:live/u); + assert.equal( + evalPackage.scripts['test:egress-proxy:live'], + 'python3 harbor/test_egress_filter_live.py', + ); + assert.doesNotMatch(evalPackage.scripts['test:dist'], /test_egress_filter_live\.py/u); +}); + const WORKFLOW_DIR = new URL('../.github/workflows/', import.meta.url); function readWorkflow(name) { From 83ba226c2228cb0c2d55aab0e37ecb517251320d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=83=AD=E5=90=89=E6=B5=A9?= <1625567290@qq.com> Date: Sun, 23 Aug 2026 14:43:14 +0800 Subject: [PATCH 09/12] =?UTF-8?q?=E8=A1=A5=E5=85=85=E5=87=BA=E5=8F=A3?= =?UTF-8?q?=E4=BB=A3=E7=90=86=E5=AE=9E=E6=B5=8B=E8=AE=B8=E5=8F=AF=E5=A4=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/eval/harbor/test_egress_filter_live.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/packages/eval/harbor/test_egress_filter_live.py b/packages/eval/harbor/test_egress_filter_live.py index cb975bf643..9d81ddfd57 100644 --- a/packages/eval/harbor/test_egress_filter_live.py +++ b/packages/eval/harbor/test_egress_filter_live.py @@ -1,3 +1,20 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + """Prove raw tunnels die on a real mitmproxy 12.2.3 without breaking HTTPS or WebSocket. Unit tests cannot see addon order: script `next_layer` runs before the built-in From 02f45541273a4f990f483eef6ebb38defaeba89a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=83=AD=E5=90=89=E6=B5=A9?= <1625567290@qq.com> Date: Sun, 23 Aug 2026 20:32:39 +0800 Subject: [PATCH 10/12] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=87=BA=E5=8F=A3?= =?UTF-8?q?=E4=BB=A3=E7=90=86=E7=A2=8E=E7=89=87=E5=8C=96=20TLS=20=E4=B8=8E?= =?UTF-8?q?=E5=85=B3=E9=97=AD=E9=AA=8C=E8=AF=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/eval/harbor/egress_filter.py | 9 +- packages/eval/harbor/test_egress_filter.py | 18 ++++ .../eval/harbor/test_egress_filter_live.py | 101 ++++++++++++++++-- 3 files changed, 120 insertions(+), 8 deletions(-) diff --git a/packages/eval/harbor/egress_filter.py b/packages/eval/harbor/egress_filter.py index 288b76d3d5..c7c9e6b87c 100644 --- a/packages/eval/harbor/egress_filter.py +++ b/packages/eval/harbor/egress_filter.py @@ -256,7 +256,7 @@ def looks_like_raw_tcp(nextlayer: object) -> bool: """ data_client = _next_layer_bytes(nextlayer, "data_client") data_server = _next_layer_bytes(nextlayer, "data_server") - if starts_like_tls_record(data_client): + if _could_start_tls_record(data_client): return False if not data_client and not data_server: return False @@ -265,6 +265,13 @@ def looks_like_raw_tcp(nextlayer: object) -> bool: return not _still_could_be_http(data_client) +def _could_start_tls_record(data: bytes) -> bool: + """Keep a fragmented ClientHello undecided until its 3-byte prefix exists.""" + if starts_like_tls_record(data): + return True + return 0 < len(data) < 3 and b"\x16\x03".startswith(data) + + def _still_could_be_http(data: bytes) -> bool: first_line, newline, _rest = data.partition(b"\n") line = first_line.rstrip(b"\r") diff --git a/packages/eval/harbor/test_egress_filter.py b/packages/eval/harbor/test_egress_filter.py index 5d580f935d..60b3995526 100644 --- a/packages/eval/harbor/test_egress_filter.py +++ b/packages/eval/harbor/test_egress_filter.py @@ -240,6 +240,24 @@ def test_next_layer_leaves_tls_and_http_for_the_builtin_classifier(self) -> None MODULE.next_layer(tls) self.assertIsNone(tls.layer) + for first, remainder in ( + (b"\x16", b"\x03\x01\x00\x00"), + (b"\x16\x03", b"\x01\x00\x00"), + ): + with self.subTest(tls_prefix=first): + buffered = bytearray(first) + fragmented = SimpleNamespace( + layer=None, + context=context, + data_client=lambda: bytes(buffered), + data_server=lambda: b"", + ) + MODULE.next_layer(fragmented) + self.assertIsNone(fragmented.layer) + buffered.extend(remainder) + MODULE.next_layer(fragmented) + self.assertIsNone(fragmented.layer) + http = SimpleNamespace( layer=None, context=context, diff --git a/packages/eval/harbor/test_egress_filter_live.py b/packages/eval/harbor/test_egress_filter_live.py index 9d81ddfd57..4d1cad69f1 100644 --- a/packages/eval/harbor/test_egress_filter_live.py +++ b/packages/eval/harbor/test_egress_filter_live.py @@ -33,6 +33,7 @@ import os import shutil import socket +import ssl import subprocess import tempfile import time @@ -50,7 +51,7 @@ import base64, hashlib, json, socket, threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -stats = {"raw_recv": 0, "upgrade_recv": 0} +stats = {"raw_recv": 0, "raw_closed": 0, "upgrade_recv": 0, "upgrade_closed": 0} class HttpHandler(BaseHTTPRequestHandler): def do_GET(self): @@ -98,6 +99,8 @@ def do_GET(self): stats["upgrade_recv"] += len(chunk) except OSError: pass + finally: + stats["upgrade_closed"] += 1 def log_message(self, format, *args): return @@ -119,6 +122,7 @@ def serve_raw(): except OSError: pass finally: + stats["raw_closed"] += 1 conn.close() class StatsHandler(BaseHTTPRequestHandler): @@ -305,12 +309,17 @@ def _recv_until_close(cls, sock: socket.socket, limit: int = 8192) -> bytes: return data @classmethod - def http_via_proxy(cls, port: int, extra_headers: str = "") -> bytes: + def http_via_proxy( + cls, + port: int, + extra_headers: str = "", + connection: str = "close", + ) -> bytes: request = ( f"GET http://origin:{port}/ HTTP/1.1\r\n" f"Host: origin:{port}\r\n" f"{extra_headers}" - "Connection: close\r\n\r\n" + f"Connection: {connection}\r\n\r\n" ).encode() with socket.create_connection(("127.0.0.1", cls.proxy_port), 5) as sock: sock.sendall(request) @@ -337,6 +346,58 @@ def connect_via_proxy(cls, host: str, port: int, payload: bytes = b"CLIENT\n") - return header, leftover return header, leftover + cls._recv_until_close(sock) + @classmethod + def fragmented_tls_via_proxy(cls, first_fragment_size: int) -> str: + incoming = ssl.MemoryBIO() + outgoing = ssl.MemoryBIO() + tls = ssl.create_default_context().wrap_bio( + incoming, + outgoing, + server_side=False, + server_hostname="example.com", + ) + try: + tls.do_handshake() + except ssl.SSLWantReadError: + pass + client_hello = outgoing.read() + if len(client_hello) <= first_fragment_size: + raise AssertionError("TLS ClientHello was unexpectedly short") + + with socket.create_connection(("127.0.0.1", cls.proxy_port), 5) as sock: + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + sock.sendall(b"CONNECT example.com:443 HTTP/1.1\r\nHost: example.com:443\r\n\r\n") + header = b"" + while b"\r\n\r\n" not in header: + chunk = sock.recv(4096) + if not chunk: + raise AssertionError("proxy closed before the CONNECT response") + header += chunk + if b" 200 " not in header.split(b"\r\n", 1)[0]: + raise AssertionError(header.decode("latin1", errors="replace")) + + sock.sendall(client_hello[:first_fragment_size]) + time.sleep(0.05) + sock.sendall(client_hello[first_fragment_size:]) + sock.settimeout(10) + deadline = time.time() + 20 + while time.time() < deadline: + pending = outgoing.read() + if pending: + sock.sendall(pending) + try: + tls.do_handshake() + pending = outgoing.read() + if pending: + sock.sendall(pending) + return tls.version() or "" + except ssl.SSLWantReadError: + response = sock.recv(16 * 1024) + if not response: + raise AssertionError("proxy closed during the fragmented TLS handshake") + incoming.write(response) + raise AssertionError("fragmented TLS handshake did not complete") + @classmethod def audit_records(cls) -> list[dict[str, object]]: listed = subprocess.run( @@ -377,6 +438,17 @@ def origin_stats(cls) -> dict[str, int]: raise AssertionError(listed.stderr) return json.loads(listed.stdout) + @classmethod + def wait_for_origin_counter(cls, key: str, minimum: int) -> dict[str, int]: + deadline = time.time() + CLOSE_TIMEOUT_S + last = cls.origin_stats() + while time.time() < deadline: + if last.get(key, 0) >= minimum: + return last + time.sleep(0.05) + last = cls.origin_stats() + raise AssertionError(f"origin counter {key} did not reach {minimum}: {last}") + def test_https_and_plain_http_still_forward(self) -> None: http = self.http_via_proxy(19080) self.assertIn(b"http-ok", http) @@ -403,27 +475,42 @@ def test_https_and_plain_http_still_forward(self) -> None: ) self.assertEqual(curl.stdout, "200", curl.stderr) + def test_fragmented_tls_record_prefix_still_handshakes(self) -> None: + for first_fragment_size in (1, 2): + with self.subTest(first_fragment_size=first_fragment_size): + self.assertTrue( + self.fragmented_tls_via_proxy(first_fragment_size).startswith("TLS") + ) + def test_connect_to_a_blocklisted_host_is_451(self) -> None: header, _ = self.connect_via_proxy("tbench.ai", 443, b"") self.assertIn(b"451", header.split(b"\r\n", 1)[0]) self.assertIn(b"tbench_domain", header) def test_raw_connect_relays_no_bytes_and_is_audited(self) -> None: + closed_before = self.origin_stats()["raw_closed"] header, body = self.connect_via_proxy("origin", 19081) self.assertIn(b"200", header.split(b"\r\n", 1)[0]) self.assertNotIn(b"RAW-BANNER", body) self.assertEqual(body, b"") - self.assertEqual(self.origin_stats()["raw_recv"], 0) + stats = self.wait_for_origin_counter("raw_closed", closed_before + 1) + self.assertEqual(stats["raw_recv"], 0) self.assertIn( {"host": "origin", "normalizedPath": ":19081", "ruleId": "raw_tunnel"}, [{key: record.get(key) for key in ("host", "normalizedPath", "ruleId")} for record in self.audit_records()], ) def test_http_101_raw_upgrade_is_audited_without_relaying_the_banner(self) -> None: - data = self.http_via_proxy(19083) + closed_before = self.origin_stats()["upgrade_closed"] + data = self.http_via_proxy( + 19083, + extra_headers="Upgrade: raw\r\n", + connection="Upgrade", + ) self.assertIn(b"101", data.split(b"\r\n", 1)[0]) self.assertNotIn(b"UPGRADE-BANNER", data) - self.assertEqual(self.origin_stats()["upgrade_recv"], 0) + stats = self.wait_for_origin_counter("upgrade_closed", closed_before + 1) + self.assertEqual(stats["upgrade_recv"], 0) self.assertIn( {"host": "origin", "normalizedPath": ":19083", "ruleId": "raw_tunnel"}, [{key: record.get(key) for key in ("host", "normalizedPath", "ruleId")} for record in self.audit_records()], @@ -435,10 +522,10 @@ def test_websocket_upgrade_still_completes(self) -> None: 19082, extra_headers=( "Upgrade: websocket\r\n" - "Connection: Upgrade\r\n" f"Sec-WebSocket-Key: {key}\r\n" "Sec-WebSocket-Version: 13\r\n" ), + connection="Upgrade", ) self.assertIn(b"101", data.split(b"\r\n", 1)[0]) self.assertIn(b"ws-ok", data) From 4edaeca05e4037dc286537ebbd9f03bf679c6b20 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=83=AD=E5=90=89=E6=B5=A9?= <1625567290@qq.com> Date: Sun, 23 Aug 2026 22:02:12 +0800 Subject: [PATCH 11/12] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20TLS=20=E5=88=86?= =?UTF-8?q?=E7=89=87=E6=8F=A1=E6=89=8B=E6=B5=8B=E8=AF=95=E6=AD=BB=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../eval/harbor/test_egress_filter_live.py | 94 +++++++++++++++---- 1 file changed, 77 insertions(+), 17 deletions(-) diff --git a/packages/eval/harbor/test_egress_filter_live.py b/packages/eval/harbor/test_egress_filter_live.py index 4d1cad69f1..6bcedf9ed5 100644 --- a/packages/eval/harbor/test_egress_filter_live.py +++ b/packages/eval/harbor/test_egress_filter_live.py @@ -163,6 +163,82 @@ def docker_image_present(image: str) -> bool: return inspect.returncode == 0 +def finish_memory_bio_handshake(tls, incoming, outgoing, sock, timeout_s: float) -> str: + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + try: + tls.do_handshake() + except ssl.SSLWantReadError: + # Processing a server flight can produce the next client flight + # before OpenSSL asks for more input. Flush it now; waiting for + # another recv first deadlocks both peers. + pending = outgoing.read() + if pending: + sock.sendall(pending) + response = sock.recv(16 * 1024) + if not response: + raise AssertionError("proxy closed during the fragmented TLS handshake") + incoming.write(response) + continue + pending = outgoing.read() + if pending: + sock.sendall(pending) + return tls.version() or "" + raise AssertionError("fragmented TLS handshake did not complete") + + +class MemoryBioHandshakeDriverTest(unittest.TestCase): + def test_flushes_client_flight_before_waiting_for_more_server_bytes(self) -> None: + events = [] + + class FakeOutgoing: + pending = b"" + + def read(self): + pending, self.pending = self.pending, b"" + return pending + + class FakeIncoming: + def write(self, data): + events.append(("write", data)) + + outgoing = FakeOutgoing() + + class FakeTls: + calls = 0 + + def do_handshake(self): + self.calls += 1 + if self.calls == 1: + outgoing.pending = b"client-finished" + raise ssl.SSLWantReadError() + + def version(self): + return "TLSv1.3" + + class FakeSocket: + def sendall(self, data): + events.append(("send", data)) + + def recv(self, _size): + events.append(("recv", None)) + return b"server-finished" + + result = finish_memory_bio_handshake( + FakeTls(), FakeIncoming(), outgoing, FakeSocket(), timeout_s=1 + ) + + self.assertEqual(result, "TLSv1.3") + self.assertEqual( + events, + [ + ("send", b"client-finished"), + ("recv", None), + ("write", b"server-finished"), + ], + ) + + @unittest.skipUnless( os.environ.get("MAKA_EVAL_EGRESS_PROXY_TEST") == "1", "set MAKA_EVAL_EGRESS_PROXY_TEST=1 to run the live mitmproxy proxy test", @@ -380,23 +456,7 @@ def fragmented_tls_via_proxy(cls, first_fragment_size: int) -> str: time.sleep(0.05) sock.sendall(client_hello[first_fragment_size:]) sock.settimeout(10) - deadline = time.time() + 20 - while time.time() < deadline: - pending = outgoing.read() - if pending: - sock.sendall(pending) - try: - tls.do_handshake() - pending = outgoing.read() - if pending: - sock.sendall(pending) - return tls.version() or "" - except ssl.SSLWantReadError: - response = sock.recv(16 * 1024) - if not response: - raise AssertionError("proxy closed during the fragmented TLS handshake") - incoming.write(response) - raise AssertionError("fragmented TLS handshake did not complete") + return finish_memory_bio_handshake(tls, incoming, outgoing, sock, timeout_s=20) @classmethod def audit_records(cls) -> list[dict[str, object]]: From c02400d25f6df7cbda050fb91dde4acc6ad7615e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=83=AD=E5=90=89=E6=B5=A9?= <1625567290@qq.com> Date: Mon, 24 Aug 2026 11:17:15 +0800 Subject: [PATCH 12/12] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=87=BA=E5=8F=A3?= =?UTF-8?q?=E4=BB=A3=E7=90=86=E7=9A=84=E7=A2=8E=E7=89=87=E5=8C=96=20TLS=20?= =?UTF-8?q?=E5=88=86=E7=B1=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/eval/harbor/egress_filter.py | 15 ++++++++ packages/eval/harbor/test_egress_filter.py | 38 +++++++++++++++---- .../eval/harbor/test_egress_filter_live.py | 11 +++++- 3 files changed, 54 insertions(+), 10 deletions(-) diff --git a/packages/eval/harbor/egress_filter.py b/packages/eval/harbor/egress_filter.py index c7c9e6b87c..84c24956ab 100644 --- a/packages/eval/harbor/egress_filter.py +++ b/packages/eval/harbor/egress_filter.py @@ -121,10 +121,13 @@ def public_trajectory_repository(host: str, path_query: str) -> bool: try: from mitmproxy.proxy import commands as proxy_commands from mitmproxy.proxy.layer import Layer + from mitmproxy.proxy.layers import ClientTLSLayer, ServerTLSLayer from mitmproxy.proxy.layers.tcp import TCPLayer except ImportError: proxy_commands = None Layer = object + ClientTLSLayer = None + ServerTLSLayer = None TCPLayer = None try: @@ -193,6 +196,14 @@ def next_layer(nextlayer: object) -> None: current = getattr(nextlayer, "layer", None) context = getattr(nextlayer, "context", None) if current is None: + data_client = _next_layer_bytes(nextlayer, "data_client") + if _is_fragmented_tls_record_prefix(data_client): + if ClientTLSLayer is None or ServerTLSLayer is None: + return + server_tls = ServerTLSLayer(context) + server_tls.child_layer = ClientTLSLayer(context) + nextlayer.layer = server_tls + return if not looks_like_raw_tcp(nextlayer): return record_raw_tunnel(context) @@ -269,6 +280,10 @@ def _could_start_tls_record(data: bytes) -> bool: """Keep a fragmented ClientHello undecided until its 3-byte prefix exists.""" if starts_like_tls_record(data): return True + return _is_fragmented_tls_record_prefix(data) + + +def _is_fragmented_tls_record_prefix(data: bytes) -> bool: return 0 < len(data) < 3 and b"\x16\x03".startswith(data) diff --git a/packages/eval/harbor/test_egress_filter.py b/packages/eval/harbor/test_egress_filter.py index 60b3995526..9e78699711 100644 --- a/packages/eval/harbor/test_egress_filter.py +++ b/packages/eval/harbor/test_egress_filter.py @@ -229,7 +229,26 @@ def data_client() -> bytes: self.assertEqual(record["ruleId"], "raw_tunnel") self.assertEqual(record["host"], "ssh.github.com") - def test_next_layer_leaves_tls_and_http_for_the_builtin_classifier(self) -> None: + def test_next_layer_routes_tls_and_http_without_raw_fallback(self) -> None: + class FakeServerTLSLayer: + child_layer: object | None = None + + def __init__(self, context: object) -> None: + self.context = context + context.layers.append(self) + + class FakeClientTLSLayer: + def __init__(self, context: object) -> None: + self.context = context + context.layers.append(self) + + previous_server_tls = MODULE.ServerTLSLayer + previous_client_tls = MODULE.ClientTLSLayer + self.addCleanup(setattr, MODULE, "ServerTLSLayer", previous_server_tls) + self.addCleanup(setattr, MODULE, "ClientTLSLayer", previous_client_tls) + MODULE.ServerTLSLayer = FakeServerTLSLayer + MODULE.ClientTLSLayer = FakeClientTLSLayer + context = SimpleNamespace(layers=[]) tls = SimpleNamespace( layer=None, @@ -245,18 +264,21 @@ def test_next_layer_leaves_tls_and_http_for_the_builtin_classifier(self) -> None (b"\x16\x03", b"\x01\x00\x00"), ): with self.subTest(tls_prefix=first): - buffered = bytearray(first) + fragmented_context = SimpleNamespace(layers=[]) fragmented = SimpleNamespace( layer=None, - context=context, - data_client=lambda: bytes(buffered), + context=fragmented_context, + data_client=lambda first=first: first, data_server=lambda: b"", ) MODULE.next_layer(fragmented) - self.assertIsNone(fragmented.layer) - buffered.extend(remainder) - MODULE.next_layer(fragmented) - self.assertIsNone(fragmented.layer) + self.assertIsInstance(fragmented.layer, FakeServerTLSLayer) + self.assertIsInstance(fragmented.layer.child_layer, FakeClientTLSLayer) + self.assertEqual( + fragmented_context.layers, + [fragmented.layer, fragmented.layer.child_layer], + ) + self.assertTrue((first + remainder).startswith(b"\x16\x03")) http = SimpleNamespace( layer=None, diff --git a/packages/eval/harbor/test_egress_filter_live.py b/packages/eval/harbor/test_egress_filter_live.py index 6bcedf9ed5..1197807651 100644 --- a/packages/eval/harbor/test_egress_filter_live.py +++ b/packages/eval/harbor/test_egress_filter_live.py @@ -426,7 +426,12 @@ def connect_via_proxy(cls, host: str, port: int, payload: bytes = b"CLIENT\n") - def fragmented_tls_via_proxy(cls, first_fragment_size: int) -> str: incoming = ssl.MemoryBIO() outgoing = ssl.MemoryBIO() - tls = ssl.create_default_context().wrap_bio( + tls_context = ssl.create_default_context() + # The proxy image creates an ephemeral private CA. This test exercises + # protocol classification and the handshake, not CA distribution. + tls_context.check_hostname = False + tls_context.verify_mode = ssl.CERT_NONE + tls = tls_context.wrap_bio( incoming, outgoing, server_side=False, @@ -536,7 +541,9 @@ def test_https_and_plain_http_still_forward(self) -> None: self.assertEqual(curl.stdout, "200", curl.stderr) def test_fragmented_tls_record_prefix_still_handshakes(self) -> None: - for first_fragment_size in (1, 2): + # Three bytes are sufficient for mitmproxy's built-in TLS classifier + # and provide a control for the one- and two-byte fragmented prefixes. + for first_fragment_size in (1, 2, 3): with self.subTest(first_fragment_size=first_fragment_size): self.assertTrue( self.fragmented_tls_via_proxy(first_fragment_size).startswith("TLS")