ci: run the Envoy ext_proc path against a real Envoy, in every body mode

.github/workflows/ci.yml

@@ -100,6 +100,28 @@ jobs:
             PASU_E2E_KERNEL=1 \
             cargo test -p pasu-cli --test run_e2e -- --test-threads=1 --nocapture
 
+  envoy-e2e:
+    name: Envoy ext_proc E2E (linux)
+    runs-on: ubuntu-latest
+    env:
+      # 정확한 패치 버전으로 고정한다. v1.37-latest 는 패치 릴리스마다 움직여서,
+      # 이 잡이 깨졌을 때 pasu 가 바뀐 건지 Envoy 가 바뀐 건지 구분할 수 없다.
+      # 수동 검증도 1.37 에서 했다. 올릴 때는 이 한 줄만 바꾼다.
+      ENVOY_IMAGE: envoyproxy/envoy:v1.37.6
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+      - uses: Swatinem/rust-cache@v2
+      - name: build pasu-proxy
+        run: cargo build --locked -p pasu-proxy
+      - name: pull Envoy
+        run: docker pull "$ENVOY_IMAGE"
+      # A real Envoy between curl and a mock provider, pasu judging with the
+      # production ruleset (deploy/rules.yaml). On Linux the harness runs Envoy
+      # with --network host, so it reaches pasu and the mock on 127.0.0.1.
+      - name: run E2E (BUFFERED, PII block/redact, NONE misconfig, pasu down)
+        run: tests/e2e/envoy/run.sh target/debug/pasu-proxy "$ENVOY_IMAGE"
+
   msrv:
     name: MSRV (1.86) 빌드
     runs-on: ubuntu-latest

AGENTS.md

@@ -37,8 +37,21 @@ bpf toolchain to work on the portable crates. The kernel paths are validated by
 the privileged **eBPF E2E** CI job on a real Ubuntu kernel — you generally
 **cannot build the eBPF crates on macOS**, so lean on CI for those.
 
-CI is 4 jobs (all must be green): `fmt·clippy·test`, `eBPF build+unit`,
-`eBPF E2E` (privileged), `cargo-deny`.
+CI (`.github/workflows/ci.yml`) is 6 jobs, all must be green: `fmt·clippy·test`,
+`eBPF build+unit`, `eBPF E2E` (privileged), `Envoy ext_proc E2E`, `MSRV (1.86)`,
+`cargo-deny`. `pasu-pii-kr` has its own workflow (`pii.yml`: check + bench),
+run when that crate changes.
+
+The Envoy E2E is a script you can run locally too — it needs docker, curl and
+python3, builds nothing and pulls nothing:
+
+```bash
+cargo build -p pasu-proxy
+docker pull envoyproxy/envoy:v1.37.6
+tests/e2e/envoy/run.sh target/debug/pasu-proxy envoyproxy/envoy:v1.37.6
+```
+
+Ports and the docker network mode are env overrides (see the script header).
 
 ## Crate map (acyclic — everything depends only on `pasu-core`)
 

CHANGELOG.md

@@ -60,6 +60,18 @@ All notable changes to pasu are documented here. The format follows
   byte; `/v1/embeddings` and an RRN prompt before the provider sees them; an
   unreachable pasu is a 500. Leaving out `processingMode` sends pasu nothing and lets everything
   through; the README says so.
+- **An Envoy ext_proc E2E in CI** (`tests/e2e/envoy/run.sh`, job
+  `Envoy ext_proc E2E`, Envoy pinned to 1.37.6). The Envoy behaviour above was
+  checked by hand; it is now asserted on every run against a real Envoy and the
+  production ruleset — passthrough, text and streamed text intact, shell tool
+  calls (streamed or not) refused with no chunk delivered, unknown endpoints and
+  Korean RRNs refused before the provider, redaction with a matching
+  `content-length`, `identity` upstream, `response_body_mode: NONE` refused
+  (500, `pasu_misconfigured`), and a stopped pasu answered with a 500.
+  `STREAMED` and `FULL_DUPLEX_STREAMED` are run as well, against a provider that
+  streams in paced events and a client that sends its prompt in pieces — the
+  only way piece-by-piece holding is exercised. The mock records a request cut
+  off mid-body, so "never reached the provider" counts bytes that did.
 - **The proxy inspects requests, not only responses.** It forwarded every
   request untouched, so a prompt carrying whatever an agent had just read out of
   a file or a database row reached the provider unexamined. The kernel layer

integrations/README.md

@@ -81,6 +81,16 @@ holds: over `max_request_bytes` a prompt is refused (413), over
 passthrough path, a provider error, a `count_tokens` response, a prompt with
 no inspector on — streams through as it arrives.
 
+These are checked against a real Envoy on every CI run:
+[`tests/e2e/envoy/run.sh`](../tests/e2e/envoy/run.sh) puts Envoy between curl
+and a mock provider, with pasu on the production ruleset, and asserts each
+refusal above — including that `NONE` is refused and that a stopped pasu is a
+500 — plus that a refused request never reached the provider. `STREAMED` and
+`FULL_DUPLEX_STREAMED` run too, with the provider streaming in paced events and
+a prompt sent in pieces, so holding piece by piece is exercised: with the
+clearing of held pieces disabled, the run fails on a leaked streamed tool call
+and on a resident number reaching the provider.
+
 pasu asks the upstream for `accept-encoding: identity`, so no decompressor
 filter is needed. Envoy Gateway ([below](#envoy-gateway)) and agentgateway
 ([below](#agentgateway)) speak the same

tests/e2e/envoy/envoy.yaml.tmpl

@@ -0,0 +1,61 @@
+# Rendered by run.sh. The ext_proc filter block is integrations/envoy/envoy.yaml's,
+# with one knob opened up: @RESPONSE_BODY_MODE@ is BUFFERED for the supported
+# setup and NONE for the misconfiguration pasu must refuse. The provider is the
+# plain-HTTP mock instead of api.openai.com over TLS.
+admin:
+  address: { socket_address: { address: @ADMIN_ADDR@, port_value: @ADMIN_PORT@ } }
+static_resources:
+  listeners:
+  - name: llm
+    address: { socket_address: { address: @LISTEN_ADDR@, port_value: @LISTEN_PORT@ } }
+    per_connection_buffer_limit_bytes: 67108864
+    filter_chains:
+    - filters:
+      - name: envoy.filters.network.http_connection_manager
+        typed_config:
+          "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
+          stat_prefix: llm
+          route_config:
+            virtual_hosts:
+            - name: provider
+              domains: ["*"]
+              routes:
+              - match: { prefix: "/" }
+                route: { cluster: provider, auto_host_rewrite: true, timeout: 300s }
+          http_filters:
+          - name: envoy.filters.http.ext_proc
+            typed_config:
+              "@type": type.googleapis.com/envoy.extensions.filters.http.ext_proc.v3.ExternalProcessor
+              grpc_service:
+                envoy_grpc: { cluster_name: pasu }
+                timeout: 10s
+              failure_mode_allow: false
+              message_timeout: 10s
+              processing_mode:
+                request_header_mode: SEND
+                response_header_mode: SEND
+                request_body_mode: @REQUEST_BODY_MODE@
+                response_body_mode: @RESPONSE_BODY_MODE@
+                request_trailer_mode: @TRAILER_MODE@
+                response_trailer_mode: @TRAILER_MODE@
+          - name: envoy.filters.http.router
+            typed_config:
+              "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router
+  clusters:
+  - name: pasu
+    type: STRICT_DNS
+    dns_lookup_family: V4_ONLY
+    typed_extension_protocol_options:
+      envoy.extensions.upstreams.http.v3.HttpProtocolOptions:
+        "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions
+        explicit_http_config: { http2_protocol_options: {} }
+    load_assignment:
+      cluster_name: pasu
+      endpoints: [{ lb_endpoints: [{ endpoint: { address: { socket_address: { address: @UPSTREAM_HOST@, port_value: @PASU_PORT@ } } } }] }]
+  - name: provider
+    type: LOGICAL_DNS
+    dns_lookup_family: V4_ONLY
+    per_connection_buffer_limit_bytes: 67108864
+    load_assignment:
+      cluster_name: provider
+      endpoints: [{ lb_endpoints: [{ endpoint: { address: { socket_address: { address: @UPSTREAM_HOST@, port_value: @MOCK_PORT@ } } } }] }]

tests/e2e/envoy/mock_provider.py

@@ -0,0 +1,233 @@
+#!/usr/bin/env python3
+"""A stand-in OpenAI provider for the Envoy ext_proc E2E (stdlib only).
+
+Every request that reaches it is appended to a JSONL record *before* the reply
+is sent, so the harness can assert that a refused request never arrived rather
+than assume it. What the model "answers" is chosen by a marker in the prompt:
+
+    TOOL_SHELL   -> the reply is a `shell` tool call
+    anything else -> a plain text reply
+
+`"stream": true` in the request selects the SSE form of the same reply, sent
+chunked the way a real provider streams it.
+
+    mock_provider.py <bind-host> <port> <record.jsonl>
+"""
+
+import json
+import sys
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+MARKER_SHELL = "TOOL_SHELL"
+MODEL = "mock-gpt"
+REPLY_TEXT = "hello from the mock provider"
+
+_record_lock = threading.Lock()
+
+
+EVENT_GAP_SECONDS = 0.05
+
+
+def prompt_text(body):
+    """The concatenated message contents of a chat request, or ''."""
+    try:
+        request = json.loads(body)
+    except ValueError:
+        return ""
+    if not isinstance(request, dict):
+        return ""
+    parts = []
+    for message in request.get("messages") or []:
+        content = message.get("content") if isinstance(message, dict) else None
+        if isinstance(content, str):
+            parts.append(content)
+    return "\n".join(parts)
+
+
+def wants_stream(body):
+    try:
+        return bool(json.loads(body).get("stream"))
+    except (ValueError, AttributeError):
+        return False
+
+
+def completion(tool_call):
+    message = {"role": "assistant", "content": None if tool_call else REPLY_TEXT}
+    if tool_call:
+        message["tool_calls"] = [
+            {
+                "id": "call_1",
+                "type": "function",
+                "function": {"name": "shell", "arguments": '{"cmd":"rm -rf /"}'},
+            }
+        ]
+    return {
+        "id": "chatcmpl-mock",
+        "object": "chat.completion",
+        "created": 0,
+        "model": MODEL,
+        "choices": [
+            {
+                "index": 0,
+                "message": message,
+                "finish_reason": "tool_calls" if tool_call else "stop",
+            }
+        ],
+    }
+
+
+def chunk(delta, finish=None):
+    return {
+        "id": "chatcmpl-mock",
+        "object": "chat.completion.chunk",
+        "created": 0,
+        "model": MODEL,
+        "choices": [{"index": 0, "delta": delta, "finish_reason": finish}],
+    }
+
+
+def stream_events(tool_call):
+    """SSE events, one per chunk, ending with [DONE]."""
+    if tool_call:
+        deltas = [
+            chunk({"role": "assistant", "content": None}),
+            chunk(
+                {
+                    "tool_calls": [
+                        {
+                            "index": 0,
+                            "id": "call_1",
+                            "type": "function",
+                            "function": {"name": "shell", "arguments": ""},
+                        }
+                    ]
+                }
+            ),
+            chunk({"tool_calls": [{"index": 0, "function": {"arguments": '{"cmd":'}}]}),
+            chunk({"tool_calls": [{"index": 0, "function": {"arguments": '"rm -rf /"}'}}]}),
+            chunk({}, "tool_calls"),
+        ]
+    else:
+        words = REPLY_TEXT.split(" ")
+        deltas = [chunk({"role": "assistant", "content": ""})]
+        deltas += [chunk({"content": w if i == 0 else " " + w}) for i, w in enumerate(words)]
+        deltas.append(chunk({}, "stop"))
+    events = ["data: " + json.dumps(d, separators=(",", ":")) + "\n\n" for d in deltas]
+    events.append("data: [DONE]\n\n")
+    return [e.encode() for e in events]
+
+
+class Handler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+    record_path = ""
+
+    def log_message(self, fmt, *args):
+        sys.stderr.write("mock: " + (fmt % args) + "\n")
+
+    def read_body(self):
+        """The request body. A body cut off mid-way (the data plane reset the
+        request) is recorded as far as it arrived — "never reached the
+        provider" must count bytes that did, not only requests that finished —
+        and then the handler stops."""
+        body = b""
+        try:
+            if "chunked" in (self.headers.get("transfer-encoding") or "").lower():
+                while True:
+                    size = int(self.rfile.readline().split(b";")[0].strip() or b"0", 16)
+                    if size == 0:
+                        self.rfile.readline()
+                        return body
+                    piece = self.rfile.read(size)
+                    body += piece
+                    if len(piece) < size:
+                        raise ConnectionError("chunk cut short")
+                    self.rfile.readline()
+            length = int(self.headers.get("content-length") or 0)
+            body = self.rfile.read(length) if length else b""
+            if len(body) < length:
+                raise ConnectionError("body cut short")
+            return body
+        except (ConnectionError, OSError, ValueError):
+            self.record(body, b"", truncated=True)
+            raise
+
+    def record(self, body, sent, truncated=False):
+        entry = {
+            "method": self.command,
+            "path": self.path,
+            "accept_encoding": self.headers.get("accept-encoding"),
+            "content_length": self.headers.get("content-length"),
+            "transfer_encoding": self.headers.get("transfer-encoding"),
+            "body_bytes": len(body),
+            "body": body.decode("utf-8", "replace"),
+            "prompt": prompt_text(body),
+            "sent": sent.decode("utf-8", "replace"),
+            "truncated": truncated,
+        }
+        with _record_lock, open(self.record_path, "a", encoding="utf-8") as f:
+            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+            f.flush()
+
+    def send_json(self, status, obj, body):
+        payload = json.dumps(obj).encode()
+        self.record(body, payload)
+        self.send_response(status)
+        self.send_header("content-type", "application/json")
+        self.send_header("content-length", str(len(payload)))
+        self.end_headers()
+        self.wfile.write(payload)
+
+    def send_stream(self, events, body):
+        self.record(body, b"".join(events))
+        self.send_response(200)
+        self.send_header("content-type", "text/event-stream")
+        self.send_header("cache-control", "no-cache")
+        self.send_header("transfer-encoding", "chunked")
+        self.end_headers()
+        for event in events:
+            self.wfile.write(b"%x\r\n%s\r\n" % (len(event), event))
+            self.wfile.flush()
+            # A pause between events, as a model generating tokens has, so the
+            # data plane sees the stream in pieces. Without it Envoy coalesces
+            # a small stream into one piece and piece-by-piece holding is never
+            # exercised.
+            time.sleep(EVENT_GAP_SECONDS)
+        self.wfile.write(b"0\r\n\r\n")
+        self.wfile.flush()
+
+    def do_GET(self):
+        body = self.read_body()
+        if self.path.split("?")[0] == "/v1/models":
+            models = {"object": "list", "data": [{"id": MODEL, "object": "model"}]}
+            return self.send_json(200, models, body)
+        return self.send_json(404, {"error": {"message": "not found"}}, body)
+
+    def do_POST(self):
+        body = self.read_body()
+        path = self.path.split("?")[0]
+        if path == "/v1/chat/completions":
+            tool_call = MARKER_SHELL in prompt_text(body)
+            if wants_stream(body):
+                return self.send_stream(stream_events(tool_call), body)
+            return self.send_json(200, completion(tool_call), body)
+        if path == "/v1/embeddings":
+            embedding = {"object": "list", "data": [{"embedding": [0.0], "index": 0}]}
+            return self.send_json(200, embedding, body)
+        return self.send_json(404, {"error": {"message": "not found"}}, body)
+
+
+def main():
+    if len(sys.argv) != 4:
+        sys.exit("usage: mock_provider.py <bind-host> <port> <record.jsonl>")
+    host, port, record = sys.argv[1], int(sys.argv[2]), sys.argv[3]
+    Handler.record_path = record
+    server = ThreadingHTTPServer((host, port), Handler)
+    server.daemon_threads = True
+    sys.stderr.write("mock provider on %s:%d\n" % (host, port))
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()