OpenShell/e2e/rust/tests/proxy_egress_pipeline.rs

// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

#![cfg(feature = "e2e")]

//! E2E coverage for the transparent sandbox egress pipeline.
//!
//! Workloads connect directly to their requested destinations. Seccomp
//! notification diverts those sockets to the supervisor without proxy
//! environment variables or explicit CONNECT requests. These tests cover:
//! - live policy reloads affect new requests and close a pre-existing HTTP
//!   stream before its next request is forwarded;
//! - `tls: skip` selects a byte-transparent TCP relay;
//! - provider placeholders in HTTP headers and opted-in REST bodies are
//!   resolved without appearing in test output.

use std::io::{self, Error, ErrorKind, Write};
use std::process::Stdio;
use std::sync::{
    Arc, Mutex,
    atomic::{AtomicUsize, Ordering},
};

use openshell_e2e::harness::binary::openshell_cmd;
use openshell_e2e::harness::container::{SupportContainer, e2e_network_name};
use openshell_e2e::harness::sandbox::SandboxGuard;
use serde_json::Value;
use serial_test::serial;
use tempfile::{Builder as TempFileBuilder, NamedTempFile};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::task::JoinHandle;

const TEST_SERVER_HOST: &str = "host.openshell.internal";
const PROVIDER_NAME: &str = "e2e-proxy-egress-credentials";
const PROVIDER_PROFILE_ID: &str = "e2e-proxy-egress-credentials";
const TOKEN_ENV: &str = "PROXY_E2E_TOKEN";
const TEST_SECRET: &str = "sk-e2e-proxy-egress-secret";
const PLACEHOLDER_PREFIX: &str = "openshell:resolve:env:";
const PRIVATE_ALLOWED_IPS: &str = r#"        allowed_ips:
          - "10.0.0.0/8"
          - "172.0.0.0/8"
          - "192.168.0.0/16"
          - "fc00::/7""#;
static PROVIDER_LOCK: Mutex<()> = Mutex::new(());

async fn run_cli(args: &[&str]) -> Result<String, String> {
    let mut cmd = openshell_cmd();
    cmd.args(args).stdout(Stdio::piped()).stderr(Stdio::piped());

    let output = cmd
        .output()
        .await
        .map_err(|error| format!("failed to spawn openshell {}: {error}", args.join(" ")))?;
    let stdout = String::from_utf8_lossy(&output.stdout);
    let stderr = String::from_utf8_lossy(&output.stderr);
    let combined = format!("{stdout}{stderr}");

    if !output.status.success() {
        return Err(format!(
            "openshell {} failed (exit {:?}):\n{combined}",
            args.join(" "),
            output.status.code()
        ));
    }

    Ok(combined)
}

async fn wait_for_sandbox_logs(
    sandbox_name: &str,
    expected: impl Fn(&str) -> bool,
) -> Result<String, String> {
    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);

    loop {
        let logs = run_cli(&[
            "logs",
            sandbox_name,
            "-n",
            "500",
            "--since",
            "2m",
            "--source",
            "sandbox",
        ])
        .await?;
        if expected(&logs) {
            return Ok(logs);
        }
        if tokio::time::Instant::now() >= deadline {
            return Err(format!(
                "timed out waiting for expected sandbox logs:\n{logs}"
            ));
        }
        tokio::time::sleep(std::time::Duration::from_millis(250)).await;
    }
}

async fn delete_provider(name: &str) {
    let mut cmd = openshell_cmd();
    cmd.args(["provider", "delete", name])
        .stdout(Stdio::null())
        .stderr(Stdio::null());
    let _ = cmd.status().await;
}

async fn delete_provider_profile(id: &str) {
    let mut cmd = openshell_cmd();
    cmd.args(["profile", "delete", id])
        .stdout(Stdio::null())
        .stderr(Stdio::null());
    let _ = cmd.status().await;
}

async fn create_bound_provider(name: &str) -> Result<String, String> {
    let credential = format!("{TOKEN_ENV}={TEST_SECRET}");
    run_cli(&[
        "provider",
        "create",
        "--name",
        name,
        "--type",
        PROVIDER_PROFILE_ID,
        "--credential",
        &credential,
    ])
    .await
}

fn write_credential_profile(port: u16) -> Result<NamedTempFile, String> {
    let mut file = TempFileBuilder::new()
        .suffix(".yaml")
        .tempfile()
        .map_err(|error| format!("create provider profile: {error}"))?;
    let profile = format!(
        r#"id: {PROVIDER_PROFILE_ID}
display_name: E2E proxy egress credentials
category: other
credentials:
  - name: proxy_e2e_token
    env_vars: [{TOKEN_ENV}]
    required: true
    auth_style: bearer
    header_name: authorization
endpoints:
  - host: {TEST_SERVER_HOST}
    port: {port}
    path: /probe
    protocol: rest
    access: full
    enforcement: enforce
    request_body_credential_rewrite: true
    allowed_ips:
      - "10.0.0.0/8"
      - "172.0.0.0/8"
      - "192.168.0.0/16"
      - "fc00::/7"
binaries: ["/**"]
"#
    );
    file.write_all(profile.as_bytes())
        .map_err(|error| format!("write provider profile: {error}"))?;
    file.flush()
        .map_err(|error| format!("flush provider profile: {error}"))?;
    Ok(file)
}

fn write_policy_document(
    host: &str,
    port: u16,
    endpoint_options: &str,
    network_middlewares: &str,
) -> Result<NamedTempFile, String> {
    let mut file = NamedTempFile::new().map_err(|error| format!("create policy: {error}"))?;
    let policy = format!(
        r#"version: 1

filesystem_policy:
  include_workdir: true
  read_only:
    - /usr
    - /lib
    - /proc
    - /dev/urandom
    - /app
    - /etc
    - /var/log
  read_write:
    - /sandbox
    - /tmp
    - /dev/null

landlock:
  compatibility: best_effort

process:
  run_as_user: sandbox
  run_as_group: sandbox

{network_middlewares}
network_policies:
  proxy_egress_test:
    name: proxy_egress_test
    endpoints:
      - host: {host}
        port: {port}
{endpoint_options}
{PRIVATE_ALLOWED_IPS}
    binaries:
      - path: "/**"
"#
    );
    file.write_all(policy.as_bytes())
        .map_err(|error| format!("write policy: {error}"))?;
    file.flush()
        .map_err(|error| format!("flush policy: {error}"))?;
    Ok(file)
}

fn write_policy(host: &str, port: u16, endpoint_options: &str) -> Result<NamedTempFile, String> {
    write_policy_document(host, port, endpoint_options, "")
}

fn write_middleware_policy(
    host: &str,
    port: u16,
    endpoint_options: &str,
    on_error: &str,
) -> Result<NamedTempFile, String> {
    let network_middlewares = format!(
        r#"network_middlewares:
  regex-redactor:
    name: Redact API tokens
    middleware: openshell/regex
    order: 10
    config:
      mode: redact
    on_error: {on_error}
    endpoints:
      include: ["{host}"]
      exclude: []
"#
    );
    write_policy_document(host, port, endpoint_options, &network_middlewares)
}

fn write_denied_policy() -> Result<NamedTempFile, String> {
    let mut file = NamedTempFile::new().map_err(|error| format!("create policy: {error}"))?;
    let policy = r#"version: 1

filesystem_policy:
  include_workdir: true
  read_only:
    - /usr
    - /lib
    - /proc
    - /dev/urandom
    - /app
    - /etc
    - /var/log
  read_write:
    - /sandbox
    - /tmp
    - /dev/null

landlock:
  compatibility: best_effort

process:
  run_as_user: sandbox
  run_as_group: sandbox

network_policies: {}
"#;
    file.write_all(policy.as_bytes())
        .map_err(|error| format!("write policy: {error}"))?;
    file.flush()
        .map_err(|error| format!("flush policy: {error}"))?;
    Ok(file)
}

fn write_ambiguous_policy(host: &str, port: u16) -> Result<NamedTempFile, String> {
    let mut file = NamedTempFile::new().map_err(|error| format!("create policy: {error}"))?;
    let policy = format!(
        r#"version: 1

filesystem_policy:
  include_workdir: true
  read_only: [/usr, /lib, /proc, /dev/urandom, /app, /etc, /var/log]
  read_write: [/sandbox, /tmp, /dev/null]

landlock:
  compatibility: best_effort

process:
  run_as_user: sandbox
  run_as_group: sandbox

network_policies:
  terminating:
    name: terminating
    endpoints:
      - host: {host}
        port: {port}
{PRIVATE_ALLOWED_IPS}
    binaries:
      - path: "/**"
  passthrough:
    name: passthrough
    endpoints:
      - host: {host}
        port: {port}
        tls: skip
{PRIVATE_ALLOWED_IPS}
    binaries:
      - path: "/**"
"#
    );
    file.write_all(policy.as_bytes())
        .map_err(|error| format!("write policy: {error}"))?;
    file.flush()
        .map_err(|error| format!("flush policy: {error}"))?;
    Ok(file)
}

fn write_destination_denial_policy() -> Result<NamedTempFile, String> {
    let mut file = NamedTempFile::new().map_err(|error| format!("create policy: {error}"))?;
    let policy = r#"version: 1

filesystem_policy:
  include_workdir: true
  read_only: [/usr, /lib, /proc, /dev/urandom, /app, /etc, /var/log]
  read_write: [/sandbox, /tmp, /dev/null]

landlock:
  compatibility: best_effort

process:
  run_as_user: sandbox
  run_as_group: sandbox

network_policies:
  destination_denials:
    name: destination_denials
    endpoints:
      - { host: 169.254.169.254, port: 80 }
      - { host: 127.0.0.1, port: 80 }
      - { host: 203.0.113.10, port: 6443 }
      - host: 203.0.113.10
        port: 8080
        allowed_ips: ["198.51.100.0/24"]
    binaries:
      - path: "/**"
"#;
    file.write_all(policy.as_bytes())
        .map_err(|error| format!("write policy: {error}"))?;
    file.flush()
        .map_err(|error| format!("flush policy: {error}"))?;
    Ok(file)
}

fn write_ip_literal_success_policy(
    explicit_ip: &str,
    implicit_ip: &str,
    port: u16,
) -> Result<NamedTempFile, String> {
    let mut file = NamedTempFile::new().map_err(|error| format!("create policy: {error}"))?;
    let policy = format!(
        r#"version: 1

filesystem_policy:
  include_workdir: true
  read_only: [/usr, /lib, /proc, /dev/urandom, /app, /etc, /var/log]
  read_write: [/sandbox, /tmp, /dev/null]

landlock:
  compatibility: best_effort

process:
  run_as_user: sandbox
  run_as_group: sandbox

network_policies:
  destination_successes:
    name: destination_successes
    endpoints:
      - host: {ip}
        port: {port}
        allowed_ips: ["{explicit_ip}/32"]
      - host: {implicit_ip}
        port: {port}
    binaries:
      - path: "/**"
"#,
        ip = explicit_ip,
    );
    file.write_all(policy.as_bytes())
        .map_err(|error| format!("write policy: {error}"))?;
    file.flush()
        .map_err(|error| format!("flush policy: {error}"))?;
    Ok(file)
}

fn policy_path(file: &NamedTempFile) -> String {
    file.path()
        .to_str()
        .expect("temporary policy path should be utf-8")
        .to_string()
}

async fn read_until(stream: &mut TcpStream, marker: &[u8]) -> io::Result<Vec<u8>> {
    let mut data = Vec::new();
    let mut buffer = [0_u8; 4096];
    loop {
        let read = stream.read(&mut buffer).await?;
        if read == 0 {
            return Ok(data);
        }
        data.extend_from_slice(&buffer[..read]);
        if data.windows(marker.len()).any(|window| window == marker) {
            return Ok(data);
        }
    }
}

fn header_end(bytes: &[u8]) -> Option<usize> {
    bytes
        .windows(4)
        .position(|window| window == b"\r\n\r\n")
        .map(|position| position + 4)
}

fn content_length(headers: &[u8]) -> io::Result<usize> {
    let text =
        std::str::from_utf8(headers).map_err(|error| Error::new(ErrorKind::InvalidData, error))?;
    Ok(text
        .lines()
        .find_map(|line| {
            let (name, value) = line.split_once(':')?;
            name.trim()
                .eq_ignore_ascii_case("content-length")
                .then(|| value.trim().parse::<usize>().ok())
                .flatten()
        })
        .unwrap_or(0))
}

async fn read_http_request(stream: &mut TcpStream) -> io::Result<Option<Vec<u8>>> {
    let mut request = read_until(stream, b"\r\n\r\n").await?;
    if request.is_empty() {
        return Ok(None);
    }
    let headers_end = header_end(&request)
        .ok_or_else(|| Error::new(ErrorKind::UnexpectedEof, "incomplete HTTP headers"))?;
    let body_length = content_length(&request[..headers_end])?;
    let total_length = headers_end + body_length;
    while request.len() < total_length {
        let mut buffer = vec![0_u8; total_length - request.len()];
        let read = stream.read(&mut buffer).await?;
        if read == 0 {
            return Err(Error::new(ErrorKind::UnexpectedEof, "incomplete HTTP body"));
        }
        request.extend_from_slice(&buffer[..read]);
    }
    request.truncate(total_length);
    Ok(Some(request))
}

struct KeepAliveHttpServer {
    port: u16,
    connections: Arc<AtomicUsize>,
    task: JoinHandle<()>,
}

impl KeepAliveHttpServer {
    async fn start() -> Result<Self, String> {
        let listener = TcpListener::bind(("0.0.0.0", 0))
            .await
            .map_err(|error| format!("bind HTTP server: {error}"))?;
        let port = listener
            .local_addr()
            .map_err(|error| format!("read HTTP server address: {error}"))?
            .port();
        let connections = Arc::new(AtomicUsize::new(0));
        let task_connections = Arc::clone(&connections);
        let task = tokio::spawn(async move {
            while let Ok((stream, _)) = listener.accept().await {
                task_connections.fetch_add(1, Ordering::AcqRel);
                tokio::spawn(async move {
                    let _ = handle_keep_alive_connection(stream).await;
                });
            }
        });
        Ok(Self {
            port,
            connections,
            task,
        })
    }

    fn connection_count(&self) -> usize {
        self.connections.load(Ordering::Acquire)
    }
}

impl Drop for KeepAliveHttpServer {
    fn drop(&mut self) {
        self.task.abort();
    }
}

async fn handle_keep_alive_connection(mut stream: TcpStream) -> io::Result<()> {
    while let Some(request) = read_http_request(&mut stream).await? {
        let close = String::from_utf8_lossy(&request)
            .lines()
            .any(|line| line.eq_ignore_ascii_case("connection: close"));
        let connection = if close { "close" } else { "keep-alive" };
        let response =
            format!("HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: {connection}\r\n\r\nok");
        stream.write_all(response.as_bytes()).await?;
        if close {
            return Ok(());
        }
    }
    Ok(())
}

struct EchoServer {
    port: u16,
    observed: Arc<Mutex<Vec<u8>>>,
    task: JoinHandle<()>,
}

struct RequestBodyEchoServer {
    port: u16,
    task: JoinHandle<()>,
}

impl RequestBodyEchoServer {
    async fn start() -> Result<Self, String> {
        let listener = TcpListener::bind(("0.0.0.0", 0))
            .await
            .map_err(|error| format!("bind request body echo server: {error}"))?;
        let port = listener
            .local_addr()
            .map_err(|error| format!("read request body echo server address: {error}"))?
            .port();
        let task = tokio::spawn(async move {
            while let Ok((stream, _)) = listener.accept().await {
                tokio::spawn(async move {
                    let _ = handle_request_body_echo(stream).await;
                });
            }
        });
        Ok(Self { port, task })
    }
}

impl Drop for RequestBodyEchoServer {
    fn drop(&mut self) {
        self.task.abort();
    }
}

async fn handle_request_body_echo(mut stream: TcpStream) -> io::Result<()> {
    let request = read_http_request(&mut stream)
        .await?
        .ok_or_else(|| Error::new(ErrorKind::UnexpectedEof, "missing HTTP request"))?;
    let headers_end = header_end(&request)
        .ok_or_else(|| Error::new(ErrorKind::UnexpectedEof, "incomplete HTTP headers"))?;
    let body = &request[headers_end..];
    let response = format!(
        "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
        body.len()
    );
    stream.write_all(response.as_bytes()).await?;
    stream.write_all(body).await
}

struct PipelineProbeServer {
    port: u16,
    observed: Arc<Mutex<Vec<u8>>>,
    task: JoinHandle<()>,
}

impl PipelineProbeServer {
    async fn start() -> Result<Self, String> {
        let listener = TcpListener::bind(("0.0.0.0", 0))
            .await
            .map_err(|error| format!("bind pipeline probe: {error}"))?;
        let port = listener
            .local_addr()
            .map_err(|error| format!("read pipeline probe address: {error}"))?
            .port();
        let observed = Arc::new(Mutex::new(Vec::new()));
        let observed_task = observed.clone();
        let task = tokio::spawn(async move {
            while let Ok((mut stream, _)) = listener.accept().await {
                let observed = observed_task.clone();
                tokio::spawn(async move {
                    let mut request = Vec::new();
                    let mut buffer = [0_u8; 4096];
                    loop {
                        match tokio::time::timeout(
                            std::time::Duration::from_millis(200),
                            stream.read(&mut buffer),
                        )
                        .await
                        {
                            Ok(Ok(0)) | Err(_) => break,
                            Ok(Ok(read)) => request.extend_from_slice(&buffer[..read]),
                            Ok(Err(_)) => return,
                        }
                    }
                    observed.lock().unwrap().extend_from_slice(&request);
                    // Keep the first response reusable so the queued request reaches policy evaluation.
                    let _ = stream
                        .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
                        .await;
                });
            }
        });
        Ok(Self {
            port,
            observed,
            task,
        })
    }

    fn observed_request(&self) -> Vec<u8> {
        self.observed.lock().unwrap().clone()
    }
}

impl Drop for PipelineProbeServer {
    fn drop(&mut self) {
        self.task.abort();
    }
}

impl EchoServer {
    async fn start() -> Result<Self, String> {
        let listener = TcpListener::bind(("0.0.0.0", 0))
            .await
            .map_err(|error| format!("bind echo server: {error}"))?;
        let port = listener
            .local_addr()
            .map_err(|error| format!("read echo server address: {error}"))?
            .port();
        let observed = Arc::new(Mutex::new(Vec::new()));
        let task_observed = Arc::clone(&observed);
        let task = tokio::spawn(async move {
            while let Ok((mut stream, _)) = listener.accept().await {
                let observed = Arc::clone(&task_observed);
                tokio::spawn(async move {
                    let mut buffer = [0_u8; 4096];
                    loop {
                        let Ok(read) = stream.read(&mut buffer).await else {
                            break;
                        };
                        if read == 0 {
                            break;
                        }
                        observed.lock().unwrap().extend_from_slice(&buffer[..read]);
                        if stream.write_all(&buffer[..read]).await.is_err() {
                            break;
                        }
                    }
                });
            }
        });
        Ok(Self {
            port,
            observed,
            task,
        })
    }

    fn observed_bytes(&self) -> Vec<u8> {
        self.observed.lock().unwrap().clone()
    }
}

impl Drop for EchoServer {
    fn drop(&mut self) {
        self.task.abort();
    }
}

struct CredentialProbeServer {
    port: u16,
    task: JoinHandle<()>,
}

impl CredentialProbeServer {
    async fn start() -> Result<Self, String> {
        let listener = TcpListener::bind(("0.0.0.0", 0))
            .await
            .map_err(|error| format!("bind credential probe: {error}"))?;
        let port = listener
            .local_addr()
            .map_err(|error| format!("read credential probe address: {error}"))?
            .port();
        let task = tokio::spawn(async move {
            while let Ok((stream, _)) = listener.accept().await {
                tokio::spawn(async move {
                    let _ = handle_credential_probe(stream).await;
                });
            }
        });
        Ok(Self { port, task })
    }
}

impl Drop for CredentialProbeServer {
    fn drop(&mut self) {
        self.task.abort();
    }
}

async fn handle_credential_probe(mut stream: TcpStream) -> io::Result<()> {
    let request = read_http_request(&mut stream)
        .await?
        .ok_or_else(|| Error::new(ErrorKind::UnexpectedEof, "missing HTTP request"))?;
    let headers_end = header_end(&request)
        .ok_or_else(|| Error::new(ErrorKind::UnexpectedEof, "incomplete HTTP headers"))?;
    let headers = String::from_utf8_lossy(&request[..headers_end]);
    let expected_authorization = format!("Bearer {TEST_SECRET}");
    let header_resolved = headers.lines().any(|line| {
        line.split_once(':').is_some_and(|(name, value)| {
            name.eq_ignore_ascii_case("authorization") && value.trim() == expected_authorization
        })
    });
    let body_resolved = request[headers_end..]
        .windows(TEST_SECRET.len())
        .any(|window| window == TEST_SECRET.as_bytes());
    let saw_placeholder = request
        .windows(PLACEHOLDER_PREFIX.len())
        .any(|window| window == PLACEHOLDER_PREFIX.as_bytes());
    let body = serde_json::json!({
        "body_resolved": body_resolved,
        "header_resolved": header_resolved,
        "saw_placeholder": saw_placeholder,
    })
    .to_string();
    let response = format!(
        "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    );
    stream.write_all(response.as_bytes()).await
}

fn transparent_status_script(host: &str, port: u16) -> String {
    format!(
        r#"
import json
import socket

HOST = {host:?}
PORT = {port}

def read_headers(sock):
    data = b""
    while b"\r\n\r\n" not in data:
        chunk = sock.recv(4096)
        if not chunk:
            break
        data += chunk
    return data

def status(response):
    parts = response.split(None, 2)
    return int(parts[1]) if len(parts) > 1 else 0

def request_status(path):
    target = f"{{HOST}}:{{PORT}}"
    try:
        with socket.create_connection((HOST, PORT), timeout=10) as sock:
            sock.sendall(
                f"GET {{path}} HTTP/1.1\r\n"
                f"Host: {{target}}\r\nConnection: close\r\n\r\n".encode()
            )
            return status(read_headers(sock))
    except OSError as error:
        return {{"errno": error.errno, "error": repr(error)}}

print(json.dumps({{
    "first": request_status("/first"),
    "second": request_status("/second"),
}}, sort_keys=True))
"#,
        host = host,
        port = port,
    )
}

fn persistent_transparent_script(host: &str, port: u16) -> String {
    format!(
        r#"
import json
import os
import socket
import time

HOST = {host:?}
PORT = {port}
READY = "/tmp/proxy-reload-ready"
GO = "/tmp/proxy-reload-go"
RESULT = "/tmp/proxy-reload-result"

def read_response(sock):
    data = b""
    while b"\r\n\r\n" not in data:
        chunk = sock.recv(4096)
        if not chunk:
            return 0
        data += chunk
    headers, body = data.split(b"\r\n\r\n", 1)
    length = 0
    for line in headers.split(b"\r\n")[1:]:
        if line.lower().startswith(b"content-length:"):
            length = int(line.split(b":", 1)[1].strip())
    while len(body) < length:
        chunk = sock.recv(4096)
        if not chunk:
            return 0
        body += chunk
    return int(headers.split(None, 2)[1])

target = f"{{HOST}}:{{PORT}}"
failed_closed = False
second_status = 0
try:
    with socket.create_connection((HOST, PORT), timeout=10) as sock:
        sock.sendall(
            f"GET /before-reload HTTP/1.1\r\nHost: {{target}}\r\nConnection: keep-alive\r\n\r\n".encode()
        )
        if read_response(sock) != 200:
            raise RuntimeError("initial tunneled request was denied")
        open(READY, "w").close()
        deadline = time.monotonic() + 120
        while not os.path.exists(GO) and time.monotonic() < deadline:
            time.sleep(0.1)
        if not os.path.exists(GO):
            raise RuntimeError("timed out waiting for policy reload signal")
        try:
            sock.sendall(
                f"GET /after-reload HTTP/1.1\r\nHost: {{target}}\r\nConnection: close\r\n\r\n".encode()
            )
            second_status = read_response(sock)
        except OSError:
            second_status = 0
        failed_closed = second_status != 200
finally:
    with open(RESULT, "w") as result:
        json.dump({{"failed_closed": failed_closed, "second_status": second_status}}, result, sort_keys=True)
"#,
        host = host,
        port = port,
    )
}

async fn wait_for_sandbox_file(guard: &SandboxGuard, path: &str, log_path: &str) -> String {
    let script = format!(
        r#"import os, time
deadline = time.monotonic() + 60
while not os.path.exists({path:?}) and time.monotonic() < deadline:
    time.sleep(0.1)
if not os.path.exists({path:?}):
    if os.path.exists({log_path:?}):
        print(open({log_path:?}).read())
    raise SystemExit("timed out waiting for {path}")
print(open({path:?}).read())
"#
    );
    guard
        .exec(&["python3", "-c", &script])
        .await
        .unwrap_or_else(|error| panic!("wait for sandbox file {path}: {error}"))
}

fn parse_json_line(output: &str) -> Value {
    output
        .lines()
        .filter_map(|line| serde_json::from_str::<Value>(line.trim()).ok())
        .next_back()
        .unwrap_or_else(|| panic!("missing JSON result in sandbox output:\n{output}"))
}

#[tokio::test]
#[serial(proxy_egress_pipeline)]
async fn policy_reload_updates_transparent_requests_and_closes_existing_http_stream() {
    let server = KeepAliveHttpServer::start()
        .await
        .expect("start keep-alive HTTP server");
    let policy_a = write_policy(TEST_SERVER_HOST, server.port, "").expect("write policy A");
    let policy_b = write_denied_policy().expect("write policy B");
    let policy_a_path = policy_path(&policy_a);
    let policy_b_path = policy_path(&policy_b);

    let mut guard = SandboxGuard::create_keep_with_args(
        &["--policy", &policy_a_path],
        &["sh", "-c", "echo Ready; sleep infinity"],
        "Ready",
    )
    .await
    .expect("create keep sandbox");

    run_cli(&[
        "policy",
        "set",
        &guard.name,
        "--policy",
        &policy_a_path,
        "--wait",
        "--timeout",
        "120",
    ])
    .await
    .expect("wait for policy A");

    let persistent_script = persistent_transparent_script(TEST_SERVER_HOST, server.port);
    guard
        .exec(&[
            "sh",
            "-c",
            "nohup python3 -c \"$1\" >/tmp/proxy-reload-client.log 2>&1 &",
            "proxy-reload-client",
            &persistent_script,
        ])
        .await
        .expect("start persistent transparent client");
    wait_for_sandbox_file(
        &guard,
        "/tmp/proxy-reload-ready",
        "/tmp/proxy-reload-client.log",
    )
    .await;

    let status_script = transparent_status_script(TEST_SERVER_HOST, server.port);
    let before = guard
        .exec(&["python3", "-c", &status_script])
        .await
        .expect("exercise transparent requests before reload");
    let before = parse_json_line(&before);
    assert_eq!(
        before["first"], 200,
        "first request before reload: {before}"
    );
    assert_eq!(
        before["second"], 200,
        "second request before reload: {before}"
    );

    run_cli(&[
        "policy",
        "set",
        &guard.name,
        "--policy",
        &policy_b_path,
        "--wait",
        "--timeout",
        "120",
    ])
    .await
    .expect("publish and wait for policy B");

    guard
        .exec(&["sh", "-c", "touch /tmp/proxy-reload-go"])
        .await
        .expect("release persistent transparent client");
    let stale_tunnel = wait_for_sandbox_file(
        &guard,
        "/tmp/proxy-reload-result",
        "/tmp/proxy-reload-client.log",
    )
    .await;
    let stale_tunnel = parse_json_line(&stale_tunnel);
    assert_eq!(
        stale_tunnel["failed_closed"], true,
        "existing transparent HTTP stream forwarded after policy reload: {stale_tunnel}"
    );

    let after = guard
        .exec(&["python3", "-c", &status_script])
        .await
        .expect("exercise transparent requests after reload");
    let after = parse_json_line(&after);
    assert_ne!(after["first"], 200, "first request after reload: {after}");
    assert_ne!(after["second"], 200, "second request after reload: {after}");

    guard.cleanup().await;
}

#[tokio::test]
#[serial(proxy_egress_pipeline)]
async fn ambiguous_policy_update_is_rejected_without_replacing_active_policy() {
    let server = KeepAliveHttpServer::start()
        .await
        .expect("start keep-alive HTTP server");
    let valid_policy = write_policy(TEST_SERVER_HOST, server.port, "").expect("write valid policy");
    let ambiguous_policy =
        write_ambiguous_policy(TEST_SERVER_HOST, server.port).expect("write ambiguous policy");
    let valid_policy_path = policy_path(&valid_policy);
    let ambiguous_policy_path = policy_path(&ambiguous_policy);

    let mut guard = SandboxGuard::create_keep_with_args(
        &["--policy", &valid_policy_path],
        &["sh", "-c", "echo Ready; sleep infinity"],
        "Ready",
    )
    .await
    .expect("create keep sandbox");

    run_cli(&[
        "policy",
        "set",
        &guard.name,
        "--policy",
        &valid_policy_path,
        "--wait",
        "--timeout",
        "120",
    ])
    .await
    .expect("wait for valid policy");

    let status_script = transparent_status_script(TEST_SERVER_HOST, server.port);
    let before = guard
        .exec(&["python3", "-c", &status_script])
        .await
        .expect("exercise transparent requests before invalid update");
    let before = parse_json_line(&before);
    assert_eq!(
        before["first"], 200,
        "first request before update: {before}"
    );
    assert_eq!(
        before["second"], 200,
        "second request before update: {before}"
    );
    let history_before = run_cli(&["policy", "list", &guard.name])
        .await
        .expect("list policy history before rejected update");
    let connections_before_rejection = server.connection_count();

    let update_error = run_cli(&[
        "policy",
        "set",
        &guard.name,
        "--policy",
        &ambiguous_policy_path,
        "--wait",
        "--timeout",
        "120",
    ])
    .await
    .expect_err("ambiguous policy must be rejected before persistence");
    assert!(
        update_error.contains("ambiguity validation failed"),
        "policy update should explain the ambiguity:\n{update_error}"
    );

    let history_after = run_cli(&["policy", "list", &guard.name])
        .await
        .expect("list policy history after rejected update");
    assert_eq!(
        history_after, history_before,
        "rejected policy must not create a revision"
    );

    let after_rejection = guard
        .exec(&["python3", "-c", &status_script])
        .await
        .expect("exercise transparent requests after rejected update");
    let after_rejection = parse_json_line(&after_rejection);
    assert_eq!(
        after_rejection["first"], 200,
        "first request should keep using the active valid policy: {after_rejection}"
    );
    assert_eq!(
        after_rejection["second"], 200,
        "second request should keep using the active valid policy: {after_rejection}"
    );
    assert!(
        server.connection_count() > connections_before_rejection,
        "active-policy requests should still contact the upstream server"
    );

    guard.cleanup().await;
}

#[tokio::test]
#[serial(proxy_egress_pipeline)]
async fn transparent_destination_denials_fail_connect_with_eacces() {
    let policy = write_destination_denial_policy().expect("write destination denial policy");
    let policy_path = policy_path(&policy);
    let script = r#"
import json
import socket

targets = {
    "metadata": ("169.254.169.254", 80),
    "control_plane": ("203.0.113.10", 6443),
    "outside_allowed_ips": ("203.0.113.10", 8080),
}
result = {}
for name, target in targets.items():
    try:
        with socket.create_connection(target, timeout=10):
            result[name] = 0
    except OSError as error:
        result[name] = error.errno
print(json.dumps(result, sort_keys=True))
"#;

    let guard = SandboxGuard::create(&["--policy", &policy_path, "--", "python3", "-c", script])
        .await
        .expect("sandbox create");
    let result = parse_json_line(&guard.create_output);
    for name in ["metadata", "control_plane", "outside_allowed_ips"] {
        assert_eq!(result[name], 13, "{name} should fail with EACCES: {result}");
    }
}

#[tokio::test]
#[serial(proxy_egress_pipeline)]
async fn explicit_allowed_ips_and_implicit_ip_literals_succeed_transparently() {
    if e2e_network_name().is_none() {
        eprintln!("skipping IP-literal success assertions without a shared container network");
        return;
    }

    const HTTP_SERVER: &str = r#"
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body = b"ok"
        self.send_response(200)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format, *args):
        pass

ThreadingHTTPServer(("0.0.0.0", 8000), Handler).serve_forever()
"#;
    let explicit_server =
        SupportContainer::start_python("explicit-ip.openshell.test", HTTP_SERVER, 8000)
            .await
            .expect("start explicit allowed_ips support container");
    let implicit_server =
        SupportContainer::start_python("implicit-ip.openshell.test", HTTP_SERVER, 8000)
            .await
            .expect("start implicit IP-literal support container");
    let explicit_ip = explicit_server.ip().expect("explicit support container IP");
    let implicit_ip = implicit_server.ip().expect("implicit support container IP");
    let policy = write_ip_literal_success_policy(&explicit_ip, &implicit_ip, 8000)
        .expect("write IP literal policy");
    let policy_path = policy_path(&policy);
    let mut guard = SandboxGuard::create_keep_with_args(
        &["--policy", &policy_path],
        &["sh", "-c", "echo Ready; sleep infinity"],
        "Ready",
    )
    .await
    .expect("create keep sandbox");

    for (mode, destination) in [
        ("explicit_allowed_ips", explicit_ip.as_str()),
        ("implicit_ip_literal", implicit_ip.as_str()),
    ] {
        let output = guard
            .exec(&[
                "python3",
                "-c",
                &transparent_status_script(destination, 8000),
            ])
            .await
            .unwrap_or_else(|error| panic!("exercise {mode}: {error}"));
        let statuses = parse_json_line(&output);
        assert_eq!(statuses["first"], 200, "{mode} first request: {statuses}");
        assert_eq!(statuses["second"], 200, "{mode} second request: {statuses}");
    }

    guard.cleanup().await;
}

#[tokio::test]
#[serial(proxy_egress_pipeline)]
async fn tls_skip_connect_relays_opaque_bytes_bidirectionally() {
    let server = EchoServer::start().await.expect("start TCP echo server");
    let policy = write_policy(TEST_SERVER_HOST, server.port, "        tls: skip")
        .expect("write tls: skip policy");
    let policy_path = policy_path(&policy);
    let script = format!(
        r#"
import socket

HOST = {host:?}
PORT = {port}
PAYLOAD = bytes([0x00, 0xff, 0x13, 0x37, 0x80, 0x0a]) + b"not-http-or-tls" + bytes(range(64))

with socket.create_connection((HOST, PORT), timeout=10) as sock:
    sock.sendall(PAYLOAD)
    echoed = b""
    while len(echoed) < len(PAYLOAD):
        chunk = sock.recv(len(PAYLOAD) - len(echoed))
        if not chunk:
            break
        echoed += chunk
    if echoed != PAYLOAD:
        raise RuntimeError("opaque payload changed in transit")
print("RAW_RELAY_OK")
"#,
        host = TEST_SERVER_HOST,
        port = server.port,
    );

    let guard = SandboxGuard::create(&["--policy", &policy_path, "--", "python3", "-c", &script])
        .await
        .expect("sandbox create");
    assert!(
        guard.create_output.contains("RAW_RELAY_OK"),
        "raw relay did not preserve the opaque payload:\n{}",
        guard.create_output
    );
}

#[tokio::test]
#[serial(proxy_egress_pipeline)]
async fn middleware_redacts_transparent_request_bodies() {
    let server = RequestBodyEchoServer::start()
        .await
        .expect("start request body echo server");
    let policy = write_middleware_policy(TEST_SERVER_HOST, server.port, "", "fail_closed")
        .expect("write middleware policy");
    let policy_path = policy_path(&policy);
    let script = format!(
        r#"
import json
import socket

HOST = {host:?}
PORT = {port}
SECRET = "sk-1234567890abcdef"

def read_response(sock):
    data = b""
    while b"\r\n\r\n" not in data:
        chunk = sock.recv(4096)
        if not chunk:
            raise RuntimeError("incomplete response headers")
        data += chunk
    headers, body = data.split(b"\r\n\r\n", 1)
    length = 0
    for line in headers.split(b"\r\n")[1:]:
        if line.lower().startswith(b"content-length:"):
            length = int(line.split(b":", 1)[1].strip())
    while len(body) < length:
        chunk = sock.recv(4096)
        if not chunk:
            break
        body += chunk
    status = int(headers.split(None, 2)[1])
    if status != 200:
        raise RuntimeError(f"request failed with HTTP {{status}}: {{body!r}}")
    return json.loads(body[:length])

def request_bytes(target):
    body = json.dumps({{"api_key": SECRET}}, separators=(",", ":")).encode()
    return (
        f"POST {{target}} HTTP/1.1\r\n"
        f"Host: {{HOST}}:{{PORT}}\r\n"
        "Content-Type: application/json\r\n"
        f"Content-Length: {{len(body)}}\r\n"
        "Connection: close\r\n\r\n"
    ).encode() + body

def request_once():
    with socket.create_connection((HOST, PORT), timeout=10) as sock:
        sock.sendall(request_bytes("/middleware"))
        return read_response(sock)

print(json.dumps({{"first": request_once(), "second": request_once()}}, sort_keys=True))
"#,
        host = TEST_SERVER_HOST,
        port = server.port,
    );

    let guard = SandboxGuard::create(&["--policy", &policy_path, "--", "python3", "-c", &script])
        .await
        .expect("sandbox create");
    let result = parse_json_line(&guard.create_output);
    for request in ["first", "second"] {
        assert_eq!(
            result[request]["api_key"], "[REDACTED]",
            "{request} did not deliver the middleware-transformed body: {result}"
        );
    }
}

#[tokio::test]
#[serial(proxy_egress_pipeline)]
async fn fail_closed_middleware_blocks_uninspectable_transparent_payload_before_upstream() {
    let server = EchoServer::start().await.expect("start TCP echo server");
    let policy = write_middleware_policy(TEST_SERVER_HOST, server.port, "", "fail_closed")
        .expect("write fail-closed middleware policy");
    let policy_path = policy_path(&policy);
    let script = format!(
        r#"
import socket

HOST = {host:?}
PORT = {port}
PAYLOAD = bytes([0x00, 0xff, 0x13, 0x37]) + b"not-http-or-tls"

with socket.create_connection((HOST, PORT), timeout=10) as sock:
    sock.sendall(PAYLOAD)
    denial = b""
    while True:
        try:
            chunk = sock.recv(4096)
        except ConnectionResetError:
            break
        if not chunk:
            break
        denial += chunk
    if denial and (
        b"HTTP/1.1 403 Forbidden" not in denial
        or b"unsupported_l7_protocol" not in denial
    ):
        raise RuntimeError(f"missing fail-closed middleware denial: {{denial!r}}")
print("UNINSPECTABLE_MIDDLEWARE_BLOCKED")
"#,
        host = TEST_SERVER_HOST,
        port = server.port,
    );

    let mut guard = SandboxGuard::create_keep_with_args(
        &["--policy", &policy_path],
        &["sh", "-c", "echo Ready; sleep infinity"],
        "Ready",
    )
    .await
    .expect("create keep sandbox");
    let output = guard
        .exec(&["python3", "-c", &script])
        .await
        .expect("exercise uninspectable fail-closed middleware");
    assert!(
        output.contains("UNINSPECTABLE_MIDDLEWARE_BLOCKED"),
        "uninspectable payload was not blocked:\n{output}"
    );
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    assert!(
        server.observed_bytes().is_empty(),
        "uninspectable payload reached upstream before middleware denial"
    );

    let logs = wait_for_sandbox_logs(&guard.name, |logs| {
        logs.contains("openshell.middleware.traffic_uninspectable")
            && logs
                .contains("Unsupported tunnel protocol cannot be inspected by required middleware")
    })
    .await
    .expect("fetch sandbox logs after middleware denial");
    assert!(
        logs.contains("openshell.middleware.traffic_uninspectable")
            && logs
                .contains("Unsupported tunnel protocol cannot be inspected by required middleware"),
        "OCSF logs should explain the fail-closed denial:\n{logs}"
    );

    guard.cleanup().await;
}

#[tokio::test]
#[serial(proxy_egress_pipeline)]
async fn fail_open_middleware_bypasses_uninspectable_transparent_tls_skip() {
    let server = EchoServer::start().await.expect("start TCP echo server");
    let policy = write_middleware_policy(
        TEST_SERVER_HOST,
        server.port,
        "        tls: skip",
        "fail_open",
    )
    .expect("write fail-open middleware policy");
    let policy_path = policy_path(&policy);
    let script = format!(
        r#"
import socket

HOST = {host:?}
PORT = {port}
PAYLOAD = bytes([0x00, 0xff, 0x13, 0x37, 0x80]) + b"middleware-bypass"

with socket.create_connection((HOST, PORT), timeout=10) as sock:
    sock.sendall(PAYLOAD)
    echoed = b""
    while len(echoed) < len(PAYLOAD):
        chunk = sock.recv(len(PAYLOAD) - len(echoed))
        if not chunk:
            break
        echoed += chunk
    if echoed != PAYLOAD:
        raise RuntimeError(f"fail-open middleware did not preserve raw relay: {{echoed!r}}")
print("UNINSPECTABLE_MIDDLEWARE_BYPASSED")
"#,
        host = TEST_SERVER_HOST,
        port = server.port,
    );

    let guard = SandboxGuard::create(&["--policy", &policy_path, "--", "python3", "-c", &script])
        .await
        .expect("sandbox create");
    assert!(
        guard
            .create_output
            .contains("UNINSPECTABLE_MIDDLEWARE_BYPASSED"),
        "fail-open middleware did not bypass uninspectable traffic:\n{}",
        guard.create_output
    );
    assert_eq!(
        server.observed_bytes(),
        [0x00, 0xff, 0x13, 0x37, 0x80]
            .into_iter()
            .chain(*b"middleware-bypass")
            .collect::<Vec<_>>(),
        "upstream did not receive the unchanged fail-open payload"
    );
}

#[tokio::test]
#[serial(proxy_egress_pipeline)]
async fn transparent_pipeline_never_reaches_upstream_as_first_request_overflow() {
    let server = PipelineProbeServer::start()
        .await
        .expect("start pipeline probe server");
    let endpoint_options = r#"        protocol: rest
        enforcement: enforce
        rules:
          - allow:
              method: GET
              path: "/allowed""#;
    let policy = write_policy(TEST_SERVER_HOST, server.port, endpoint_options)
        .expect("write pipeline policy");
    let policy_path = policy_path(&policy);
    let script = format!(
        r#"
import socket

target = "{host}:{port}"
first = (
    f"GET /allowed HTTP/1.1\r\n"
    f"Host: {{target}}\r\nConnection: keep-alive\r\n\r\n"
)
second = (
    f"POST /blocked HTTP/1.1\r\n"
    f"Host: {{target}}\r\nContent-Length: 0\r\n\r\n"
)
with socket.create_connection(({host:?}, {port}), timeout=10) as sock:
    sock.sendall((first + second).encode())
    response = b""
    while True:
        chunk = sock.recv(4096)
        if not chunk:
            break
        response += chunk
responses = response.count(b"HTTP/1.1 ")
first_status = response.split(b"\r\n", 1)[0]
if responses != 2 or b" 200 " not in first_status or b"HTTP/1.1 403 Forbidden" not in response:
    raise RuntimeError(f"unexpected pipelined response: {{response!r}}")
print("TRANSPARENT_PIPELINE_DENIED")
"#,
        host = TEST_SERVER_HOST,
        port = server.port,
    );

    let guard = SandboxGuard::create(&["--policy", &policy_path, "--", "python3", "-c", &script])
        .await
        .expect("sandbox create");
    assert!(
        guard.create_output.contains("TRANSPARENT_PIPELINE_DENIED"),
        "transparent HTTP stream did not deny the disallowed pipelined request:\n{}",
        guard.create_output
    );

    let observed = String::from_utf8(server.observed_request()).expect("upstream HTTP request");
    assert!(observed.starts_with("GET /allowed HTTP/1.1\r\n"));
    assert!(
        !observed.to_ascii_lowercase().contains("\r\nconnection:"),
        "shared relay must remove hop-by-hop connection headers:\n{observed}"
    );
    assert!(!observed.contains("/blocked"));
}

#[tokio::test]
#[serial(proxy_egress_pipeline)]
async fn chunked_pipeline_is_authorized_separately_before_reaching_upstream() {
    let server = PipelineProbeServer::start()
        .await
        .expect("start pipeline probe server");
    let endpoint_options = r#"        protocol: rest
        enforcement: enforce
        rules:
          - allow:
              method: POST
              path: "/allowed""#;
    let policy = write_policy(TEST_SERVER_HOST, server.port, endpoint_options)
        .expect("write pipeline policy");
    let policy_path = policy_path(&policy);
    let script = format!(
        r#"
import socket

target = "{host}:{port}"
first = (
    f"POST /allowed HTTP/1.1\r\n"
    f"Host: {{target}}\r\nTransfer-Encoding: chunked\r\nConnection: keep-alive\r\n\r\n"
    f"0\r\n\r\n"
)
second = (
    f"DELETE /blocked HTTP/1.1\r\n"
    f"Host: {{target}}\r\nContent-Length: 0\r\n\r\n"
)
with socket.create_connection(({host:?}, {port}), timeout=10) as sock:
    sock.sendall((first + second).encode())
    response = b""
    while True:
        chunk = sock.recv(4096)
        if not chunk:
            break
        response += chunk
responses = response.count(b"HTTP/1.1 ")
first_status = response.split(b"\r\n", 1)[0]
if responses != 2 or b" 200 " not in first_status or b"HTTP/1.1 403 Forbidden" not in response:
    raise RuntimeError(f"unexpected chunked pipeline response: {{response!r}}")
print("CHUNKED_PIPELINE_DENIED")
"#,
        host = TEST_SERVER_HOST,
        port = server.port,
    );

    let guard = SandboxGuard::create(&["--policy", &policy_path, "--", "python3", "-c", &script])
        .await
        .expect("sandbox create");
    assert!(
        guard.create_output.contains("CHUNKED_PIPELINE_DENIED"),
        "transparent HTTP stream did not evaluate the chunked pipeline separately:\n{}",
        guard.create_output
    );

    let observed = String::from_utf8(server.observed_request()).expect("upstream HTTP request");
    assert!(observed.starts_with("POST /allowed HTTP/1.1\r\n"));
    assert!(observed.ends_with("0\r\n\r\n"));
    assert!(!observed.contains("/blocked"));
}

#[tokio::test]
#[serial(proxy_egress_pipeline)]
async fn http_credentials_are_rewritten_in_transparent_headers_and_bodies() {
    let _provider_lock = PROVIDER_LOCK
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    delete_provider(PROVIDER_NAME).await;
    delete_provider_profile(PROVIDER_PROFILE_ID).await;

    let result = async {
        let server = CredentialProbeServer::start().await?;
        let profile = write_credential_profile(server.port)?;
        let profile_path = profile.path().to_string_lossy().into_owned();
        run_cli(&["profile", "import", "--file", &profile_path]).await?;
        create_bound_provider(PROVIDER_NAME).await?;
        let endpoint_options = r#"        path: /probe
        protocol: rest
        enforcement: enforce
        request_body_credential_rewrite: true
        access: full"#;
        let policy = write_policy(TEST_SERVER_HOST, server.port, endpoint_options)?;
        let policy_path = policy_path(&policy);
        let script = format!(
            r#"
import json
import os
import socket

HOST = {host:?}
PORT = {port}
TOKEN = os.environ[{token_env:?}]

def read_response(sock):
    data = b""
    while b"\r\n\r\n" not in data:
        chunk = sock.recv(4096)
        if not chunk:
            raise RuntimeError("incomplete response headers")
        data += chunk
    headers, body = data.split(b"\r\n\r\n", 1)
    length = 0
    for line in headers.split(b"\r\n")[1:]:
        if line.lower().startswith(b"content-length:"):
            length = int(line.split(b":", 1)[1].strip())
    while len(body) < length:
        chunk = sock.recv(4096)
        if not chunk:
            break
        body += chunk
    code = int(headers.split(None, 2)[1])
    if code != 200:
        raise RuntimeError(f"request failed with HTTP {{code}}")
    return json.loads(body[:length])

def request_bytes(target):
    body = json.dumps({{"credential": TOKEN}}, separators=(",", ":")).encode()
    return (
        f"POST {{target}} HTTP/1.1\r\n"
        f"Host: {{HOST}}:{{PORT}}\r\n"
        f"Authorization: Bearer {{TOKEN}}\r\n"
        "Content-Type: application/json\r\n"
        f"Content-Length: {{len(body)}}\r\n"
        "Connection: close\r\n\r\n"
    ).encode() + body

def request_once():
    with socket.create_connection((HOST, PORT), timeout=10) as sock:
        sock.sendall(request_bytes("/probe"))
        return read_response(sock)

print(json.dumps({{"first": request_once(), "second": request_once()}}, sort_keys=True))
"#,
            host = TEST_SERVER_HOST,
            port = server.port,
            token_env = TOKEN_ENV,
        );

        SandboxGuard::create(&[
            "--policy",
            &policy_path,
            "--provider",
            PROVIDER_NAME,
            "--",
            "python3",
            "-c",
            &script,
        ])
        .await
    }
    .await;

    delete_provider(PROVIDER_NAME).await;
    delete_provider_profile(PROVIDER_PROFILE_ID).await;

    let guard = result.expect("sandbox create");
    let result = parse_json_line(&guard.create_output);
    for request in ["first", "second"] {
        assert_eq!(
            result[request]["header_resolved"], true,
            "{request} header placeholder was not resolved: {result}"
        );
        assert_eq!(
            result[request]["body_resolved"], true,
            "{request} body placeholder was not resolved: {result}"
        );
        assert_eq!(
            result[request]["saw_placeholder"], false,
            "{request} leaked an unresolved placeholder upstream: {result}"
        );
    }
    assert!(
        !guard.create_output.contains(TEST_SECRET),
        "sandbox output exposed the raw provider credential:\n{}",
        guard.create_output
    );
    assert!(
        !guard.create_output.contains(PLACEHOLDER_PREFIX),
        "sandbox output exposed an unresolved provider placeholder:\n{}",
        guard.create_output
    );
}