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/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 d86ae896f8..84c24956ab 100644 --- a/packages/eval/harbor/egress_filter.py +++ b/packages/eval/harbor/egress_filter.py @@ -120,14 +120,52 @@ 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: + 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 + # 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) + + def http_connect(flow: object) -> None: try: raw_url = connect_target_url(flow) @@ -137,6 +175,9 @@ def http_connect(flow: object) -> None: def tcp_start(flow: object) -> None: + # 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) @@ -149,12 +190,31 @@ 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) - if current is None or type(current).__name__ != "TCPLayer": - return 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) + nextlayer.layer = CloseRawLayer(context) + return + if TCPLayer is None or not isinstance(current, TCPLayer): + return 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: @@ -197,6 +257,58 @@ def connect_target_url(flow: object) -> str: return f"https://{host}:{port}/" +def looks_like_raw_tcp(nextlayer: object) -> bool: + """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 _could_start_tls_record(data_client): + return False + if not data_client and not data_server: + return False + if data_server or data_client.startswith(b"SSH"): + return True + 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 _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) + + +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: + 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 @@ -224,9 +336,31 @@ def kill_flow(flow: object) -> None: pass -class CloseRawLayer: +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: - 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..9e78699711 100644 --- a/packages/eval/harbor/test_egress_filter.py +++ b/packages/eval/harbor/test_egress_filter.py @@ -201,20 +201,231 @@ 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_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, + context=context, + data_client=lambda: b"\x16\x03\x01\x00\x00", + data_server=lambda: b"", + ) + 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): + fragmented_context = SimpleNamespace(layers=[]) + fragmented = SimpleNamespace( + layer=None, + context=fragmented_context, + data_client=lambda first=first: first, + data_server=lambda: b"", + ) + MODULE.next_layer(fragmented) + 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, + 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) + + 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" + 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"} + ), + "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 TCPLayer: - pass + class FakeTCPLayer: + def __init__(self, context: object) -> None: + self.context = context + context.layers.append(self) + + 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)) + 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)) 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_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 + + 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" 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..1197807651 --- /dev/null +++ b/packages/eval/harbor/test_egress_filter_live.py @@ -0,0 +1,606 @@ +# 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 +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, 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 +""" + +from __future__ import annotations + +import base64 +import json +import os +import shutil +import socket +import ssl +import subprocess +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" +ORIGIN_IMAGE = "python:3.12-slim" +COMMAND_TIMEOUT_S = 60 +CLOSE_TIMEOUT_S = 2.0 + +ORIGIN_SCRIPT = r""" +import base64, hashlib, json, socket, threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +stats = {"raw_recv": 0, "raw_closed": 0, "upgrade_recv": 0, "upgrade_closed": 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(60) + try: + while True: + chunk = self.connection.recv(64) + if not chunk: + break + stats["upgrade_recv"] += len(chunk) + except OSError: + pass + finally: + stats["upgrade_closed"] += 1 + 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() + try: + conn.sendall(b"RAW-BANNER\n") + conn.settimeout(60) + while True: + chunk = conn.recv(64) + if not chunk: + break + stats["raw_recv"] += len(chunk) + except OSError: + pass + finally: + stats["raw_closed"] += 1 + 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 + + +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 + + +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", +) +class LiveEgressFilterTest(unittest.TestCase): + workdir: Path | None = None + network: str = "" + proxy: str = "" + origin: str = "" + proxy_port: int = 0 + + @classmethod + def setUpClass(cls) -> None: + if not docker_available(): + raise unittest.SkipTest("Docker daemon is unavailable") + 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" + 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) + subprocess.run(["docker", "network", "create", cls.network], check=True, timeout=COMMAND_TIMEOUT_S) + subprocess.run( + [ + "docker", + "run", + "-d", + "--name", + cls.origin, + "--network", + cls.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", + cls.proxy, + "--network", + cls.network, + "-p", + "127.0.0.1::8080", + "-v", + f"{HARBOR_DIR / 'egress_filter.py'}:/opt/maka-eval/egress_filter.py:ro", + PROXY_IMAGE, + ], + 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 (cls.proxy, cls.origin): + if not name: + continue + subprocess.run( + ["docker", "rm", "-f", name], + 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: + deadline = time.time() + 30 + last_error = "proxy did not listen" + while time.time() < deadline: + try: + with socket.create_connection(("127.0.0.1", cls.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_until_close(cls, sock: socket.socket, limit: int = 8192) -> bytes: + sock.settimeout(CLOSE_TIMEOUT_S) + data = b"" + while len(data) < limit: + try: + chunk = sock.recv(4096) + except ConnectionResetError: + return data + if not chunk: + return data + data += chunk + return data + + @classmethod + 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}" + 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) + 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", 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(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: + 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_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, + 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) + return finish_memory_bio_handshake(tls, incoming, outgoing, sock, timeout_s=20) + + @classmethod + def audit_records(cls) -> list[dict[str, object]]: + listed = subprocess.run( + [ + "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, + ) + if listed.returncode != 0: + raise AssertionError(listed.stderr) + 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", + cls.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) + + @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) + curl = subprocess.run( + [ + "curl", + "--silent", + "--show-error", + "--http1.1", + "--max-time", + "20", + "--proxy", + f"http://127.0.0.1:{self.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_fragmented_tls_record_prefix_still_handshakes(self) -> None: + # 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") + ) + + 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"") + 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: + 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) + 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()], + ) + + 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" + 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) + 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..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" + "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) {