codex/codex-rs/exec-server/tests/http_request_logging.rs

use std::io::Write;
use std::sync::Arc;
use std::sync::Mutex;

use codex_exec_server::HttpClient;
use codex_exec_server::HttpRedirectPolicy;
use codex_exec_server::HttpRequestParams;
use codex_exec_server::RouteAwareHttpClient;
use codex_http_client::HttpClientFactory;
use codex_http_client::OutboundProxyPolicy;
use pretty_assertions::assert_eq;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
use tokio::net::TcpListener;
use tracing_subscriber::Layer;
use tracing_subscriber::layer::SubscriberExt;

#[tokio::test(flavor = "current_thread")]
async fn delegated_http_success_logs_do_not_expose_sensitive_request_or_response_data()
-> anyhow::Result<()> {
    let log_buffer = Arc::new(Mutex::new(Vec::new()));
    let writer_buffer = Arc::clone(&log_buffer);
    let subscriber = tracing_subscriber::registry().with(
        tracing_subscriber::fmt::layer()
            .with_ansi(false)
            .with_writer(move || TestLogWriter(Arc::clone(&writer_buffer)))
            .with_filter(
                tracing_subscriber::filter::Targets::new()
                    .with_target("codex_http_client", tracing::Level::TRACE)
                    .with_target("codex_exec_server", tracing::Level::TRACE),
            ),
    );
    let _guard = tracing::subscriber::set_default(subscriber);
    tracing::debug!(target: "codex_exec_server", "log capture sentinel");
    let client =
        RouteAwareHttpClient::new(HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault));

    for (redirect_policy, status, query_secret, cookie_secret, location_secret) in [
        (
            HttpRedirectPolicy::Follow,
            "200 OK",
            "follow-query-secret",
            "follow-cookie-secret",
            "follow-location-secret",
        ),
        (
            HttpRedirectPolicy::Stop,
            "302 Found",
            "stop-query-secret",
            "stop-cookie-secret",
            "stop-location-secret",
        ),
    ] {
        let listener = TcpListener::bind(("127.0.0.1", 0)).await?;
        let address = listener.local_addr()?;
        let response = format!(
            "HTTP/1.1 {status}\r\nSet-Cookie: session={cookie_secret}\r\nLocation: http://127.0.0.1/private?token={location_secret}\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok"
        );
        let server = tokio::spawn(async move {
            let (stream, _) = listener.accept().await?;
            let mut reader = BufReader::new(stream);
            loop {
                let mut line = String::new();
                if reader.read_line(&mut line).await? == 0 {
                    anyhow::bail!("HTTP client disconnected before completing request headers");
                }
                if line == "\r\n" {
                    break;
                }
            }
            reader.get_mut().write_all(response.as_bytes()).await?;
            anyhow::Ok(())
        });

        let response = client
            .http_request(HttpRequestParams {
                method: "GET".to_string(),
                url: format!("http://{address}/delegated?token={query_secret}"),
                headers: Vec::new(),
                body: None,
                timeout_ms: Some(5_000),
                redirect_policy,
                request_id: "sensitive-request".to_string(),
                stream_response: false,
            })
            .await?;
        let expected_status = match redirect_policy {
            HttpRedirectPolicy::Follow => 200,
            HttpRedirectPolicy::Stop => 302,
        };
        assert_eq!(response.status, expected_status);
        server.await??;
    }

    let logs = String::from_utf8(log_buffer.lock().expect("log buffer lock").clone())?;
    assert!(logs.contains("log capture sentinel"));
    for secret in [
        "follow-query-secret",
        "follow-cookie-secret",
        "follow-location-secret",
        "stop-query-secret",
        "stop-cookie-secret",
        "stop-location-secret",
    ] {
        assert!(!logs.contains(secret), "logs exposed {secret}:\n{logs}");
    }

    Ok(())
}

#[tokio::test(flavor = "current_thread")]
async fn delegated_http_failure_warning_redacts_request_url() -> anyhow::Result<()> {
    let log_buffer = Arc::new(Mutex::new(Vec::new()));
    let writer_buffer = Arc::clone(&log_buffer);
    let subscriber = tracing_subscriber::registry().with(
        tracing_subscriber::fmt::layer()
            .with_ansi(false)
            .with_writer(move || TestLogWriter(Arc::clone(&writer_buffer)))
            .with_filter(
                tracing_subscriber::filter::Targets::new()
                    .with_target("codex_http_client", tracing::Level::TRACE)
                    .with_target("codex_exec_server", tracing::Level::TRACE),
            ),
    );
    let _guard = tracing::subscriber::set_default(subscriber);
    let unavailable_server = std::net::TcpListener::bind(("127.0.0.1", 0))?;
    let unavailable_address = unavailable_server.local_addr()?;
    drop(unavailable_server);
    let client =
        RouteAwareHttpClient::new(HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault));

    let error = client
        .http_request(HttpRequestParams {
            method: "GET".to_string(),
            url: format!(
                "http://{unavailable_address}/private-path-secret?token=failure-query-secret"
            ),
            headers: Vec::new(),
            body: None,
            timeout_ms: None,
            redirect_policy: HttpRedirectPolicy::Follow,
            request_id: "failed-sensitive-request".to_string(),
            stream_response: false,
        })
        .await;
    assert!(error.is_err(), "request to a closed port should fail");

    let logs = String::from_utf8(log_buffer.lock().expect("log buffer lock").clone())?;
    assert!(logs.contains("http/request send failed"));
    assert!(logs.contains("error_is_connect=true"));
    for secret in ["private-path-secret", "failure-query-secret"] {
        assert!(!logs.contains(secret), "logs exposed {secret}:\n{logs}");
    }

    Ok(())
}

#[derive(Clone)]
struct TestLogWriter(Arc<Mutex<Vec<u8>>>);

impl Write for TestLogWriter {
    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
        self.0
            .lock()
            .map_err(|_| std::io::Error::other("log buffer lock"))?
            .extend_from_slice(bytes);
        Ok(bytes.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}