From 761029b23cd7d9eccddcff4c7f31f5740c490881 Mon Sep 17 00:00:00 2001 From: Ho Geun Choi Date: Wed, 30 Sep 2026 23:44:54 +0900 Subject: [PATCH 1/3] ci: run the Envoy ext_proc path against a real Envoy on every push The ext_proc front-end was verified against Envoy 1.37 by hand. Those runs are now a harness (tests/e2e/envoy/run.sh) and a CI job: a real Envoy between curl and a stdlib mock OpenAI provider, pasu-proxy judging with the production ruleset (deploy/rules.yaml), 13 cases each asserting PASS/FAIL: passthrough, text and streamed text byte-equal to what the provider sent, identity upstream for a gzip client, shell tool calls (streamed or not) refused 403 with no chunk delivered, /v1/embeddings and a Korean RRN refused before the provider, redaction with a matching content-length, response_body_mode NONE refused with 500, and a stopped pasu answered 500. The mock records every request before replying, so "never reached the provider" is asserted rather than assumed. Readiness waits for pasu's gRPC port and for Envoy's admin to report LIVE with the listener bound; a trap removes the container and processes. On Linux Envoy runs with --network host; elsewhere it reaches the host through host.docker.internal. Envoy is pinned to v1.37.6 rather than v1.37-latest, so a red job means pasu changed, not Envoy. With misconfigured() forced to return None the NONE case fails with the shell call delivered as a 200. Signed-off-by: Ho Geun Choi --- .github/workflows/ci.yml | 22 ++ tests/e2e/envoy/envoy.yaml.tmpl | 59 ++++ tests/e2e/envoy/mock_provider.py | 209 ++++++++++++++ tests/e2e/envoy/run.sh | 450 +++++++++++++++++++++++++++++++ 4 files changed, 740 insertions(+) create mode 100644 tests/e2e/envoy/envoy.yaml.tmpl create mode 100755 tests/e2e/envoy/mock_provider.py create mode 100755 tests/e2e/envoy/run.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0f940b9..9b6e8c7 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/tests/e2e/envoy/envoy.yaml.tmpl b/tests/e2e/envoy/envoy.yaml.tmpl new file mode 100644 index 0000000..01d92bc --- /dev/null +++ b/tests/e2e/envoy/envoy.yaml.tmpl @@ -0,0 +1,59 @@ +# 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: BUFFERED + response_body_mode: @RESPONSE_BODY_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@ } } } }] }] diff --git a/tests/e2e/envoy/mock_provider.py b/tests/e2e/envoy/mock_provider.py new file mode 100755 index 0000000..66c72ae --- /dev/null +++ b/tests/e2e/envoy/mock_provider.py @@ -0,0 +1,209 @@ +#!/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 +""" + +import json +import sys +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +MARKER_SHELL = "TOOL_SHELL" +MODEL = "mock-gpt" +REPLY_TEXT = "hello from the mock provider" + +_record_lock = threading.Lock() + + +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): + if "chunked" in (self.headers.get("transfer-encoding") or "").lower(): + body = b"" + while True: + size = int(self.rfile.readline().split(b";")[0].strip() or b"0", 16) + if size == 0: + self.rfile.readline() + return body + body += self.rfile.read(size) + self.rfile.readline() + length = int(self.headers.get("content-length") or 0) + return self.rfile.read(length) if length else b"" + + def record(self, body, sent): + 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"), + } + 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() + 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 ") + 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() diff --git a/tests/e2e/envoy/run.sh b/tests/e2e/envoy/run.sh new file mode 100755 index 0000000..67c31d1 --- /dev/null +++ b/tests/e2e/envoy/run.sh @@ -0,0 +1,450 @@ +#!/usr/bin/env bash +# Envoy ext_proc E2E: a real Envoy in front of a mock OpenAI provider, with +# pasu-proxy (--frontend ext-proc) deciding, judged against the PRODUCTION +# ruleset (deploy/rules.yaml, default deny). +# +# tests/e2e/envoy/run.sh +# +# Builds nothing and pulls nothing: both inputs must already exist. Every case +# prints PASS or FAIL; the script exits non-zero if any case fails or the setup +# cannot come up. Needs bash, curl, python3 and docker. +# +# Environment (all optional): +# PASU_E2E_ENVOY_PORT Envoy listener (default 10000) +# PASU_E2E_ADMIN_PORT Envoy admin (default 9901) +# PASU_E2E_PASU_PORT pasu ext_proc gRPC (default 8791) +# PASU_E2E_MOCK_PORT mock provider (default 18901) +# PASU_E2E_NETWORK host | bridge | auto (default auto: host on Linux, +# bridge + host.docker.internal elsewhere) +# PASU_E2E_BIND address pasu and the mock bind (default 127.0.0.1; +# 0.0.0.0 for bridge on Linux, where the container +# reaches the host through the bridge, not loopback) +# PASU_E2E_KEEP=1 keep the work directory (logs, rendered configs) +set -euo pipefail + +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +ROOT="$(cd "$HERE/../../.." && pwd)" +# Not configurable on purpose: the E2E validates what ships (CLAUDE.md §3). +POLICY="$ROOT/deploy/rules.yaml" + +PASU_BIN="${1:-}" +ENVOY_IMAGE="${2:-}" +if [[ -z "$PASU_BIN" || -z "$ENVOY_IMAGE" ]]; then + echo "usage: $0 " >&2 + exit 2 +fi +if [[ ! -x "$PASU_BIN" ]]; then + echo "error: $PASU_BIN is not an executable" >&2 + exit 2 +fi +PASU_BIN="$(cd "$(dirname "$PASU_BIN")" && pwd)/$(basename "$PASU_BIN")" +if ! docker image inspect "$ENVOY_IMAGE" >/dev/null 2>&1; then + echo "error: image $ENVOY_IMAGE is not present (the harness does not pull)" >&2 + exit 2 +fi + +ENVOY_PORT="${PASU_E2E_ENVOY_PORT:-10000}" +ADMIN_PORT="${PASU_E2E_ADMIN_PORT:-9901}" +PASU_PORT="${PASU_E2E_PASU_PORT:-8791}" +MOCK_PORT="${PASU_E2E_MOCK_PORT:-18901}" +NETWORK="${PASU_E2E_NETWORK:-auto}" +if [[ "$NETWORK" == auto ]]; then + [[ "$(uname -s)" == Linux ]] && NETWORK=host || NETWORK=bridge +fi + +case "$NETWORK" in + host) + UPSTREAM_HOST=127.0.0.1 + ENVOY_ADDR=127.0.0.1 + DEFAULT_BIND=127.0.0.1 + DOCKER_NET=(--network host) + ;; + bridge) + UPSTREAM_HOST=host.docker.internal + ENVOY_ADDR=0.0.0.0 + DOCKER_NET=(-p "127.0.0.1:$ENVOY_PORT:$ENVOY_PORT" -p "127.0.0.1:$ADMIN_PORT:$ADMIN_PORT") + if [[ "$(uname -s)" == Linux ]]; then + DOCKER_NET+=(--add-host host.docker.internal:host-gateway) + DEFAULT_BIND=0.0.0.0 + else + DEFAULT_BIND=127.0.0.1 # Docker Desktop delivers host.docker.internal to loopback + fi + ;; + *) + echo "error: PASU_E2E_NETWORK must be host, bridge or auto (got $NETWORK)" >&2 + exit 2 + ;; +esac +BIND="${PASU_E2E_BIND:-$DEFAULT_BIND}" + +WORK="$(mktemp -d "${TMPDIR:-/tmp}/pasu-envoy-e2e.XXXXXX")" +RECORD="$WORK/provider-requests.jsonl" +: >"$RECORD" +CONTAINER="pasu-envoy-ci-$$" +MOCK_PID="" +PASU_PID="" +PHASE="" + +log() { printf '%s\n' "$*"; } + +cleanup() { + local status=$? + set +e + [[ -n "$PASU_PID" ]] && kill "$PASU_PID" 2>/dev/null && wait "$PASU_PID" 2>/dev/null + [[ -n "$MOCK_PID" ]] && kill "$MOCK_PID" 2>/dev/null && wait "$MOCK_PID" 2>/dev/null + docker rm -f "$CONTAINER" >/dev/null 2>&1 + if [[ "${PASU_E2E_KEEP:-}" == 1 ]]; then + log "work directory kept: $WORK" + else + rm -rf "$WORK" + fi + exit "$status" +} +trap cleanup EXIT +trap 'exit 130' INT TERM + +port_open() { (exec 3<>"/dev/tcp/127.0.0.1/$1") 2>/dev/null; } + +# wait_for +wait_for() { + local deadline=$((SECONDS + $1)) what=$2 + shift 2 + until "$@"; do + if ((SECONDS >= deadline)); then + log "SETUP FAILED: timed out waiting for $what" + return 1 + fi + sleep 0.2 + done +} + +alive() { kill -0 "$1" 2>/dev/null; } + +dump_logs() { + local f + if envoy_running; then docker logs "$CONTAINER" >"$WORK/envoy-$PHASE.log" 2>&1 || true; fi + for f in "$WORK"/pasu-*.log "$WORK"/envoy-*.log "$WORK/mock.log"; do + [[ -f "$f" ]] || continue + log "---- $(basename "$f") (last 30 lines) ----" + tail -n 30 "$f" + done +} + +setup_failed() { + dump_logs + exit 2 +} + +start_mock() { + python3 "$HERE/mock_provider.py" "$BIND" "$MOCK_PORT" "$RECORD" >"$WORK/mock.log" 2>&1 & + MOCK_PID=$! + mock_up_or_dead() { port_open "$MOCK_PORT" || ! alive "$MOCK_PID"; } + wait_for 15 "mock provider on :$MOCK_PORT" mock_up_or_dead || setup_failed + alive "$MOCK_PID" || { log "SETUP FAILED: mock provider exited"; setup_failed; } +} + +start_pasu() { + if port_open "$PASU_PORT"; then + log "SETUP FAILED: something already listens on :$PASU_PORT" + setup_failed + fi + "$PASU_BIN" --frontend ext-proc --listen "$BIND:$PASU_PORT" --policy "$POLICY" "$@" \ + >"$WORK/pasu-$PHASE.log" 2>&1 & + PASU_PID=$! + pasu_up_or_dead() { port_open "$PASU_PORT" || ! alive "$PASU_PID"; } + wait_for 15 "pasu gRPC on :$PASU_PORT" pasu_up_or_dead || setup_failed + alive "$PASU_PID" || { log "SETUP FAILED: pasu-proxy exited"; setup_failed; } +} + +stop_pasu() { + [[ -z "$PASU_PID" ]] && return 0 + kill "$PASU_PID" 2>/dev/null || true + wait "$PASU_PID" 2>/dev/null || true + PASU_PID="" + pasu_down() { ! port_open "$PASU_PORT"; } + wait_for 15 "pasu to release :$PASU_PORT" pasu_down || setup_failed +} + +envoy_ready() { + [[ "$(curl -s --max-time 2 "http://127.0.0.1:$ADMIN_PORT/ready" 2>/dev/null)" == LIVE* ]] && + curl -s --max-time 2 "http://127.0.0.1:$ADMIN_PORT/listeners" 2>/dev/null | + grep -q ":$ENVOY_PORT" +} + +envoy_running() { + [[ "$(docker inspect -f '{{.State.Running}}' "$CONTAINER" 2>/dev/null)" == true ]] +} + +# start_envoy +start_envoy() { + local cfg="$WORK/envoy-$1.yaml" admin_addr=127.0.0.1 + [[ "$NETWORK" == bridge ]] && admin_addr=0.0.0.0 + sed -e "s/@RESPONSE_BODY_MODE@/$1/" \ + -e "s/@ADMIN_ADDR@/$admin_addr/" -e "s/@ADMIN_PORT@/$ADMIN_PORT/" \ + -e "s/@LISTEN_ADDR@/$ENVOY_ADDR/" -e "s/@LISTEN_PORT@/$ENVOY_PORT/" \ + -e "s/@UPSTREAM_HOST@/$UPSTREAM_HOST/g" \ + -e "s/@PASU_PORT@/$PASU_PORT/" -e "s/@MOCK_PORT@/$MOCK_PORT/" \ + "$HERE/envoy.yaml.tmpl" >"$cfg" + chmod 644 "$cfg" + docker rm -f "$CONTAINER" >/dev/null 2>&1 || true + docker run -d --name "$CONTAINER" "${DOCKER_NET[@]}" \ + -v "$cfg:/etc/envoy/envoy.yaml:ro" \ + "$ENVOY_IMAGE" -c /etc/envoy/envoy.yaml >/dev/null || setup_failed + ready_or_dead() { envoy_ready || ! envoy_running; } + wait_for 60 "Envoy admin LIVE with listener :$ENVOY_PORT" ready_or_dead || setup_failed + envoy_running || { log "SETUP FAILED: Envoy exited"; setup_failed; } +} + +stop_envoy() { + docker logs "$CONTAINER" >"$WORK/envoy-$PHASE.log" 2>&1 || true + docker rm -f "$CONTAINER" >/dev/null 2>&1 || true +} + +# ---- cases ------------------------------------------------------------------ + +PASSED=0 +FAILED=0 +SUMMARY=() +CASE="" +CASE_OK=1 + +begin() { + CASE=$1 + CASE_OK=1 + : >"$RECORD" +} + +check() { # check + local what=$1 + shift + if ! "$@"; then + CASE_OK=0 + log " expected: $what" + fi +} + +end() { + if ((CASE_OK)); then + PASSED=$((PASSED + 1)) + SUMMARY+=("PASS $CASE") + log "PASS $CASE" + else + FAILED=$((FAILED + 1)) + SUMMARY+=("FAIL $CASE") + log "FAIL $CASE (status $(cat "$WORK/$CASE.status"), body: $(head -c 300 "$WORK/$CASE.body"))" + fi +} + +# send [JSON body] [extra curl args...] -> $WORK/$CASE.{status,body} +send() { + local method=$1 path=$2 body=${3:-} + shift $(($# < 3 ? $# : 3)) + local args=(-sS --max-time 30 -X "$method" -o "$WORK/$CASE.body" -w '%{http_code}') + [[ -n "$body" ]] && args+=(-H 'content-type: application/json' --data-binary "$body") + curl "${args[@]}" "$@" "http://127.0.0.1:$ENVOY_PORT$path" >"$WORK/$CASE.status" 2>"$WORK/$CASE.curl" || + true + [[ -s "$WORK/$CASE.status" ]] || echo 000 >"$WORK/$CASE.status" + touch "$WORK/$CASE.body" +} + +chat() { # chat [stream true|false] + python3 -c 'import json,sys; print(json.dumps({"model":"mock-gpt","stream":sys.argv[2]=="true","messages":[{"role":"user","content":sys.argv[1]}]}, ensure_ascii=False))' \ + "$1" "${2:-false}" +} + +status_is() { [[ "$(cat "$WORK/$CASE.status")" == "$1" ]]; } +reached_times() { [[ "$(wc -l <"$RECORD" | tr -d ' ')" == "$1" ]]; } + +# py_check +py_check() { + python3 - "$WORK/$CASE.body" "$RECORD" "$1" <<'EOF' +import json, sys +b = open(sys.argv[1], "rb").read() +r = [json.loads(l) for l in open(sys.argv[2], encoding="utf-8") if l.strip()] +def pasu_block(): + try: + return json.loads(b)["error"]["type"] == "pasu_policy_block" + except (ValueError, KeyError, TypeError): + return False +sys.exit(0 if eval(sys.argv[3]) else 1) +EOF +} + +is_pasu_block() { py_check 'pasu_block()'; } + +# ---- phase 1: supported setup (BUFFERED), no inspectors --------------------- + +run_buffered_plain() { + PHASE=buffered + log "== phase: Envoy BUFFERED, pasu (no inspector) ==" + start_pasu + start_envoy BUFFERED + + begin models-passthrough + send GET /v1/models + check "200" status_is 200 + check "provider reached once" reached_times 1 + check "the provider's model list" py_check 'b"mock-gpt" in b' + end + + begin text-completion + send POST /v1/chat/completions "$(chat 'say hello')" + check "200" status_is 200 + check "provider reached once" reached_times 1 + check "the text reply, unchanged" py_check 'b == r[-1]["sent"].encode()' + end + + begin text-completion-streamed + send POST /v1/chat/completions "$(chat 'say hello' true)" + check "200" status_is 200 + check "provider reached once" reached_times 1 + check "every SSE data line intact (byte-equal to what the provider sent)" \ + py_check 'b == r[-1]["sent"].encode() and b.count(b"data: ") >= 3 and b.endswith(b"data: [DONE]\n\n")' + end + + begin accept-encoding-identity + send POST /v1/chat/completions "$(chat 'say hello')" -H 'accept-encoding: gzip' + check "200" status_is 200 + check "provider saw accept-encoding: identity, not the client's gzip" \ + py_check 'len(r) == 1 and r[0]["accept_encoding"] == "identity"' + end + + begin shell-tool-call + send POST /v1/chat/completions "$(chat 'TOOL_SHELL please')" + check "403" status_is 403 + check "pasu_policy_block JSON" is_pasu_block + check "the model did answer with the tool call (refused on the response)" reached_times 1 + check "no tool call in what the client got" py_check 'b"call_1" not in b and b"rm -rf" not in b' + end + + begin shell-tool-call-streamed + send POST /v1/chat/completions "$(chat 'TOOL_SHELL please' true)" + check "403" status_is 403 + check "pasu_policy_block JSON" is_pasu_block + check "the model did stream the tool call" reached_times 1 + check "no SSE chunk delivered" py_check 'b"data:" not in b and b"call_1" not in b' + end + + begin unknown-endpoint-embeddings + send POST /v1/embeddings '{"model":"mock-embed","input":"hello"}' + check "403" status_is 403 + check "pasu_policy_block JSON" is_pasu_block + check "never reached the provider" reached_times 0 + end + + stop_envoy + stop_pasu +} + +# ---- phase 2: Korean PII refused -------------------------------------------- + +RRN='900101-1234567' + +run_pii_block() { + PHASE=pii-block + log "== phase: Envoy BUFFERED, pasu --block-pii-kr ==" + start_pasu --block-pii-kr + start_envoy BUFFERED + + begin pii-clean-prompt-passes + send POST /v1/chat/completions "$(chat 'my order number is 12345')" + check "200" status_is 200 + check "provider reached once" reached_times 1 + end + + begin pii-rrn-refused + send POST /v1/chat/completions "$(chat "주민번호는 $RRN 입니다")" + check "403" status_is 403 + check "pasu_policy_block JSON" is_pasu_block + check "never reached the provider" reached_times 0 + check "the refusal does not quote the value" py_check "b'$RRN' not in b" + end + + stop_envoy + stop_pasu +} + +# ---- phase 3: Korean PII redacted ------------------------------------------- + +run_pii_redact() { + PHASE=pii-redact + log "== phase: Envoy BUFFERED, pasu --block-pii-kr --redact ==" + start_pasu --block-pii-kr --redact + start_envoy BUFFERED + + begin pii-rrn-redacted + send POST /v1/chat/completions "$(chat "주민번호는 $RRN 입니다")" + check "200 (Envoy accepted the rewritten body)" status_is 200 + check "provider reached once" reached_times 1 + check "provider got [REDACTED:ko-rrn] and not the number" \ + py_check "'[REDACTED:ko-rrn]' in r[0]['prompt'] and '$RRN' not in r[0]['body']" + check "content-length matches the rewritten body" \ + py_check 'r[0]["content_length"] is not None and int(r[0]["content_length"]) == r[0]["body_bytes"]' + end + + stop_envoy + stop_pasu +} + +# ---- phase 4: misconfigured Envoy (response_body_mode: NONE) ---------------- + +run_misconfigured() { + PHASE=none + log "== phase: Envoy response_body_mode NONE, pasu (no inspector) ==" + start_pasu + start_envoy NONE + + begin none-passthrough-still-works + send GET /v1/models + check "200 (nothing to judge on a passthrough)" status_is 200 + check "provider reached once" reached_times 1 + end + + begin none-shell-refused + send POST /v1/chat/completions "$(chat 'TOOL_SHELL please')" + check "500 (pasu refuses a completion it would never be shown)" status_is 500 + check "pasu_policy_block JSON naming the setting" \ + py_check 'pasu_block() and b"response_body_mode" in b' + check "never reached the provider" reached_times 0 + end + + stop_envoy + stop_pasu +} + +# ---- phase 5: pasu not running ---------------------------------------------- + +run_pasu_down() { + PHASE=pasu-down + log "== phase: Envoy BUFFERED, pasu NOT running ==" + start_envoy BUFFERED + + begin pasu-down-fails-closed + check "nothing listening on the pasu port" eval '! port_open "$PASU_PORT"' + send POST /v1/chat/completions "$(chat 'say hello')" + check "500 (failure_mode_allow: false)" status_is 500 + check "never reached the provider" reached_times 0 + end + + stop_envoy +} + +log "pasu-proxy: $PASU_BIN" +log "envoy: $ENVOY_IMAGE" +log "policy: $POLICY" +log "network: $NETWORK (envoy :$ENVOY_PORT admin :$ADMIN_PORT, pasu $BIND:$PASU_PORT, mock $BIND:$MOCK_PORT)" + +start_mock +run_buffered_plain +run_pii_block +run_pii_redact +run_misconfigured +run_pasu_down + +log "" +log "== summary: $PASSED passed, $FAILED failed ==" +for line in "${SUMMARY[@]}"; do log " $line"; done +if ((FAILED > 0)); then + dump_logs + exit 1 +fi From 9cf50fedc16a76befbef3152fd30b93ce00bc8cc Mon Sep 17 00:00:00 2001 From: Ho Geun Choi Date: Wed, 30 Sep 2026 23:44:54 +0900 Subject: [PATCH 2/3] docs: list the CI jobs as they are, and point at the Envoy E2E AGENTS.md said CI was four jobs; it is six in ci.yml (MSRV was missing, and the Envoy E2E is new), with pii.yml alongside. It now says how to run the Envoy harness locally. integrations/README.md notes that its Envoy claims are checked by that harness; CHANGELOG records it. Signed-off-by: Ho Geun Choi --- AGENTS.md | 17 +++++++++++++++-- CHANGELOG.md | 8 ++++++++ integrations/README.md | 6 ++++++ 3 files changed, 29 insertions(+), 2 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 2d90a98..e044577 100644 --- a/AGENTS.md +++ b/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`) diff --git a/CHANGELOG.md b/CHANGELOG.md index 64a49c4..4697d1c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -60,6 +60,14 @@ 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), and a stopped pasu answered with a 500. - **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 diff --git a/integrations/README.md b/integrations/README.md index b1dcf82..5eaa573 100644 --- a/integrations/README.md +++ b/integrations/README.md @@ -81,6 +81,12 @@ 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. + 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 From df9ab2c26da14b3e7ce6803f0d238e6cb5e9c22e Mon Sep 17 00:00:00 2001 From: Ho Geun Choi Date: Thu, 1 Oct 2026 00:19:02 +0900 Subject: [PATCH 3/3] ci(e2e): run the Envoy harness in STREAMED and FULL_DUPLEX_STREAMED too The harness only ran BUFFERED, and its mock wrote small streams in one go and recorded a request only once it had read the whole body. Under those conditions disabling pasu's piece clearing in STREAMED still passed: Envoy coalesced every body into one piece, and a prompt reset mid-body left no record at the provider. The mock now paces its SSE events and records a request cut off mid-body; a small client sends a prompt in two pieces with the resident number in the first. With clearing disabled the run now fails three ways: a streamed text completion duplicated, a streamed shell tool call delivered with a 200, and the number reaching the provider. The misconfiguration case checks the new pasu_misconfigured type. 25 cases, all passing on Envoy 1.38. Signed-off-by: Ho Geun Choi --- CHANGELOG.md | 6 +- integrations/README.md | 6 +- tests/e2e/envoy/envoy.yaml.tmpl | 4 +- tests/e2e/envoy/mock_provider.py | 48 ++++++++++++---- tests/e2e/envoy/run.sh | 95 ++++++++++++++++++++++++++++++-- tests/e2e/envoy/slow_post.py | 54 ++++++++++++++++++ 6 files changed, 193 insertions(+), 20 deletions(-) create mode 100644 tests/e2e/envoy/slow_post.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 4697d1c..3915a5d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -67,7 +67,11 @@ All notable changes to pasu are documented here. The format follows 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), and a stopped pasu answered with a 500. + (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 diff --git a/integrations/README.md b/integrations/README.md index 5eaa573..945a8d3 100644 --- a/integrations/README.md +++ b/integrations/README.md @@ -85,7 +85,11 @@ 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. +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 diff --git a/tests/e2e/envoy/envoy.yaml.tmpl b/tests/e2e/envoy/envoy.yaml.tmpl index 01d92bc..ec6d9a7 100644 --- a/tests/e2e/envoy/envoy.yaml.tmpl +++ b/tests/e2e/envoy/envoy.yaml.tmpl @@ -34,8 +34,10 @@ static_resources: processing_mode: request_header_mode: SEND response_header_mode: SEND - request_body_mode: BUFFERED + 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 diff --git a/tests/e2e/envoy/mock_provider.py b/tests/e2e/envoy/mock_provider.py index 66c72ae..8addad4 100755 --- a/tests/e2e/envoy/mock_provider.py +++ b/tests/e2e/envoy/mock_provider.py @@ -17,6 +17,7 @@ import json import sys import threading +import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer MARKER_SHELL = "TOOL_SHELL" @@ -26,6 +27,9 @@ _record_lock = threading.Lock() +EVENT_GAP_SECONDS = 0.05 + + def prompt_text(body): """The concatenated message contents of a chat request, or ''.""" try: @@ -123,19 +127,33 @@ def log_message(self, fmt, *args): sys.stderr.write("mock: " + (fmt % args) + "\n") def read_body(self): - if "chunked" in (self.headers.get("transfer-encoding") or "").lower(): - body = b"" - while True: - size = int(self.rfile.readline().split(b";")[0].strip() or b"0", 16) - if size == 0: + """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() - return body - body += self.rfile.read(size) - self.rfile.readline() - length = int(self.headers.get("content-length") or 0) - return self.rfile.read(length) if length else b"" - - def record(self, body, sent): + 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, @@ -146,6 +164,7 @@ def record(self, body, sent): "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") @@ -170,6 +189,11 @@ def send_stream(self, events, body): 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() diff --git a/tests/e2e/envoy/run.sh b/tests/e2e/envoy/run.sh index 67c31d1..2295432 100755 --- a/tests/e2e/envoy/run.sh +++ b/tests/e2e/envoy/run.sh @@ -175,11 +175,17 @@ envoy_running() { [[ "$(docker inspect -f '{{.State.Running}}' "$CONTAINER" 2>/dev/null)" == true ]] } -# start_envoy +# start_envoy +# The response body mode; the request body follows it (NONE keeps the request +# BUFFERED, so only the response side is misconfigured). Full duplex needs +# both trailer modes SEND — Envoy refuses the config otherwise. start_envoy() { - local cfg="$WORK/envoy-$1.yaml" admin_addr=127.0.0.1 + local cfg="$WORK/envoy-$1.yaml" admin_addr=127.0.0.1 req=$1 trailers=SKIP + [[ "$1" == NONE ]] && req=BUFFERED + [[ "$1" == FULL_DUPLEX_STREAMED ]] && trailers=SEND [[ "$NETWORK" == bridge ]] && admin_addr=0.0.0.0 - sed -e "s/@RESPONSE_BODY_MODE@/$1/" \ + sed -e "s/@RESPONSE_BODY_MODE@/$1/" -e "s/@REQUEST_BODY_MODE@/$req/" \ + -e "s/@TRAILER_MODE@/$trailers/" \ -e "s/@ADMIN_ADDR@/$admin_addr/" -e "s/@ADMIN_PORT@/$ADMIN_PORT/" \ -e "s/@LISTEN_ADDR@/$ENVOY_ADDR/" -e "s/@LISTEN_PORT@/$ENVOY_PORT/" \ -e "s/@UPSTREAM_HOST@/$UPSTREAM_HOST/g" \ @@ -403,8 +409,8 @@ run_misconfigured() { begin none-shell-refused send POST /v1/chat/completions "$(chat 'TOOL_SHELL please')" check "500 (pasu refuses a completion it would never be shown)" status_is 500 - check "pasu_policy_block JSON naming the setting" \ - py_check 'pasu_block() and b"response_body_mode" in b' + check "pasu_misconfigured JSON naming the setting (not a policy block)" \ + py_check 'json.loads(b)["error"]["type"] == "pasu_misconfigured" and b"response_body_mode" in b' check "never reached the provider" reached_times 0 end @@ -429,6 +435,83 @@ run_pasu_down() { stop_envoy } +# ---- phase 6: streamed body modes ------------------------------------------- +# pasu holds a judged body whole in STREAMED and FULL_DUPLEX_STREAMED too. What +# differs is the shape of a refused *streamed completion*: under STREAMED the +# response headers are already downstream, so Envoy resets the stream instead +# of answering 403 — still with no tool-call byte delivered. And in both modes +# the request headers reach the provider before the prompt is judged, so a +# refused prompt opens a request upstream whose body never arrives. + +run_streamed_mode() { # run_streamed_mode + local mode=$1 tag + tag=$(echo "$1" | tr '[:upper:]_' '[:lower:]-') + PHASE=$tag + log "== phase: Envoy $mode, pasu --block-pii-kr ==" + start_pasu --block-pii-kr + start_envoy "$mode" + + begin "$tag-text-completion" + send POST /v1/chat/completions "$(chat 'say hello')" + check "200" status_is 200 + check "the text reply, unchanged" py_check 'len(r) == 1 and b == r[-1]["sent"].encode()' + end + + begin "$tag-text-completion-streamed" + send POST /v1/chat/completions "$(chat 'say hello' true)" + check "200" status_is 200 + check "every SSE data line intact (byte-equal to what the provider sent)" \ + py_check 'len(r) == 1 and b == r[-1]["sent"].encode() and b.endswith(b"data: [DONE]\n\n")' + end + + begin "$tag-shell-tool-call" + send POST /v1/chat/completions "$(chat 'TOOL_SHELL please')" + check "the model did answer with the tool call" reached_times 1 + check "no tool call in what the client got" py_check 'b"call_1" not in b and b"rm -rf" not in b' + check "not delivered as a success" py_check 'pasu_block() or len(b) == 0' + end + + begin "$tag-shell-tool-call-streamed" + send POST /v1/chat/completions "$(chat 'TOOL_SHELL please' true)" + check "the model did stream the tool call" reached_times 1 + check "no SSE chunk delivered" py_check 'b"data:" not in b and b"call_1" not in b' + if [[ "$mode" == FULL_DUPLEX_STREAMED ]]; then + check "a clean 403 (pasu withholds the response headers in full duplex)" status_is 403 + check "pasu_policy_block JSON" is_pasu_block + else + # A reset when the headers were already downstream; a clean 403 when the + # refusal beat them there (a small completion in one piece). Either way + # no tool-call byte — which is the invariant. + check "a reset with no body, or a clean 403" py_check 'len(b) == 0 or pasu_block()' + fi + end + + begin "$tag-pii-rrn-large-prompt-refused" + # A prompt pasu receives in pieces, with the number in the first one: + # forwarding any piece before the whole is judged would send the number. + # Written to a file: one argument that size exceeds Linux's per-argument + # limit (128 KiB), so it goes to curl as @file. + python3 -c 'import json,sys; print(json.dumps({"model":"mock-gpt","messages":[{"role":"user","content":"주민번호 "+sys.argv[1]+" "+"x"*300000}]}, ensure_ascii=False))' \ + "$RRN" >"$WORK/large-prompt.json" + # Sent in two pieces with a pause, the number in the first: curl would send + # it in one write and Envoy would hand pasu one piece. + python3 "$HERE/slow_post.py" 127.0.0.1 "$ENVOY_PORT" /v1/chat/completions \ + "$WORK/large-prompt.json" "$WORK/$CASE.status" "$WORK/$CASE.body" + check "403" status_is 403 + check "the number never reached the provider" py_check "all('$RRN' not in x['body'] for x in r)" + end + + begin "$tag-pii-rrn-refused" + send POST /v1/chat/completions "$(chat "주민번호는 $RRN 입니다")" + check "403" status_is 403 + check "pasu_policy_block JSON" is_pasu_block + check "the number never reached the provider" py_check "all('$RRN' not in x['body'] for x in r)" + end + + stop_envoy + stop_pasu +} + log "pasu-proxy: $PASU_BIN" log "envoy: $ENVOY_IMAGE" log "policy: $POLICY" @@ -440,6 +523,8 @@ run_pii_block run_pii_redact run_misconfigured run_pasu_down +run_streamed_mode STREAMED +run_streamed_mode FULL_DUPLEX_STREAMED log "" log "== summary: $PASSED passed, $FAILED failed ==" diff --git a/tests/e2e/envoy/slow_post.py b/tests/e2e/envoy/slow_post.py new file mode 100644 index 0000000..a7a72d2 --- /dev/null +++ b/tests/e2e/envoy/slow_post.py @@ -0,0 +1,54 @@ +"""POST a JSON body as a chunked request, in pieces with a pause between them. + +curl sends a body in one write, and the data plane then hands it to pasu in one +piece — so a mode that forwards pieces before the whole body is judged is never +tested. This sends the first `split` bytes, waits, then the rest, the way a +slow client or a proxy in front would. + +usage: slow_post.py +""" + +import socket +import sys +import time + +host, port, path, body_file, status_out, body_out = sys.argv[1:7] +body = open(body_file, "rb").read() +split = min(4096, len(body)) +pieces = [body[:split], body[split:]] + +status, response = "000", b"" +try: + with socket.create_connection((host, int(port)), timeout=30) as s: + s.sendall( + b"POST %s HTTP/1.1\r\nhost: %s\r\ncontent-type: application/json\r\n" + b"transfer-encoding: chunked\r\nconnection: close\r\n\r\n" + % (path.encode(), host.encode()) + ) + for piece in pieces: + if piece: + s.sendall(b"%x\r\n%s\r\n" % (len(piece), piece)) + time.sleep(0.3) + s.sendall(b"0\r\n\r\n") + while True: + data = s.recv(65536) + if not data: + break + response += data +except OSError: + pass + +head, _, rest = response.partition(b"\r\n\r\n") +if head.startswith(b"HTTP/"): + status = head.split(b" ")[1].decode() + if b"transfer-encoding: chunked" in head.lower(): + out, buf = b"", rest + while buf: + line, _, buf = buf.partition(b"\r\n") + n = int(line.split(b";")[0] or b"0", 16) + if n == 0: + break + out, buf = out + buf[:n], buf[n + 2 :] + rest = out +open(status_out, "w").write(status) +open(body_out, "wb").write(rest)