test: cover Noise runtime security invariants

Co-authored-by: Codex noreply@openai.com

codex-rs/exec-server/src/noise_channel_tests.rs

@@ -1,8 +1,230 @@
 use pretty_assertions::assert_eq;
 
+use super::InitiatorHandshake;
+use super::MAX_TRANSPORT_RECORDS_PER_DIRECTION;
 use super::NOISE_CHANNEL_SUITE;
+use super::NoiseChannelError;
 use super::NoiseChannelIdentity;
 use super::NoiseChannelPublicKey;
+use super::PendingResponderHandshake;
+use super::noise_channel_prologue;
+
+#[test]
+fn hybrid_ik_roundtrip_authenticates_both_endpoints() {
+    let initiator = NoiseChannelIdentity::generate().expect("generate initiator identity");
+    let responder = NoiseChannelIdentity::generate().expect("generate responder identity");
+    let prologue =
+        noise_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
+    let authorization = b"harness-key-authorization";
+
+    let (initiator_handshake, request) = InitiatorHandshake::start(
+        &initiator,
+        &responder.public_key(),
+        &prologue,
+        authorization,
+    )
+    .expect("start initiator handshake");
+    let mut responder_handshake =
+        PendingResponderHandshake::read_request(&responder, &prologue, &request)
+            .expect("read responder handshake");
+
+    assert_eq!(
+        responder_handshake.initiator_public_key(),
+        &initiator.public_key()
+    );
+    assert_eq!(responder_handshake.take_payload(), authorization);
+
+    let (mut responder_transport, response) = responder_handshake
+        .complete()
+        .expect("complete responder handshake");
+    let mut initiator_transport = initiator_handshake
+        .finish(&response)
+        .expect("complete initiator handshake");
+
+    let request_ciphertext = initiator_transport
+        .encrypt(b"request")
+        .expect("encrypt request");
+    assert_ne!(request_ciphertext, b"request");
+    assert_eq!(
+        responder_transport
+            .decrypt(&request_ciphertext)
+            .expect("decrypt request"),
+        b"request"
+    );
+
+    let response_ciphertext = responder_transport
+        .encrypt(b"response")
+        .expect("encrypt response");
+    assert_ne!(response_ciphertext, b"response");
+    assert_eq!(
+        initiator_transport
+            .decrypt(&response_ciphertext)
+            .expect("decrypt response"),
+        b"response"
+    );
+}
+
+#[test]
+fn initiator_rejects_wrong_responder_key() {
+    let initiator = NoiseChannelIdentity::generate().expect("generate initiator identity");
+    let expected_responder = NoiseChannelIdentity::generate().expect("generate expected identity");
+    let actual_responder = NoiseChannelIdentity::generate().expect("generate actual identity");
+    let prologue =
+        noise_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
+
+    let (_initiator_handshake, request) = InitiatorHandshake::start(
+        &initiator,
+        &expected_responder.public_key(),
+        &prologue,
+        b"authorization",
+    )
+    .expect("start initiator handshake");
+
+    assert!(
+        PendingResponderHandshake::read_request(&actual_responder, &prologue, &request).is_err()
+    );
+}
+
+#[test]
+fn responder_rejects_mismatched_prologue() {
+    let initiator = NoiseChannelIdentity::generate().expect("generate initiator identity");
+    let responder = NoiseChannelIdentity::generate().expect("generate responder identity");
+    let initiator_prologue =
+        noise_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
+    let responder_prologue =
+        noise_channel_prologue("env-1", "registration-1", "stream-2").expect("build prologue");
+    let (_initiator_handshake, request) = InitiatorHandshake::start(
+        &initiator,
+        &responder.public_key(),
+        &initiator_prologue,
+        b"authorization",
+    )
+    .expect("start initiator handshake");
+
+    assert!(
+        PendingResponderHandshake::read_request(&responder, &responder_prologue, &request).is_err()
+    );
+}
+
+#[test]
+fn prologue_encoding_is_stable_and_unambiguous() {
+    let prologue =
+        noise_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
+
+    assert_eq!(
+        prologue,
+        b"\x00\x00\x00\x20codex-exec-server-relay-noise/v1\
+          \x00\x00\x00\x05env-1\
+          \x00\x00\x00\x0eregistration-1\
+          \x00\x00\x00\x08stream-1"
+            .to_vec()
+    );
+}
+
+#[test]
+fn transport_rejects_tampered_ciphertext() {
+    let initiator = NoiseChannelIdentity::generate().expect("generate initiator identity");
+    let responder = NoiseChannelIdentity::generate().expect("generate responder identity");
+    let prologue =
+        noise_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
+    let (initiator_handshake, request) = InitiatorHandshake::start(
+        &initiator,
+        &responder.public_key(),
+        &prologue,
+        b"authorization",
+    )
+    .expect("start initiator handshake");
+    let responder_handshake =
+        PendingResponderHandshake::read_request(&responder, &prologue, &request)
+            .expect("read responder handshake");
+    let (mut responder_transport, response) = responder_handshake
+        .complete()
+        .expect("complete responder handshake");
+    let mut initiator_transport = initiator_handshake
+        .finish(&response)
+        .expect("complete initiator handshake");
+    let mut ciphertext = initiator_transport
+        .encrypt(b"request")
+        .expect("encrypt request");
+    ciphertext[0] ^= 1;
+
+    assert!(responder_transport.decrypt(&ciphertext).is_err());
+}
+
+#[test]
+fn transport_rejects_exhausted_receiving_nonce_before_decryption() {
+    let initiator = NoiseChannelIdentity::generate().expect("generate initiator identity");
+    let responder = NoiseChannelIdentity::generate().expect("generate responder identity");
+    let prologue =
+        noise_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
+    let (initiator_handshake, request) = InitiatorHandshake::start(
+        &initiator,
+        &responder.public_key(),
+        &prologue,
+        b"authorization",
+    )
+    .expect("start initiator handshake");
+    let responder_handshake =
+        PendingResponderHandshake::read_request(&responder, &prologue, &request)
+            .expect("read responder handshake");
+    let (mut responder_transport, response) = responder_handshake
+        .complete()
+        .expect("complete responder handshake");
+    let mut initiator_transport = initiator_handshake
+        .finish(&response)
+        .expect("complete initiator handshake");
+    let ciphertext = initiator_transport
+        .encrypt(b"request")
+        .expect("encrypt request");
+    responder_transport
+        .transport
+        .set_receiving_nonce(MAX_TRANSPORT_RECORDS_PER_DIRECTION);
+
+    assert!(matches!(
+        responder_transport.decrypt(&ciphertext),
+        Err(NoiseChannelError::InvalidState(
+            "transport record nonce exhausted"
+        ))
+    ));
+}
+
+#[test]
+fn transport_rejects_replayed_ciphertext() {
+    let initiator = NoiseChannelIdentity::generate().expect("generate initiator identity");
+    let responder = NoiseChannelIdentity::generate().expect("generate responder identity");
+    let prologue =
+        noise_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
+    let (initiator_handshake, request) = InitiatorHandshake::start(
+        &initiator,
+        &responder.public_key(),
+        &prologue,
+        b"authorization",
+    )
+    .expect("start initiator handshake");
+    let responder_handshake =
+        PendingResponderHandshake::read_request(&responder, &prologue, &request)
+            .expect("read responder handshake");
+    let (mut responder_transport, response) = responder_handshake
+        .complete()
+        .expect("complete responder handshake");
+    let mut initiator_transport = initiator_handshake
+        .finish(&response)
+        .expect("complete initiator handshake");
+    let ciphertext = initiator_transport
+        .encrypt(b"request")
+        .expect("encrypt request");
+
+    assert_eq!(
+        responder_transport
+            .decrypt(&ciphertext)
+            .expect("decrypt request"),
+        b"request"
+    );
+    assert!(matches!(
+        responder_transport.decrypt(&ciphertext),
+        Err(NoiseChannelError::Transport(_))
+    ));
+}
 
 #[test]
 fn public_key_validation_rejects_unknown_suite() {

codex-rs/exec-server/src/noise_relay/environment.rs

@@ -397,3 +397,7 @@ fn send_reset(physical_outgoing_tx: &mpsc::Sender<Vec<u8>>, stream_id: String) {
     // state machine behind an overloaded physical writer queue.
     let _ = physical_outgoing_tx.try_send(encode_relay_message_frame(&reset));
 }
+
+#[cfg(test)]
+#[path = "environment_tests.rs"]
+mod tests;

codex-rs/exec-server/src/noise_relay/environment_tests.rs

@@ -0,0 +1,165 @@
+use std::sync::Arc;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+
+use anyhow::Result;
+use futures::SinkExt;
+use futures::StreamExt;
+use pretty_assertions::assert_eq;
+use tokio::net::TcpListener;
+use tokio::sync::Notify;
+use tokio::time::timeout;
+use tokio_tungstenite::accept_async;
+use tokio_tungstenite::connect_async;
+use tokio_tungstenite::tungstenite::Message;
+
+use super::HarnessKeyValidator;
+use super::MAX_HARNESS_KEY_AUTHORIZATION_BYTES;
+use super::run_noise_multiplexed_environment;
+use crate::ExecServerError;
+use crate::ExecServerRuntimePaths;
+use crate::noise_channel::InitiatorHandshake;
+use crate::noise_channel::NoiseChannelIdentity;
+use crate::noise_channel::NoiseChannelPublicKey;
+use crate::noise_channel::noise_channel_prologue;
+use crate::relay::RelayFrameBodyKind;
+use crate::relay::decode_relay_message_frame;
+use crate::relay::encode_relay_message_frame;
+use crate::relay_proto::RelayMessageFrame;
+use crate::server::ConnectionProcessor;
+
+const ENVIRONMENT_ID: &str = "environment-1";
+const EXECUTOR_REGISTRATION_ID: &str = "registration-1";
+
+#[derive(Clone)]
+struct BlockingValidator {
+    calls: Arc<AtomicUsize>,
+    release: Arc<Notify>,
+}
+
+impl HarnessKeyValidator for BlockingValidator {
+    fn validate_harness_key(
+        &self,
+        _harness_public_key: &NoiseChannelPublicKey,
+        _authorization: &str,
+    ) -> impl std::future::Future<Output = Result<(), ExecServerError>> + Send {
+        let calls = Arc::clone(&self.calls);
+        let release = Arc::clone(&self.release);
+        async move {
+            calls.fetch_add(1, Ordering::SeqCst);
+            release.notified().await;
+            Ok(())
+        }
+    }
+}
+
+#[tokio::test]
+async fn pending_harness_key_validation_does_not_block_new_handshakes() -> Result<()> {
+    let listener = TcpListener::bind("127.0.0.1:0").await?;
+    let websocket_url = format!("ws://{}", listener.local_addr()?);
+    let harness_connection = tokio::spawn(connect_async(websocket_url));
+    let (socket, _peer_addr) = listener.accept().await?;
+    let environment_websocket = accept_async(socket).await?;
+    let (mut harness_websocket, _response) = harness_connection.await??;
+
+    let environment_identity = NoiseChannelIdentity::generate()?;
+    let harness_identity = NoiseChannelIdentity::generate()?;
+    let calls = Arc::new(AtomicUsize::new(0));
+    let environment_task = tokio::spawn(run_noise_multiplexed_environment(
+        environment_websocket,
+        ConnectionProcessor::new(ExecServerRuntimePaths::new(
+            std::env::current_exe()?,
+            /*codex_linux_sandbox_exe*/ None,
+        )?),
+        ENVIRONMENT_ID.to_string(),
+        EXECUTOR_REGISTRATION_ID.to_string(),
+        environment_identity.clone(),
+        BlockingValidator {
+            calls: Arc::clone(&calls),
+            release: Arc::new(Notify::new()),
+        },
+    ));
+
+    for stream_id in ["stream-1", "stream-2"] {
+        let prologue =
+            noise_channel_prologue(ENVIRONMENT_ID, EXECUTOR_REGISTRATION_ID, stream_id)?;
+        let (_handshake, request) = InitiatorHandshake::start(
+            &harness_identity,
+            &environment_identity.public_key(),
+            &prologue,
+            b"authorization",
+        )?;
+        let frame = RelayMessageFrame::handshake(stream_id.to_string(), request);
+        harness_websocket
+            .send(Message::Binary(encode_relay_message_frame(&frame).into()))
+            .await?;
+    }
+
+    timeout(Duration::from_secs(1), async {
+        while calls.load(Ordering::SeqCst) != 2 {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await?;
+
+    harness_websocket.close(None).await?;
+    timeout(Duration::from_secs(1), environment_task).await??;
+    Ok(())
+}
+
+#[tokio::test]
+async fn oversized_harness_authorization_is_rejected_before_validation() -> Result<()> {
+    let listener = TcpListener::bind("127.0.0.1:0").await?;
+    let websocket_url = format!("ws://{}", listener.local_addr()?);
+    let harness_connection = tokio::spawn(connect_async(websocket_url));
+    let (socket, _peer_addr) = listener.accept().await?;
+    let environment_websocket = accept_async(socket).await?;
+    let (mut harness_websocket, _response) = harness_connection.await??;
+
+    let environment_identity = NoiseChannelIdentity::generate()?;
+    let harness_identity = NoiseChannelIdentity::generate()?;
+    let calls = Arc::new(AtomicUsize::new(0));
+    let environment_task = tokio::spawn(run_noise_multiplexed_environment(
+        environment_websocket,
+        ConnectionProcessor::new(ExecServerRuntimePaths::new(
+            std::env::current_exe()?,
+            /*codex_linux_sandbox_exe*/ None,
+        )?),
+        ENVIRONMENT_ID.to_string(),
+        EXECUTOR_REGISTRATION_ID.to_string(),
+        environment_identity.clone(),
+        BlockingValidator {
+            calls: Arc::clone(&calls),
+            release: Arc::new(Notify::new()),
+        },
+    ));
+
+    let stream_id = "stream-1";
+    let prologue = noise_channel_prologue(ENVIRONMENT_ID, EXECUTOR_REGISTRATION_ID, stream_id)?;
+    let oversized_authorization = vec![b'a'; MAX_HARNESS_KEY_AUTHORIZATION_BYTES + 1];
+    let (_handshake, request) = InitiatorHandshake::start(
+        &harness_identity,
+        &environment_identity.public_key(),
+        &prologue,
+        &oversized_authorization,
+    )?;
+    let frame = RelayMessageFrame::handshake(stream_id.to_string(), request);
+    harness_websocket
+        .send(Message::Binary(encode_relay_message_frame(&frame).into()))
+        .await?;
+
+    let Message::Binary(payload) = timeout(Duration::from_secs(1), harness_websocket.next())
+        .await?
+        .ok_or_else(|| anyhow::anyhow!("environment closed before sending reset"))??
+    else {
+        anyhow::bail!("expected binary reset frame");
+    };
+    let reset = decode_relay_message_frame(payload.as_ref())?;
+    assert_eq!(reset.validate()?, RelayFrameBodyKind::Reset);
+    assert_eq!(calls.load(Ordering::SeqCst), 0);
+
+    harness_websocket.close(None).await?;
+    timeout(Duration::from_secs(1), environment_task).await??;
+    Ok(())
+}

codex-rs/exec-server/src/remote/noise.rs

@@ -269,3 +269,7 @@ fn validate_executor_registration_id(
     }
     Ok(())
 }
+
+#[cfg(test)]
+#[path = "noise_tests.rs"]
+mod tests;

codex-rs/exec-server/src/remote/noise_tests.rs

@@ -0,0 +1,170 @@
+use std::sync::Arc;
+
+use codex_api::AuthProvider;
+use codex_api::SharedAuthProvider;
+use http::HeaderMap;
+use http::HeaderValue;
+use wiremock::Mock;
+use wiremock::MockServer;
+use wiremock::ResponseTemplate;
+use wiremock::matchers::body_partial_json;
+use wiremock::matchers::header;
+use wiremock::matchers::method;
+use wiremock::matchers::path;
+
+use super::*;
+
+const HARNESS_KEY_AUTHORIZATION: &str = "authorization-that-must-not-leak";
+
+#[derive(Debug)]
+struct StaticRegistryAuthProvider;
+
+impl AuthProvider for StaticRegistryAuthProvider {
+    fn add_auth_headers(&self, headers: &mut HeaderMap) {
+        let _ = headers.insert(
+            http::header::AUTHORIZATION,
+            HeaderValue::from_static("Bearer registry-token"),
+        );
+    }
+}
+
+fn static_registry_auth_provider() -> SharedAuthProvider {
+    Arc::new(StaticRegistryAuthProvider)
+}
+
+#[tokio::test]
+async fn register_noise_environment_posts_security_profile_and_public_key() {
+    let server = MockServer::start().await;
+    let executor_public_key = NoiseChannelIdentity::generate()
+        .expect("identity")
+        .public_key();
+    Mock::given(method("POST"))
+        .and(path("/cloud/environment/environment-requested/register"))
+        .and(header("authorization", "Bearer registry-token"))
+        .and(body_partial_json(serde_json::json!({
+            "security_profile": NOISE_RELAY_SECURITY_PROFILE,
+            "executor_public_key": executor_public_key.clone(),
+        })))
+        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+            "environment_id": "environment-requested",
+            "url": "wss://rendezvous.test/noise",
+            "security_profile": NOISE_RELAY_SECURITY_PROFILE,
+            "executor_registration_id": "registration-1",
+        })))
+        .mount(&server)
+        .await;
+    let client = EnvironmentRegistryClient::new(server.uri(), static_registry_auth_provider())
+        .expect("client");
+
+    let response = client
+        .register_noise_environment("environment-requested", &executor_public_key)
+        .await
+        .expect("register Noise environment");
+
+    assert_eq!(response.environment_id, "environment-requested");
+    assert_eq!(response.url, "wss://rendezvous.test/noise");
+    assert_eq!(response.security_profile, NOISE_RELAY_SECURITY_PROFILE);
+    assert_eq!(response.executor_registration_id, "registration-1");
+}
+
+#[tokio::test]
+async fn validate_harness_key_requires_explicit_valid_response() {
+    let server = MockServer::start().await;
+    let harness_public_key = NoiseChannelIdentity::generate()
+        .expect("identity")
+        .public_key();
+    Mock::given(method("POST"))
+        .and(path("/cloud/environment/environment-requested/validate"))
+        .and(header("authorization", "Bearer registry-token"))
+        .and(body_partial_json(serde_json::json!({
+            "executor_registration_id": "registration-1",
+            "harness_public_key": harness_public_key.clone(),
+            "harness_key_authorization": HARNESS_KEY_AUTHORIZATION,
+        })))
+        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+            "valid": false,
+        })))
+        .mount(&server)
+        .await;
+    let client = EnvironmentRegistryClient::new(server.uri(), static_registry_auth_provider())
+        .expect("client");
+
+    let error = client
+        .validate_harness_key(
+            "environment-requested",
+            "registration-1",
+            &harness_public_key,
+            HARNESS_KEY_AUTHORIZATION,
+        )
+        .await
+        .expect_err("a false validation response must fail closed");
+
+    assert!(matches!(
+        error,
+        ExecServerError::Protocol(message)
+            if message == "environment registry rejected Noise relay harness key"
+    ));
+}
+
+#[tokio::test]
+async fn validate_harness_key_does_not_expose_error_body() {
+    let server = MockServer::start().await;
+    let harness_public_key = NoiseChannelIdentity::generate()
+        .expect("identity")
+        .public_key();
+    Mock::given(method("POST"))
+        .and(path("/cloud/environment/environment-requested/validate"))
+        .respond_with(ResponseTemplate::new(500).set_body_string(HARNESS_KEY_AUTHORIZATION))
+        .mount(&server)
+        .await;
+    let client = EnvironmentRegistryClient::new(server.uri(), static_registry_auth_provider())
+        .expect("client");
+
+    let error = client
+        .validate_harness_key(
+            "environment-requested",
+            "registration-1",
+            &harness_public_key,
+            HARNESS_KEY_AUTHORIZATION,
+        )
+        .await
+        .expect_err("validation HTTP error should fail closed");
+
+    let display = error.to_string();
+    assert!(!display.contains(HARNESS_KEY_AUTHORIZATION));
+    assert!(matches!(
+        error,
+        ExecServerError::EnvironmentRegistryHttp { message, .. }
+            if message == "environment registry harness key validation failed"
+    ));
+}
+
+#[test]
+fn noise_environment_id_validation_rejects_path_injection() {
+    validate_environment_id(
+        "ccarenv_b64_Y2Fhcy1zdGFnaW5nLWV4ZWN1dG9yLWVudmlyb25tZW50LTE",
+    )
+    .expect("valid cloud environment id");
+
+    let error = validate_environment_id("ccarenv_b64_valid/../../status")
+        .expect_err("path delimiter must not reach an authenticated registry request");
+
+    assert!(matches!(
+        error,
+        ExecServerError::EnvironmentRegistryConfig(message) if message.contains("ASCII letters")
+    ));
+}
+
+#[test]
+fn executor_registration_id_validation_rejects_ambiguous_values() {
+    for invalid in ["", " registration-1", "registration-1 "] {
+        assert!(validate_executor_registration_id(invalid).is_err());
+    }
+    assert!(
+        validate_executor_registration_id(&"x".repeat(MAX_EXECUTOR_REGISTRATION_ID_LEN)).is_ok()
+    );
+    assert!(
+        validate_executor_registration_id(&"x".repeat(MAX_EXECUTOR_REGISTRATION_ID_LEN + 1))
+            .is_err()
+    );
+}