openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-05 20:10:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: name Noise relay protocol explicitly

Browse Source 
Co-authored-by: Codex noreply@openai.com
This commit is contained in:
viyatb-oai
2026-06-03 07:50:00 -07:00
co-authored by Codex noreply@openai.com
parent 03d54d008b
commit 2af2276e59
22 changed files with 373 additions and 373 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/cli/src/main.rs
+1 -1
View File
@@ -1620,7 +1620,7 @@ async fn run_exec_server_command(
remote_config.name = name;
}
if cmd.enable_noise {
remote_config.relay_security = codex_exec_server::RemoteRelaySecurity::Noise;
remote_config.relay_protocol = codex_exec_server::RemoteRelayProtocol::Noise;
}
codex_exec_server::run_remote_environment(remote_config, runtime_paths).await?;
Ok(())
codex-rs/core-api/src/lib.rs
+6 -6
View File
@@ -47,12 +47,12 @@ pub use codex_core::thread_store_from_config;
pub use codex_exec_server::EnvironmentManager;
pub use codex_exec_server::ExecServerError;
pub use codex_exec_server::ExecServerRuntimePaths;
pub use codex_exec_server::SecureChannelIdentity;
pub use codex_exec_server::SecureChannelPublicKey;
pub use codex_exec_server::SecureRendezvousConnectArgs;
pub use codex_exec_server::SecureRendezvousConnectBundle;
pub use codex_exec_server::SecureRendezvousConnectProvider;
pub use codex_exec_server::SharedSecureRendezvousConnectProvider;
pub use codex_exec_server::NoiseChannelIdentity;
pub use codex_exec_server::NoiseChannelPublicKey;
pub use codex_exec_server::NoiseRendezvousConnectArgs;
pub use codex_exec_server::NoiseRendezvousConnectBundle;
pub use codex_exec_server::NoiseRendezvousConnectProvider;
pub use codex_exec_server::SharedNoiseRendezvousConnectProvider;
pub use codex_extension_api::empty_extension_registry;
pub use codex_features::Feature;
pub use codex_features::Features;
codex-rs/exec-server/README.md
+1 -1
View File
@@ -29,7 +29,7 @@ then reconnects to the service-provided rendezvous websocket as the environment.
It uses the legacy relay contract by default for compatibility with existing
registries and harnesses. `--enable-noise` explicitly opts into Noise-encrypted
relay registration and requires both the registry and harness to support the
secure relay contract. Container entrypoints can make the same explicit choice
Noise relay contract. Container entrypoints can make the same explicit choice
by setting `CODEX_EXEC_SERVER_ENABLE_NOISE=true`.
It uses the standard Codex ChatGPT sign-in state; run `codex login` first when
remote registration needs authentication. Containerized callers that receive an
codex-rs/exec-server/src/client.rs
+9 -9
View File
@@ -23,9 +23,9 @@ use crate::ProcessId;
use crate::client_api::ExecServerClientConnectOptions;
use crate::client_api::ExecServerTransportParams;
use crate::client_api::HttpClient;
use crate::client_api::NoiseRendezvousConnectArgs;
use crate::client_api::NoiseRendezvousConnectBundle;
use crate::client_api::RemoteExecServerConnectArgs;
use crate::client_api::SecureRendezvousConnectArgs;
use crate::client_api::SecureRendezvousConnectBundle;
use crate::client_api::StdioExecServerConnectArgs;
use crate::connection::JsonRpcConnection;
use crate::process::ExecProcessEvent;
@@ -117,8 +117,8 @@ impl From<RemoteExecServerConnectArgs> for ExecServerClientConnectOptions {
}
}
impl From<SecureRendezvousConnectArgs> for ExecServerClientConnectOptions {
fn from(value: SecureRendezvousConnectArgs) -> Self {
impl From<NoiseRendezvousConnectArgs> for ExecServerClientConnectOptions {
fn from(value: NoiseRendezvousConnectArgs) -> Self {
Self {
client_name: value.client_name,
initialize_timeout: value.initialize_timeout,
@@ -149,10 +149,10 @@ impl RemoteExecServerConnectArgs {
}
}
impl SecureRendezvousConnectArgs {
impl NoiseRendezvousConnectArgs {
pub fn new(
bundle: SecureRendezvousConnectBundle,
harness_identity: crate::SecureChannelIdentity,
bundle: NoiseRendezvousConnectBundle,
harness_identity: crate::NoiseChannelIdentity,
client_name: String,
) -> Self {
Self {
@@ -260,7 +260,7 @@ impl LazyRemoteExecServerClient {
if matches!(
&self.transport_params,
ExecServerTransportParams::WebSocketUrl { .. }
| ExecServerTransportParams::SecureRendezvous { .. }
| ExecServerTransportParams::NoiseRendezvous { .. }
) =>
{
ExecServerClient::connect_for_transport(self.transport_params.clone()).await?
@@ -348,7 +348,7 @@ pub enum ExecServerError {
#[error("environment registry request failed: {0}")]
EnvironmentRegistryRequest(#[from] reqwest::Error),
#[error(transparent)]
SecureChannel(#[from] crate::SecureChannelError),
NoiseChannel(#[from] crate::NoiseChannelError),
}
impl ExecServerClient {
codex-rs/exec-server/src/client_api.rs
+21 -21
View File
@@ -9,8 +9,8 @@ use crate::ExecServerError;
use crate::HttpRequestParams;
use crate::HttpRequestResponse;
use crate::HttpResponseBodyStream;
use crate::SecureChannelIdentity;
use crate::SecureChannelPublicKey;
use crate::NoiseChannelIdentity;
use crate::NoiseChannelPublicKey;
pub(crate) const DEFAULT_REMOTE_EXEC_SERVER_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
pub(crate) const DEFAULT_REMOTE_EXEC_SERVER_INITIALIZE_TIMEOUT: Duration = Duration::from_secs(10);
@@ -33,19 +33,19 @@ pub struct RemoteExecServerConnectArgs {
pub resume_session_id: Option<String>,
}
/// Registry-authorized material for one secure rendezvous connection attempt.
/// Registry-authorized material for one Noise rendezvous connection attempt.
#[derive(Clone)]
pub struct SecureRendezvousConnectBundle {
pub struct NoiseRendezvousConnectBundle {
pub websocket_url: String,
pub environment_id: String,
pub executor_registration_id: String,
pub executor_public_key: SecureChannelPublicKey,
pub executor_public_key: NoiseChannelPublicKey,
pub harness_key_authorization: String,
}
impl std::fmt::Debug for SecureRendezvousConnectBundle {
impl std::fmt::Debug for NoiseRendezvousConnectBundle {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SecureRendezvousConnectBundle")
f.debug_struct("NoiseRendezvousConnectBundle")
.field(
"websocket_url",
&redacted_websocket_url(&self.websocket_url),
@@ -58,20 +58,20 @@ impl std::fmt::Debug for SecureRendezvousConnectBundle {
}
}
/// Connection arguments for an authenticated secure rendezvous exec-server.
/// Connection arguments for an authenticated Noise rendezvous exec-server.
#[derive(Clone)]
pub struct SecureRendezvousConnectArgs {
pub bundle: SecureRendezvousConnectBundle,
pub harness_identity: SecureChannelIdentity,
pub struct NoiseRendezvousConnectArgs {
pub bundle: NoiseRendezvousConnectBundle,
pub harness_identity: NoiseChannelIdentity,
pub client_name: String,
pub connect_timeout: Duration,
pub initialize_timeout: Duration,
pub resume_session_id: Option<String>,
}
impl std::fmt::Debug for SecureRendezvousConnectArgs {
impl std::fmt::Debug for NoiseRendezvousConnectArgs {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SecureRendezvousConnectArgs")
f.debug_struct("NoiseRendezvousConnectArgs")
.field("bundle", &self.bundle)
.field("harness_identity", &"<redacted>")
.field("client_name", &self.client_name)
@@ -82,19 +82,19 @@ impl std::fmt::Debug for SecureRendezvousConnectArgs {
}
}
/// Supplies fresh registry-authorized material for secure rendezvous connections.
/// Supplies fresh registry-authorized material for Noise rendezvous connections.
///
/// Implementations must preserve one endpoint-local harness identity while
/// refreshing short-lived registry material for every physical connection attempt.
pub trait SecureRendezvousConnectProvider: Send + Sync {
pub trait NoiseRendezvousConnectProvider: Send + Sync {
/// Environment ID this provider is authorized to connect to.
fn environment_id(&self) -> &str;
/// Returns a fresh atomic bundle for one physical connection attempt.
fn connect_args(&self) -> BoxFuture<'_, Result<SecureRendezvousConnectArgs, ExecServerError>>;
fn connect_args(&self) -> BoxFuture<'_, Result<NoiseRendezvousConnectArgs, ExecServerError>>;
}
pub type SharedSecureRendezvousConnectProvider = Arc<dyn SecureRendezvousConnectProvider>;
pub type SharedNoiseRendezvousConnectProvider = Arc<dyn NoiseRendezvousConnectProvider>;
/// Stdio connection arguments for a command-backed exec-server.
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -122,8 +122,8 @@ pub(crate) enum ExecServerTransportParams {
connect_timeout: Duration,
initialize_timeout: Duration,
},
SecureRendezvous {
provider: SharedSecureRendezvousConnectProvider,
NoiseRendezvous {
provider: SharedNoiseRendezvousConnectProvider,
},
#[allow(dead_code)]
StdioCommand {
@@ -145,8 +145,8 @@ impl std::fmt::Debug for ExecServerTransportParams {
.field("connect_timeout", connect_timeout)
.field("initialize_timeout", initialize_timeout)
.finish(),
Self::SecureRendezvous { provider } => f
.debug_struct("SecureRendezvous")
Self::NoiseRendezvous { provider } => f
.debug_struct("NoiseRendezvous")
.field("environment_id", &provider.environment_id())
.finish(),
Self::StdioCommand {
codex-rs/exec-server/src/client_transport.rs
+11 -12
View File
@@ -12,15 +12,15 @@ use codex_utils_rustls_provider::ensure_rustls_crypto_provider;
use crate::ExecServerClient;
use crate::ExecServerError;
use crate::client_api::NoiseRendezvousConnectArgs;
use crate::client_api::RemoteExecServerConnectArgs;
use crate::client_api::SecureRendezvousConnectArgs;
use crate::client_api::StdioExecServerCommand;
use crate::client_api::StdioExecServerConnectArgs;
use crate::client_api::redacted_websocket_url;
use crate::connection::JsonRpcConnection;
use crate::noise_relay::noise_harness_connection_from_websocket;
use crate::noise_relay::noise_relay_websocket_config;
use crate::relay::harness_connection_from_websocket;
use crate::secure_relay::secure_harness_connection_from_websocket;
use crate::secure_relay::secure_relay_websocket_config;
const ENVIRONMENT_CLIENT_NAME: &str = "codex-environment";
@@ -43,15 +43,14 @@ impl ExecServerClient {
})
.await
}
crate::client_api::ExecServerTransportParams::SecureRendezvous { provider } => {
crate::client_api::ExecServerTransportParams::NoiseRendezvous { provider } => {
let args = provider.connect_args().await?;
if args.bundle.environment_id != provider.environment_id() {
return Err(ExecServerError::Protocol(
"secure rendezvous provider returned a different environment id"
.to_string(),
"Noise rendezvous provider returned a different environment id".to_string(),
));
}
Self::connect_secure_rendezvous(args).await
Self::connect_noise_rendezvous(args).await
}
crate::client_api::ExecServerTransportParams::StdioCommand {
command,
@@ -94,8 +93,8 @@ impl ExecServerClient {
Self::connect(connection, args.into()).await
}
pub async fn connect_secure_rendezvous(
args: SecureRendezvousConnectArgs,
pub async fn connect_noise_rendezvous(
args: NoiseRendezvousConnectArgs,
) -> Result<Self, ExecServerError> {
ensure_rustls_crypto_provider();
let websocket_url = args.bundle.websocket_url.clone();
@@ -105,7 +104,7 @@ impl ExecServerClient {
connect_timeout,
connect_async_with_config(
websocket_url.as_str(),
Some(secure_relay_websocket_config()),
Some(noise_relay_websocket_config()),
/*disable_nagle*/ false,
),
)
@@ -119,8 +118,8 @@ impl ExecServerClient {
source,
})?;
let connection_label = format!("secure exec-server rendezvous websocket {diagnostic_url}");
let connection = secure_harness_connection_from_websocket(
let connection_label = format!("Noise exec-server rendezvous websocket {diagnostic_url}");
let connection = noise_harness_connection_from_websocket(
stream,
connection_label,
args.bundle.environment_id.clone(),
codex-rs/exec-server/src/environment.rs
+6 -6
View File
@@ -6,7 +6,7 @@ use crate::ExecServerError;
use crate::ExecServerRuntimePaths;
use crate::ExecutorFileSystem;
use crate::HttpClient;
use crate::SharedSecureRendezvousConnectProvider;
use crate::SharedNoiseRendezvousConnectProvider;
use crate::client::LazyRemoteExecServerClient;
use crate::client::http_client::ReqwestHttpClient;
use crate::client_api::ExecServerTransportParams;
@@ -280,10 +280,10 @@ impl EnvironmentManager {
/// Adds or replaces a named remote environment that connects through an
/// authenticated, end-to-end encrypted rendezvous stream.
pub fn upsert_secure_environment(
pub fn upsert_noise_environment(
&self,
environment_id: String,
provider: SharedSecureRendezvousConnectProvider,
provider: SharedNoiseRendezvousConnectProvider,
) -> Result<(), ExecServerError> {
if environment_id.is_empty() {
return Err(ExecServerError::Protocol(
@@ -292,11 +292,11 @@ impl EnvironmentManager {
}
if environment_id != provider.environment_id() {
return Err(ExecServerError::Protocol(
"secure environment id does not match connection provider".to_string(),
"Noise environment id does not match connection provider".to_string(),
));
}
let environment = Environment::remote_with_transport(
ExecServerTransportParams::SecureRendezvous { provider },
ExecServerTransportParams::NoiseRendezvous { provider },
self.local_runtime_paths.clone(),
);
self.environments
@@ -411,7 +411,7 @@ impl Environment {
websocket_url: exec_server_url,
..
} => Some(exec_server_url.clone()),
ExecServerTransportParams::SecureRendezvous { .. } => None,
ExecServerTransportParams::NoiseRendezvous { .. } => None,
ExecServerTransportParams::StdioCommand { .. } => None,
};
let client = LazyRemoteExecServerClient::new(remote_transport.clone());
codex-rs/exec-server/src/lib.rs
+10 -10
View File
@@ -12,6 +12,8 @@ mod fs_helper_main;
mod fs_sandbox;
mod local_file_system;
mod local_process;
mod noise_channel;
mod noise_relay;
mod process;
mod process_id;
mod protocol;
@@ -23,8 +25,6 @@ mod remote_process;
mod rpc;
mod runtime_paths;
mod sandboxed_file_system;
mod secure_channel;
mod secure_relay;
mod server;
pub use client::ExecServerClient;
@@ -33,11 +33,11 @@ pub use client::http_client::HttpResponseBodyStream;
pub use client::http_client::ReqwestHttpClient;
pub use client_api::ExecServerClientConnectOptions;
pub use client_api::HttpClient;
pub use client_api::NoiseRendezvousConnectArgs;
pub use client_api::NoiseRendezvousConnectBundle;
pub use client_api::NoiseRendezvousConnectProvider;
pub use client_api::RemoteExecServerConnectArgs;
pub use client_api::SecureRendezvousConnectArgs;
pub use client_api::SecureRendezvousConnectBundle;
pub use client_api::SecureRendezvousConnectProvider;
pub use client_api::SharedSecureRendezvousConnectProvider;
pub use client_api::SharedNoiseRendezvousConnectProvider;
pub use codex_file_system::CopyOptions;
pub use codex_file_system::CreateDirectoryOptions;
pub use codex_file_system::ExecutorFileSystem;
@@ -58,6 +58,9 @@ pub use fs_helper::CODEX_FS_HELPER_ARG1;
pub use fs_helper_main::main as run_fs_helper_main;
pub use local_file_system::LOCAL_FS;
pub use local_file_system::LocalFileSystem;
pub use noise_channel::NoiseChannelError;
pub use noise_channel::NoiseChannelIdentity;
pub use noise_channel::NoiseChannelPublicKey;
pub use process::ExecBackend;
pub use process::ExecProcess;
pub use process::ExecProcessEvent;
@@ -107,12 +110,9 @@ pub use protocol::WriteParams;
pub use protocol::WriteResponse;
pub use protocol::WriteStatus;
pub use remote::RemoteEnvironmentConfig;
pub use remote::RemoteRelaySecurity;
pub use remote::RemoteRelayProtocol;
pub use remote::run_remote_environment;
pub use runtime_paths::ExecServerRuntimePaths;
pub use secure_channel::SecureChannelError;
pub use secure_channel::SecureChannelIdentity;
pub use secure_channel::SecureChannelPublicKey;
pub use server::DEFAULT_LISTEN_URL;
pub use server::ExecServerListenUrlParseError;
pub use server::run_main;
codex-rs/exec-server/src/secure_channel.rs → codex-rs/exec-server/src/noise_channel.rs
+67 -67
View File
@@ -29,7 +29,7 @@ use serde::Serialize;
use crate::aws_lc_ml_kem::AwsLcMlKem768;
use crate::aws_lc_ml_kem::PUBLIC_KEY_LEN as MLKEM768_PUBLIC_KEY_LEN;
pub const SECURE_CHANNEL_SUITE: &str = "Noise_hybridIK_X25519+MLKEM768_AESGCM_SHA256";
pub const NOISE_CHANNEL_SUITE: &str = "Noise_hybridIK_X25519+MLKEM768_AESGCM_SHA256";
const X25519_PUBLIC_KEY_LEN: usize = 32;
const MAX_TRANSPORT_RECORDS_PER_DIRECTION: u64 = u32::MAX as u64;
@@ -40,22 +40,22 @@ type Transport = TransportState<AesGcm, Sha256>;
type DhKeyPair = KeyPair<<X25519 as Dh>::PubKey, <X25519 as Dh>::PrivateKey>;
type KemKeyPair = KeyPair<<AwsLcMlKem768 as Kem>::PubKey, <AwsLcMlKem768 as Kem>::SecretKey>;
/// Public key material for the exec-server secure relay Noise suite.
/// Public key material for the exec-server Noise-over-relay suite.
///
/// The suite field is part of the serialized contract. A key from a different
/// suite must not be interpreted as compatible merely because one component has
/// a familiar byte length.
#[derive(Clone, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SecureChannelPublicKey {
pub struct NoiseChannelPublicKey {
suite: String,
x25519_public_key: String,
mlkem768_public_key: String,
}
impl std::fmt::Debug for SecureChannelPublicKey {
impl std::fmt::Debug for NoiseChannelPublicKey {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SecureChannelPublicKey")
f.debug_struct("NoiseChannelPublicKey")
.field("suite", &self.suite)
.field("x25519_public_key", &"<redacted>")
.field("mlkem768_public_key", &"<redacted>")
@@ -63,16 +63,16 @@ impl std::fmt::Debug for SecureChannelPublicKey {
}
}
impl SecureChannelPublicKey {
impl NoiseChannelPublicKey {
/// Validate suite selection, encoding, and exact key lengths.
pub fn validate(&self) -> Result<(), SecureChannelError> {
pub fn validate(&self) -> Result<(), NoiseChannelError> {
let _ = self.decode()?;
Ok(())
}
fn from_keypairs(dh: &DhKeyPair, kem: &KemKeyPair) -> Self {
Self {
suite: SECURE_CHANNEL_SUITE.to_string(),
suite: NOISE_CHANNEL_SUITE.to_string(),
x25519_public_key: STANDARD.encode(dh.public),
mlkem768_public_key: STANDARD.encode(kem.public.as_slice()),
}
@@ -80,7 +80,7 @@ impl SecureChannelPublicKey {
fn from_raw(dh: &<X25519 as Dh>::PubKey, kem: &<AwsLcMlKem768 as Kem>::PubKey) -> Self {
Self {
suite: SECURE_CHANNEL_SUITE.to_string(),
suite: NOISE_CHANNEL_SUITE.to_string(),
x25519_public_key: STANDARD.encode(dh),
mlkem768_public_key: STANDARD.encode(kem.as_slice()),
}
@@ -88,23 +88,23 @@ impl SecureChannelPublicKey {
fn decode(
&self,
) -> Result<(<X25519 as Dh>::PubKey, <AwsLcMlKem768 as Kem>::PubKey), SecureChannelError> {
if self.suite != SECURE_CHANNEL_SUITE {
return Err(SecureChannelError::InvalidPublicKey(
"unsupported secure channel suite",
) -> Result<(<X25519 as Dh>::PubKey, <AwsLcMlKem768 as Kem>::PubKey), NoiseChannelError> {
if self.suite != NOISE_CHANNEL_SUITE {
return Err(NoiseChannelError::InvalidPublicKey(
"unsupported Noise channel suite",
));
}
let dh = STANDARD
.decode(&self.x25519_public_key)
.map_err(|_| SecureChannelError::InvalidPublicKey("invalid X25519 public key"))?;
let dh: [u8; X25519_PUBLIC_KEY_LEN] = dh.try_into().map_err(|_| {
SecureChannelError::InvalidPublicKey("invalid X25519 public key length")
})?;
.map_err(|_| NoiseChannelError::InvalidPublicKey("invalid X25519 public key"))?;
let dh: [u8; X25519_PUBLIC_KEY_LEN] = dh
.try_into()
.map_err(|_| NoiseChannelError::InvalidPublicKey("invalid X25519 public key length"))?;
let kem = STANDARD
.decode(&self.mlkem768_public_key)
.map_err(|_| SecureChannelError::InvalidPublicKey("invalid ML-KEM-768 public key"))?;
.map_err(|_| NoiseChannelError::InvalidPublicKey("invalid ML-KEM-768 public key"))?;
if kem.len() != MLKEM768_PUBLIC_KEY_LEN {
return Err(SecureChannelError::InvalidPublicKey(
return Err(NoiseChannelError::InvalidPublicKey(
"invalid ML-KEM-768 public key length",
));
}
@@ -116,37 +116,37 @@ impl SecureChannelPublicKey {
}
}
/// Endpoint-local static identity for the exec-server secure relay Noise suite.
/// Endpoint-local static identity for the exec-server Noise-over-relay suite.
///
/// Private components never cross the process boundary. Cloning is used only to
/// construct Clatter handshake state for reconnects within the same process.
#[derive(Clone)]
pub struct SecureChannelIdentity {
pub struct NoiseChannelIdentity {
dh: DhKeyPair,
kem: KemKeyPair,
}
impl std::fmt::Debug for SecureChannelIdentity {
impl std::fmt::Debug for NoiseChannelIdentity {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SecureChannelIdentity")
f.debug_struct("NoiseChannelIdentity")
.field("public_key", &self.public_key())
.finish_non_exhaustive()
}
}
impl SecureChannelIdentity {
impl NoiseChannelIdentity {
/// Generate independent classical and post-quantum static keypairs.
pub fn generate() -> Result<Self, SecureChannelError> {
pub fn generate() -> Result<Self, NoiseChannelError> {
let dh = X25519::genkey()
.map_err(|error| SecureChannelError::KeyGeneration(error.to_string()))?;
.map_err(|error| NoiseChannelError::KeyGeneration(error.to_string()))?;
let kem = AwsLcMlKem768::genkey()
.map_err(|error| SecureChannelError::KeyGeneration(error.to_string()))?;
.map_err(|error| NoiseChannelError::KeyGeneration(error.to_string()))?;
Ok(Self { dh, kem })
}
/// Return the distributable public half of this endpoint identity.
pub fn public_key(&self) -> SecureChannelPublicKey {
SecureChannelPublicKey::from_keypairs(&self.dh, &self.kem)
pub fn public_key(&self) -> NoiseChannelPublicKey {
NoiseChannelPublicKey::from_keypairs(&self.dh, &self.kem)
}
}
@@ -160,11 +160,11 @@ impl InitiatorHandshake {
/// `payload` is encrypted by the first IK message. The relay uses it for
/// the short-lived registry authorization naming this harness public key.
pub(crate) fn start(
identity: &SecureChannelIdentity,
responder_public_key: &SecureChannelPublicKey,
identity: &NoiseChannelIdentity,
responder_public_key: &NoiseChannelPublicKey,
prologue: &[u8],
payload: &[u8],
) -> Result<(Self, Vec<u8>), SecureChannelError> {
) -> Result<(Self, Vec<u8>), NoiseChannelError> {
let (responder_dh, responder_kem) = responder_public_key.decode()?;
// IK authenticates both static identities. Supplying both responder
@@ -186,16 +186,16 @@ impl InitiatorHandshake {
///
/// The responder message carries no application payload in v1. Rejecting
/// one keeps future protocol additions from becoming an implicit channel.
pub(crate) fn finish(mut self, response: &[u8]) -> Result<SecureTransport, SecureChannelError> {
pub(crate) fn finish(mut self, response: &[u8]) -> Result<NoiseTransport, NoiseChannelError> {
ensure_noise_frame_len(response.len(), "handshake response is too large")?;
let mut payload = [0u8; MAX_MESSAGE_LEN];
let payload_len = self.handshake.read_message(response, &mut payload)?;
if payload_len != 0 {
return Err(SecureChannelError::InvalidMessage(
return Err(NoiseChannelError::InvalidMessage(
"handshake response payload must be empty",
));
}
Ok(SecureTransport {
Ok(NoiseTransport {
transport: self.handshake.finalize()?,
})
}
@@ -203,7 +203,7 @@ impl InitiatorHandshake {
pub(crate) struct PendingResponderHandshake {
handshake: Handshake,
initiator_public_key: SecureChannelPublicKey,
initiator_public_key: NoiseChannelPublicKey,
payload: Vec<u8>,
}
@@ -213,10 +213,10 @@ impl PendingResponderHandshake {
/// This split is intentional: callers must authorize `initiator_public_key`
/// with the registry before calling [`Self::complete`].
pub(crate) fn read_request(
identity: &SecureChannelIdentity,
identity: &NoiseChannelIdentity,
prologue: &[u8],
request: &[u8],
) -> Result<Self, SecureChannelError> {
) -> Result<Self, NoiseChannelError> {
ensure_noise_frame_len(request.len(), "handshake request is too large")?;
let params = HybridHandshakeParams::new(noise_hybrid_ik(), false)
.with_prologue(prologue)
@@ -229,17 +229,17 @@ impl PendingResponderHandshake {
// message authenticates and decrypts successfully.
let remote = handshake
.get_remote_static()
.ok_or(SecureChannelError::InvalidMessage(
.ok_or(NoiseChannelError::InvalidMessage(
"handshake request is missing initiator static key",
))?;
Ok(Self {
handshake,
initiator_public_key: SecureChannelPublicKey::from_raw(remote.dh(), remote.kem()),
initiator_public_key: NoiseChannelPublicKey::from_raw(remote.dh(), remote.kem()),
payload: payload[..payload_len].to_vec(),
})
}
pub(crate) fn initiator_public_key(&self) -> &SecureChannelPublicKey {
pub(crate) fn initiator_public_key(&self) -> &NoiseChannelPublicKey {
&self.initiator_public_key
}
@@ -248,11 +248,11 @@ impl PendingResponderHandshake {
}
/// Finish the responder handshake after external harness authorization.
pub(crate) fn complete(mut self) -> Result<(SecureTransport, Vec<u8>), SecureChannelError> {
pub(crate) fn complete(mut self) -> Result<(NoiseTransport, Vec<u8>), NoiseChannelError> {
let mut response = [0u8; MAX_MESSAGE_LEN];
let response_len = self.handshake.write_message(&[], &mut response)?;
Ok((
SecureTransport {
NoiseTransport {
transport: self.handshake.finalize()?,
},
response[..response_len].to_vec(),
@@ -260,23 +260,23 @@ impl PendingResponderHandshake {
}
}
pub(crate) struct SecureTransport {
pub(crate) struct NoiseTransport {
transport: Transport,
}
impl SecureTransport {
impl NoiseTransport {
/// Encrypt exactly one ordered transport record.
///
/// The caller owns relay sequence assignment and must never encrypt the
/// same logical record twice under different transport nonces.
pub(crate) fn encrypt(&mut self, plaintext: &[u8]) -> Result<Vec<u8>, SecureChannelError> {
pub(crate) fn encrypt(&mut self, plaintext: &[u8]) -> Result<Vec<u8>, NoiseChannelError> {
if self.transport.sending_nonce() >= MAX_TRANSPORT_RECORDS_PER_DIRECTION {
return Err(SecureChannelError::InvalidState(
return Err(NoiseChannelError::InvalidState(
"transport record nonce exhausted",
));
}
let frame_len = plaintext.len().checked_add(AesGcm::tag_len()).ok_or(
SecureChannelError::InvalidMessage("transport plaintext is too large"),
NoiseChannelError::InvalidMessage("transport plaintext is too large"),
)?;
ensure_noise_frame_len(frame_len, "transport plaintext is too large")?;
Ok(self.transport.send_vec(plaintext)?)
@@ -286,14 +286,14 @@ impl SecureTransport {
///
/// Clatter advances the receiving nonce during this call, so callers must
/// reorder and deduplicate relay frames before invoking it.
pub(crate) fn decrypt(&mut self, ciphertext: &[u8]) -> Result<Vec<u8>, SecureChannelError> {
pub(crate) fn decrypt(&mut self, ciphertext: &[u8]) -> Result<Vec<u8>, NoiseChannelError> {
if self.transport.receiving_nonce() >= MAX_TRANSPORT_RECORDS_PER_DIRECTION {
return Err(SecureChannelError::InvalidState(
return Err(NoiseChannelError::InvalidState(
"transport record nonce exhausted",
));
}
if ciphertext.len() < AesGcm::tag_len() {
return Err(SecureChannelError::InvalidMessage(
return Err(NoiseChannelError::InvalidMessage(
"transport ciphertext is too short",
));
}
@@ -307,11 +307,11 @@ impl SecureTransport {
/// A handshake captured from one environment, exec-server registration, or
/// relay stream cannot be replayed into another because every participant must
/// construct the same prologue before the first Noise message is processed.
pub(crate) fn secure_channel_prologue(
pub(crate) fn noise_channel_prologue(
environment_id: &str,
executor_registration_id: &str,
stream_id: &str,
) -> Result<Vec<u8>, SecureChannelError> {
) -> Result<Vec<u8>, NoiseChannelError> {
let mut prologue = Vec::new();
append_prologue_part(&mut prologue, PROLOGUE_DOMAIN)?;
append_prologue_part(&mut prologue, environment_id.as_bytes())?;
@@ -320,11 +320,11 @@ pub(crate) fn secure_channel_prologue(
Ok(prologue)
}
fn append_prologue_part(prologue: &mut Vec<u8>, part: &[u8]) -> Result<(), SecureChannelError> {
fn append_prologue_part(prologue: &mut Vec<u8>, part: &[u8]) -> Result<(), NoiseChannelError> {
// Length prefixes make component boundaries unambiguous. Raw concatenation
// would allow different identifier tuples to produce the same prologue.
let len = u32::try_from(part.len()).map_err(|_| {
SecureChannelError::InvalidMessage("secure channel prologue part is too large")
NoiseChannelError::InvalidMessage("Noise channel prologue part is too large")
})?;
prologue.extend_from_slice(&len.to_be_bytes());
prologue.extend_from_slice(part);
@@ -334,41 +334,41 @@ fn append_prologue_part(prologue: &mut Vec<u8>, part: &[u8]) -> Result<(), Secur
fn ensure_noise_frame_len(
frame_len: usize,
message: &'static str,
) -> Result<(), SecureChannelError> {
) -> Result<(), NoiseChannelError> {
if frame_len > MAX_MESSAGE_LEN {
return Err(SecureChannelError::InvalidMessage(message));
return Err(NoiseChannelError::InvalidMessage(message));
}
Ok(())
}
#[derive(Debug, thiserror::Error)]
pub enum SecureChannelError {
#[error("secure channel key generation failed: {0}")]
pub enum NoiseChannelError {
#[error("Noise channel key generation failed: {0}")]
KeyGeneration(String),
#[error("invalid secure channel public key: {0}")]
#[error("invalid Noise channel public key: {0}")]
InvalidPublicKey(&'static str),
#[error("invalid secure channel state: {0}")]
#[error("invalid Noise channel state: {0}")]
InvalidState(&'static str),
#[error("invalid secure channel message: {0}")]
#[error("invalid Noise channel message: {0}")]
InvalidMessage(&'static str),
#[error("secure channel handshake failed: {0}")]
#[error("Noise channel handshake failed: {0}")]
Handshake(String),
#[error("secure channel transport failed: {0}")]
#[error("Noise channel transport failed: {0}")]
Transport(String),
}
impl From<clatter::error::HandshakeError> for SecureChannelError {
impl From<clatter::error::HandshakeError> for NoiseChannelError {
fn from(error: clatter::error::HandshakeError) -> Self {
Self::Handshake(error.to_string())
}
}
impl From<clatter::error::TransportError> for SecureChannelError {
impl From<clatter::error::TransportError> for NoiseChannelError {
fn from(error: clatter::error::TransportError) -> Self {
Self::Transport(error.to_string())
}
}
#[cfg(test)]
#[path = "secure_channel_tests.rs"]
#[path = "noise_channel_tests.rs"]
mod tests;
codex-rs/exec-server/src/secure_channel_tests.rs → codex-rs/exec-server/src/noise_channel_tests.rs
+32 -32
View File
@@ -2,19 +2,19 @@ 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::SECURE_CHANNEL_SUITE;
use super::SecureChannelError;
use super::SecureChannelIdentity;
use super::SecureChannelPublicKey;
use super::secure_channel_prologue;
use super::noise_channel_prologue;
#[test]
fn hybrid_ik_roundtrip_authenticates_both_endpoints() {
let initiator = SecureChannelIdentity::generate().expect("generate initiator identity");
let responder = SecureChannelIdentity::generate().expect("generate responder identity");
let initiator = NoiseChannelIdentity::generate().expect("generate initiator identity");
let responder = NoiseChannelIdentity::generate().expect("generate responder identity");
let prologue =
secure_channel_prologue("env-1", "registration-1", "stream-1").expect("build 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(
@@ -66,11 +66,11 @@ fn hybrid_ik_roundtrip_authenticates_both_endpoints() {
#[test]
fn initiator_rejects_wrong_responder_key() {
let initiator = SecureChannelIdentity::generate().expect("generate initiator identity");
let expected_responder = SecureChannelIdentity::generate().expect("generate expected identity");
let actual_responder = SecureChannelIdentity::generate().expect("generate actual identity");
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 =
secure_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
noise_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
let (_initiator_handshake, request) = InitiatorHandshake::start(
&initiator,
@@ -87,12 +87,12 @@ fn initiator_rejects_wrong_responder_key() {
#[test]
fn responder_rejects_mismatched_prologue() {
let initiator = SecureChannelIdentity::generate().expect("generate initiator identity");
let responder = SecureChannelIdentity::generate().expect("generate responder identity");
let initiator = NoiseChannelIdentity::generate().expect("generate initiator identity");
let responder = NoiseChannelIdentity::generate().expect("generate responder identity");
let initiator_prologue =
secure_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
noise_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
let responder_prologue =
secure_channel_prologue("env-1", "registration-1", "stream-2").expect("build prologue");
noise_channel_prologue("env-1", "registration-1", "stream-2").expect("build prologue");
let (_initiator_handshake, request) = InitiatorHandshake::start(
&initiator,
&responder.public_key(),
@@ -109,7 +109,7 @@ fn responder_rejects_mismatched_prologue() {
#[test]
fn prologue_encoding_is_stable_and_unambiguous() {
let prologue =
secure_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
noise_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
assert_eq!(
prologue,
@@ -123,10 +123,10 @@ fn prologue_encoding_is_stable_and_unambiguous() {
#[test]
fn transport_rejects_tampered_ciphertext() {
let initiator = SecureChannelIdentity::generate().expect("generate initiator identity");
let responder = SecureChannelIdentity::generate().expect("generate responder identity");
let initiator = NoiseChannelIdentity::generate().expect("generate initiator identity");
let responder = NoiseChannelIdentity::generate().expect("generate responder identity");
let prologue =
secure_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
noise_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
let (initiator_handshake, request) = InitiatorHandshake::start(
&initiator,
&responder.public_key(),
@@ -153,10 +153,10 @@ fn transport_rejects_tampered_ciphertext() {
#[test]
fn transport_rejects_exhausted_receiving_nonce_before_decryption() {
let initiator = SecureChannelIdentity::generate().expect("generate initiator identity");
let responder = SecureChannelIdentity::generate().expect("generate responder identity");
let initiator = NoiseChannelIdentity::generate().expect("generate initiator identity");
let responder = NoiseChannelIdentity::generate().expect("generate responder identity");
let prologue =
secure_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
noise_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
let (initiator_handshake, request) = InitiatorHandshake::start(
&initiator,
&responder.public_key(),
@@ -182,7 +182,7 @@ fn transport_rejects_exhausted_receiving_nonce_before_decryption() {
assert!(matches!(
responder_transport.decrypt(&ciphertext),
Err(SecureChannelError::InvalidState(
Err(NoiseChannelError::InvalidState(
"transport record nonce exhausted"
))
));
@@ -190,10 +190,10 @@ fn transport_rejects_exhausted_receiving_nonce_before_decryption() {
#[test]
fn transport_rejects_replayed_ciphertext() {
let initiator = SecureChannelIdentity::generate().expect("generate initiator identity");
let responder = SecureChannelIdentity::generate().expect("generate responder identity");
let initiator = NoiseChannelIdentity::generate().expect("generate initiator identity");
let responder = NoiseChannelIdentity::generate().expect("generate responder identity");
let prologue =
secure_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
noise_channel_prologue("env-1", "registration-1", "stream-1").expect("build prologue");
let (initiator_handshake, request) = InitiatorHandshake::start(
&initiator,
&responder.public_key(),
@@ -222,19 +222,19 @@ fn transport_rejects_replayed_ciphertext() {
);
assert!(matches!(
responder_transport.decrypt(&ciphertext),
Err(SecureChannelError::Transport(_))
Err(NoiseChannelError::Transport(_))
));
}
#[test]
fn public_key_validation_rejects_unknown_suite() {
let key = SecureChannelIdentity::generate()
let key = NoiseChannelIdentity::generate()
.expect("generate identity")
.public_key();
let json = serde_json::to_value(key).expect("serialize key");
let mut object = json.as_object().expect("key object").clone();
object.insert("suite".to_string(), serde_json::json!("unknown"));
let key: SecureChannelPublicKey =
let key: NoiseChannelPublicKey =
serde_json::from_value(serde_json::Value::Object(object)).expect("deserialize key");
assert!(key.validate().is_err());
@@ -242,11 +242,11 @@ fn public_key_validation_rejects_unknown_suite() {
#[test]
fn public_key_serializes_with_expected_suite() {
let key = SecureChannelIdentity::generate()
let key = NoiseChannelIdentity::generate()
.expect("generate identity")
.public_key();
let json = serde_json::to_value(key).expect("serialize key");
assert_eq!(json["suite"], SECURE_CHANNEL_SUITE);
assert_eq!(json["suite"], NOISE_CHANNEL_SUITE);
}
codex-rs/exec-server/src/secure_relay/environment.rs → codex-rs/exec-server/src/noise_relay/environment.rs
+53 -51
View File
@@ -21,25 +21,27 @@ use crate::connection::CHANNEL_CAPACITY;
use crate::connection::JsonRpcConnection;
use crate::connection::JsonRpcConnectionEvent;
use crate::connection::JsonRpcTransport;
use crate::noise_channel::NoiseChannelIdentity;
use crate::noise_channel::NoiseChannelPublicKey;
use crate::noise_channel::NoiseTransport;
use crate::noise_channel::PendingResponderHandshake;
use crate::noise_channel::noise_channel_prologue;
use crate::noise_relay::message_framing::JsonRpcMessageDecoder;
use crate::noise_relay::message_framing::NOISE_RECORD_PLAINTEXT_LEN;
use crate::noise_relay::message_framing::frame_jsonrpc_message;
use crate::noise_relay::ordered_ciphertext::OrderedCiphertextFrames;
use crate::noise_relay::take_next_sequence;
use crate::relay::RelayFrameBodyKind;
use crate::relay::decode_relay_message_frame;
use crate::relay::encode_relay_message_frame;
use crate::relay_proto::RelayData;
use crate::relay_proto::RelayMessageFrame;
use crate::secure_channel::PendingResponderHandshake;
use crate::secure_channel::SecureChannelIdentity;
use crate::secure_channel::SecureChannelPublicKey;
use crate::secure_channel::SecureTransport;
use crate::secure_channel::secure_channel_prologue;
use crate::secure_relay::message_framing::JsonRpcMessageDecoder;
use crate::secure_relay::message_framing::SECURE_RECORD_PLAINTEXT_LEN;
use crate::secure_relay::message_framing::frame_jsonrpc_message;
use crate::secure_relay::ordered_ciphertext::OrderedCiphertextFrames;
use crate::secure_relay::take_next_sequence;
use crate::server::ConnectionProcessor;
const SECURE_RELAY_RESET_REASON: &str = "secure_relay_protocol_error";
const MAX_ACTIVE_SECURE_RELAY_STREAMS: usize = 128;
// This value is already part of the relay wire contract. Keep it stable even
// though the source module now uses the more precise Noise terminology.
const NOISE_RELAY_RESET_REASON: &str = "secure_relay_protocol_error";
const MAX_ACTIVE_NOISE_RELAY_STREAMS: usize = 128;
const MAX_PENDING_HANDSHAKE_VALIDATIONS: usize = 32;
const HARNESS_KEY_VALIDATION_TIMEOUT: Duration = Duration::from_secs(10);
@@ -51,7 +53,7 @@ const HARNESS_KEY_VALIDATION_TIMEOUT: Duration = Duration::from_secs(10);
pub(crate) trait HarnessKeyValidator: Send + Sync {
fn validate_harness_key(
&self,
harness_public_key: &SecureChannelPublicKey,
harness_public_key: &NoiseChannelPublicKey,
authorization: &str,
) -> impl std::future::Future<Output = Result<(), ExecServerError>> + Send;
}
@@ -62,12 +64,12 @@ pub(crate) trait HarnessKeyValidator: Send + Sync {
/// outer websocket and rendezvous route are treated as untrusted delivery:
/// malformed, unauthorized, or cryptographically invalid streams fail closed
/// without creating a `JsonRpcConnection`.
pub(crate) async fn run_secure_multiplexed_environment<S, V>(
pub(crate) async fn run_noise_multiplexed_environment<S, V>(
stream: WebSocketStream<S>,
processor: ConnectionProcessor,
environment_id: String,
executor_registration_id: String,
identity: SecureChannelIdentity,
identity: NoiseChannelIdentity,
validator: V,
) where
S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
@@ -76,7 +78,7 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
let mut websocket = stream;
let (physical_outgoing_tx, mut physical_outgoing_rx) =
mpsc::channel::<Vec<u8>>(CHANNEL_CAPACITY);
let mut streams: HashMap<String, SecureVirtualStream> = HashMap::new();
let mut streams: HashMap<String, NoiseVirtualStream> = HashMap::new();
let mut pending_handshakes: HashMap<String, PendingHandshake> = HashMap::new();
let mut validation_tasks: JoinSet<HarnessKeyValidationResult> = JoinSet::new();
let mut next_validation_id = 0u64;
@@ -115,12 +117,12 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
continue;
};
if let Err(error) = validation_result.result {
warn!("secure relay harness key validation failed: {error}");
warn!("Noise relay harness key validation failed: {error}");
send_reset(&physical_outgoing_tx, validation_result.stream_id).await;
continue;
}
if streams.len() >= MAX_ACTIVE_SECURE_RELAY_STREAMS {
warn!("secure relay has too many active streams");
if streams.len() >= MAX_ACTIVE_NOISE_RELAY_STREAMS {
warn!("Noise relay has too many active streams");
send_reset(&physical_outgoing_tx, validation_result.stream_id).await;
continue;
}
@@ -131,7 +133,7 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
let (transport, response) = match pending.handshake.complete() {
Ok(completed) => completed,
Err(error) => {
warn!("failed to complete secure relay handshake: {error}");
warn!("failed to complete Noise relay handshake: {error}");
send_reset(&physical_outgoing_tx, validation_result.stream_id).await;
continue;
}
@@ -149,7 +151,7 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
}
streams.insert(
validation_result.stream_id.clone(),
spawn_secure_virtual_stream(
spawn_noise_virtual_stream(
validation_result.stream_id,
processor.clone(),
physical_outgoing_tx.clone(),
@@ -158,7 +160,7 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
);
}
Some(Err(error)) => {
warn!("secure relay harness key validation task failed: {error}");
warn!("Noise relay harness key validation task failed: {error}");
let stream_ids = pending_handshakes.keys().cloned().collect::<Vec<_>>();
pending_handshakes.clear();
for stream_id in stream_ids {
@@ -173,18 +175,18 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
Some(Ok(Message::Binary(payload))) => match decode_relay_message_frame(payload.as_ref()) {
Ok(frame) => frame,
Err(error) => {
warn!("dropping malformed secure relay frame from harness: {error}");
warn!("dropping malformed Noise relay frame from harness: {error}");
continue;
}
},
Some(Ok(Message::Close(_))) | None => break,
Some(Ok(Message::Ping(_) | Message::Pong(_) | Message::Frame(_))) => continue,
Some(Ok(Message::Text(_))) => {
warn!("dropping non-binary secure relay frame from harness");
warn!("dropping non-binary Noise relay frame from harness");
continue;
}
Some(Err(error)) => {
debug!("secure multiplexed environment websocket read failed: {error}");
debug!("Noise multiplexed environment websocket read failed: {error}");
break;
}
}
@@ -193,7 +195,7 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
let kind = match frame.validate() {
Ok(kind) => kind,
Err(error) => {
warn!("dropping invalid secure relay frame: {error}");
warn!("dropping invalid Noise relay frame: {error}");
continue;
}
};
@@ -206,24 +208,24 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
send_reset(&physical_outgoing_tx, stream_id).await;
continue;
}
if streams.len() >= MAX_ACTIVE_SECURE_RELAY_STREAMS {
warn!("secure relay has too many active streams");
if streams.len() >= MAX_ACTIVE_NOISE_RELAY_STREAMS {
warn!("Noise relay has too many active streams");
send_reset(&physical_outgoing_tx, stream_id).await;
continue;
}
if validation_tasks.len() >= MAX_PENDING_HANDSHAKE_VALIDATIONS {
warn!("secure relay has too many pending harness key validations");
warn!("Noise relay has too many pending harness key validations");
send_reset(&physical_outgoing_tx, stream_id).await;
continue;
}
let prologue = match secure_channel_prologue(
let prologue = match noise_channel_prologue(
&environment_id,
&executor_registration_id,
&stream_id,
) {
Ok(prologue) => prologue,
Err(error) => {
warn!("failed to build secure relay prologue: {error}");
warn!("failed to build Noise relay prologue: {error}");
send_reset(&physical_outgoing_tx, stream_id).await;
continue;
}
@@ -231,7 +233,7 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
let request = match frame.into_handshake_payload() {
Ok(request) => request,
Err(error) => {
warn!("failed to read secure relay handshake frame: {error}");
warn!("failed to read Noise relay handshake frame: {error}");
send_reset(&physical_outgoing_tx, stream_id).await;
continue;
}
@@ -240,7 +242,7 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
match PendingResponderHandshake::read_request(&identity, &prologue, &request) {
Ok(pending) => pending,
Err(error) => {
warn!("failed to read secure relay handshake request: {error}");
warn!("failed to read Noise relay handshake request: {error}");
send_reset(&physical_outgoing_tx, stream_id).await;
continue;
}
@@ -252,7 +254,7 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
let authorization = match std::str::from_utf8(pending.payload()) {
Ok(authorization) => authorization.to_string(),
Err(_) => {
warn!("secure relay handshake authorization is not UTF-8");
warn!("Noise relay handshake authorization is not UTF-8");
send_reset(&physical_outgoing_tx, stream_id).await;
continue;
}
@@ -260,7 +262,7 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
let harness_public_key = pending.initiator_public_key().clone();
let validation_id = next_validation_id;
let Some(next_id) = next_validation_id.checked_add(1) else {
warn!("secure relay harness key validation id exhausted");
warn!("Noise relay harness key validation id exhausted");
send_reset(&physical_outgoing_tx, stream_id).await;
continue;
};
@@ -286,7 +288,7 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
{
Ok(result) => result,
Err(_) => Err(ExecServerError::Protocol(
"timed out validating secure relay harness key".to_string(),
"timed out validating Noise relay harness key".to_string(),
)),
};
HarnessKeyValidationResult {
@@ -308,14 +310,14 @@ pub(crate) async fn run_secure_multiplexed_environment<S, V>(
let data = match frame.into_data() {
Ok(data) => data,
Err(error) => {
warn!("dropping malformed secure relay data frame: {error}");
warn!("dropping malformed Noise relay data frame: {error}");
streams.remove(&stream_id);
send_reset(&physical_outgoing_tx, stream_id).await;
continue;
}
};
if let Err(error) = stream.receive_data(data).await {
warn!("failed to process secure relay payload: {error}");
warn!("failed to process Noise relay payload: {error}");
streams.remove(&stream_id);
send_reset(&physical_outgoing_tx, stream_id).await;
}
@@ -348,15 +350,15 @@ struct HarnessKeyValidationResult {
result: Result<(), ExecServerError>,
}
struct SecureVirtualStream {
struct NoiseVirtualStream {
incoming_tx: mpsc::Sender<JsonRpcConnectionEvent>,
disconnected_tx: watch::Sender<bool>,
transport: Arc<Mutex<SecureTransport>>,
transport: Arc<Mutex<NoiseTransport>>,
inbound_ciphertexts: OrderedCiphertextFrames,
inbound_decoder: JsonRpcMessageDecoder,
}
impl SecureVirtualStream {
impl NoiseVirtualStream {
async fn disconnect(self, reason: Option<String>) {
let _ = self.disconnected_tx.send(true);
let _ = self
@@ -388,12 +390,12 @@ impl SecureVirtualStream {
}
}
fn spawn_secure_virtual_stream(
fn spawn_noise_virtual_stream(
stream_id: String,
processor: ConnectionProcessor,
physical_outgoing_tx: mpsc::Sender<Vec<u8>>,
transport: SecureTransport,
) -> SecureVirtualStream {
transport: NoiseTransport,
) -> NoiseVirtualStream {
let (json_outgoing_tx, mut json_outgoing_rx) = mpsc::channel(CHANNEL_CAPACITY);
let (incoming_tx, incoming_rx) = mpsc::channel(CHANNEL_CAPACITY);
let (disconnected_tx, disconnected_rx) = watch::channel(false);
@@ -409,15 +411,15 @@ fn spawn_secure_virtual_stream(
let framed = match frame_jsonrpc_message(&message) {
Ok(framed) => framed,
Err(error) => {
warn!("failed to frame secure virtual stream JSON-RPC payload: {error}");
warn!("failed to frame Noise virtual stream JSON-RPC payload: {error}");
break;
}
};
for plaintext_record in framed.chunks(SECURE_RECORD_PLAINTEXT_LEN) {
for plaintext_record in framed.chunks(NOISE_RECORD_PLAINTEXT_LEN) {
let seq = match take_next_sequence(&mut next_seq) {
Ok(seq) => seq,
Err(error) => {
warn!("secure virtual stream sequence exhausted: {error}");
warn!("Noise virtual stream sequence exhausted: {error}");
break 'writer;
}
};
@@ -430,7 +432,7 @@ fn spawn_secure_virtual_stream(
let ciphertext = match ciphertext {
Ok(ciphertext) => ciphertext,
Err(error) => {
warn!("failed to encrypt secure virtual stream payload: {error}");
warn!("failed to encrypt Noise virtual stream payload: {error}");
break 'writer;
}
};
@@ -449,7 +451,7 @@ fn spawn_secure_virtual_stream(
// exits, including processor shutdown or a cryptographic/send failure.
// Otherwise the peer could wait indefinitely on a dead stream.
let reset =
RelayMessageFrame::reset(writer_stream_id, SECURE_RELAY_RESET_REASON.to_string());
RelayMessageFrame::reset(writer_stream_id, NOISE_RELAY_RESET_REASON.to_string());
let _ = physical_outgoing_tx
.send(encode_relay_message_frame(&reset))
.await;
@@ -466,7 +468,7 @@ fn spawn_secure_virtual_stream(
processor.run_connection(connection).await;
});
SecureVirtualStream {
NoiseVirtualStream {
incoming_tx,
disconnected_tx,
transport,
@@ -476,7 +478,7 @@ fn spawn_secure_virtual_stream(
}
async fn send_reset(physical_outgoing_tx: &mpsc::Sender<Vec<u8>>, stream_id: String) {
let reset = RelayMessageFrame::reset(stream_id, SECURE_RELAY_RESET_REASON.to_string());
let reset = RelayMessageFrame::reset(stream_id, NOISE_RELAY_RESET_REASON.to_string());
let _ = physical_outgoing_tx
.send(encode_relay_message_frame(&reset))
.await;
codex-rs/exec-server/src/secure_relay/environment_tests.rs → codex-rs/exec-server/src/noise_relay/environment_tests.rs
+10 -11
View File
@@ -13,15 +13,15 @@ use tokio_tungstenite::connect_async;
use tokio_tungstenite::tungstenite::Message;
use super::HarnessKeyValidator;
use super::run_secure_multiplexed_environment;
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::encode_relay_message_frame;
use crate::relay_proto::RelayMessageFrame;
use crate::secure_channel::InitiatorHandshake;
use crate::secure_channel::SecureChannelIdentity;
use crate::secure_channel::SecureChannelPublicKey;
use crate::secure_channel::secure_channel_prologue;
use crate::server::ConnectionProcessor;
const ENVIRONMENT_ID: &str = "environment-1";
@@ -36,7 +36,7 @@ struct BlockingValidator {
impl HarnessKeyValidator for BlockingValidator {
fn validate_harness_key(
&self,
_harness_public_key: &SecureChannelPublicKey,
_harness_public_key: &NoiseChannelPublicKey,
_authorization: &str,
) -> impl std::future::Future<Output = Result<(), ExecServerError>> + Send {
let calls = Arc::clone(&self.calls);
@@ -58,10 +58,10 @@ async fn pending_harness_key_validation_does_not_block_new_handshakes() -> Resul
let environment_websocket = accept_async(socket).await?;
let (mut harness_websocket, _response) = harness_connection.await??;
let environment_identity = SecureChannelIdentity::generate()?;
let harness_identity = SecureChannelIdentity::generate()?;
let environment_identity = NoiseChannelIdentity::generate()?;
let harness_identity = NoiseChannelIdentity::generate()?;
let calls = Arc::new(AtomicUsize::new(0));
let environment_task = tokio::spawn(run_secure_multiplexed_environment(
let environment_task = tokio::spawn(run_noise_multiplexed_environment(
environment_websocket,
ConnectionProcessor::new(ExecServerRuntimePaths::new(
std::env::current_exe()?,
@@ -77,8 +77,7 @@ async fn pending_harness_key_validation_does_not_block_new_handshakes() -> Resul
));
for stream_id in ["stream-1", "stream-2"] {
let prologue =
secure_channel_prologue(ENVIRONMENT_ID, EXECUTOR_REGISTRATION_ID, stream_id)?;
let prologue = noise_channel_prologue(ENVIRONMENT_ID, EXECUTOR_REGISTRATION_ID, stream_id)?;
let (_handshake, request) = InitiatorHandshake::start(
&harness_identity,
&environment_identity.public_key(),
codex-rs/exec-server/src/secure_relay/harness.rs → codex-rs/exec-server/src/noise_relay/harness.rs
+32 -32
View File
@@ -14,21 +14,21 @@ use crate::connection::CHANNEL_CAPACITY;
use crate::connection::JsonRpcConnection;
use crate::connection::JsonRpcConnectionEvent;
use crate::connection::JsonRpcTransport;
use crate::noise_channel::InitiatorHandshake;
use crate::noise_channel::NoiseChannelIdentity;
use crate::noise_channel::NoiseChannelPublicKey;
use crate::noise_channel::NoiseTransport;
use crate::noise_channel::noise_channel_prologue;
use crate::noise_relay::message_framing::JsonRpcMessageDecoder;
use crate::noise_relay::message_framing::NOISE_RECORD_PLAINTEXT_LEN;
use crate::noise_relay::message_framing::frame_jsonrpc_message;
use crate::noise_relay::ordered_ciphertext::OrderedCiphertextFrames;
use crate::noise_relay::take_next_sequence;
use crate::relay::RelayFrameBodyKind;
use crate::relay::decode_relay_message_frame;
use crate::relay::encode_relay_message_frame;
use crate::relay_proto::RelayData;
use crate::relay_proto::RelayMessageFrame;
use crate::secure_channel::InitiatorHandshake;
use crate::secure_channel::SecureChannelIdentity;
use crate::secure_channel::SecureChannelPublicKey;
use crate::secure_channel::SecureTransport;
use crate::secure_channel::secure_channel_prologue;
use crate::secure_relay::message_framing::JsonRpcMessageDecoder;
use crate::secure_relay::message_framing::SECURE_RECORD_PLAINTEXT_LEN;
use crate::secure_relay::message_framing::frame_jsonrpc_message;
use crate::secure_relay::ordered_ciphertext::OrderedCiphertextFrames;
use crate::secure_relay::take_next_sequence;
/// Adapt one harness rendezvous websocket into an authenticated JSON-RPC connection.
///
@@ -36,13 +36,13 @@ use crate::secure_relay::take_next_sequence;
/// hybrid IK against the registry-pinned exec-server key. Rendezvous can see
/// stream metadata and ciphertext, but never JSON-RPC plaintext or either
/// endpoint's private key.
pub(crate) fn secure_harness_connection_from_websocket<T, E>(
pub(crate) fn noise_harness_connection_from_websocket<T, E>(
stream: T,
connection_label: String,
environment_id: String,
executor_registration_id: String,
identity: SecureChannelIdentity,
responder_public_key: SecureChannelPublicKey,
identity: NoiseChannelIdentity,
responder_public_key: NoiseChannelPublicKey,
harness_key_authorization: String,
) -> JsonRpcConnection
where
@@ -61,13 +61,13 @@ where
// virtual relay stream before emitting any handshake bytes. A captured
// handshake cannot be spliced onto a different routed connection.
let prologue =
match secure_channel_prologue(&environment_id, &executor_registration_id, &stream_id) {
match noise_channel_prologue(&environment_id, &executor_registration_id, &stream_id) {
Ok(prologue) => prologue,
Err(error) => {
send_disconnected(
&incoming_tx,
&disconnected_tx,
format!("failed to build secure relay prologue: {error}"),
format!("failed to build Noise relay prologue: {error}"),
)
.await;
return;
@@ -84,7 +84,7 @@ where
send_disconnected(
&incoming_tx,
&disconnected_tx,
format!("failed to start secure relay handshake: {error}"),
format!("failed to start Noise relay handshake: {error}"),
)
.await;
return;
@@ -119,7 +119,7 @@ where
send_disconnected(
&incoming_tx,
&disconnected_tx,
"secure relay websocket closed during handshake".to_string(),
"Noise relay websocket closed during handshake".to_string(),
)
.await;
return;
@@ -130,7 +130,7 @@ where
send_disconnected(
&incoming_tx,
&disconnected_tx,
"secure relay websocket closed during handshake".to_string(),
"Noise relay websocket closed during handshake".to_string(),
)
.await;
return;
@@ -140,7 +140,7 @@ where
send_disconnected(
&incoming_tx,
&disconnected_tx,
"secure relay transport expects binary protobuf frames".to_string(),
"Noise relay transport expects binary protobuf frames".to_string(),
)
.await;
return;
@@ -150,7 +150,7 @@ where
&incoming_tx,
&disconnected_tx,
format!(
"failed to read secure relay websocket from {connection_label}: {error}"
"failed to read Noise relay websocket from {connection_label}: {error}"
),
)
.await;
@@ -163,7 +163,7 @@ where
send_disconnected(
&incoming_tx,
&disconnected_tx,
format!("failed to parse secure relay frame: {error}"),
format!("failed to parse Noise relay frame: {error}"),
)
.await;
return;
@@ -188,7 +188,7 @@ where
send_disconnected(
&incoming_tx,
&disconnected_tx,
format!("secure relay handshake failed: {error}"),
format!("Noise relay handshake failed: {error}"),
)
.await;
return;
@@ -201,7 +201,7 @@ where
&disconnected_tx,
frame
.into_reset_reason()
.unwrap_or_else(|| "secure relay reset during handshake".to_string()),
.unwrap_or_else(|| "Noise relay reset during handshake".to_string()),
)
.await;
return;
@@ -215,7 +215,7 @@ where
send_disconnected(
&incoming_tx,
&disconnected_tx,
"secure relay received data before handshake completion".to_string(),
"Noise relay received data before handshake completion".to_string(),
)
.await;
return;
@@ -238,22 +238,22 @@ where
let framed = match frame_jsonrpc_message(&message) {
Ok(framed) => framed,
Err(error) => {
warn!("failed to frame JSON-RPC payload for secure relay: {error}");
warn!("failed to frame JSON-RPC payload for Noise relay: {error}");
break;
}
};
for plaintext_record in framed.chunks(SECURE_RECORD_PLAINTEXT_LEN) {
for plaintext_record in framed.chunks(NOISE_RECORD_PLAINTEXT_LEN) {
let seq = match take_next_sequence(&mut next_outbound_seq) {
Ok(seq) => seq,
Err(error) => {
warn!("secure relay sequence exhausted: {error}");
warn!("Noise relay sequence exhausted: {error}");
break 'relay;
}
};
let ciphertext = match transport.encrypt(plaintext_record) {
Ok(ciphertext) => ciphertext,
Err(error) => {
warn!("failed to encrypt JSON-RPC payload for secure relay: {error}");
warn!("failed to encrypt JSON-RPC payload for Noise relay: {error}");
break 'relay;
}
};
@@ -320,7 +320,7 @@ where
Ok(RelayFrameBodyKind::Handshake) | Err(_) => {
send_malformed(
&incoming_tx,
"secure relay received invalid post-handshake frame".to_string(),
"Noise relay received invalid post-handshake frame".to_string(),
)
.await;
break;
@@ -332,13 +332,13 @@ where
Ok(Message::Text(_)) => {
send_malformed(
&incoming_tx,
"secure relay transport expects binary protobuf frames".to_string(),
"Noise relay transport expects binary protobuf frames".to_string(),
)
.await;
break;
}
Err(error) => {
debug!("secure relay websocket read failed: {error}");
debug!("Noise relay websocket read failed: {error}");
break;
}
}
@@ -359,7 +359,7 @@ where
async fn receive_data(
inbound_ciphertexts: &mut OrderedCiphertextFrames,
transport: &mut SecureTransport,
transport: &mut NoiseTransport,
decoder: &mut JsonRpcMessageDecoder,
data: RelayData,
incoming_tx: &mpsc::Sender<JsonRpcConnectionEvent>,
codex-rs/exec-server/src/secure_relay/message_framing.rs → codex-rs/exec-server/src/noise_relay/message_framing.rs
+10 -10
View File
@@ -3,8 +3,8 @@ use codex_app_server_protocol::JSONRPCMessage;
use crate::ExecServerError;
const LENGTH_PREFIX_BYTES: usize = size_of::<u32>();
const MAX_SECURE_JSONRPC_MESSAGE_LEN: usize = 64 * 1024 * 1024;
pub(crate) const SECURE_RECORD_PLAINTEXT_LEN: usize = 60 * 1024;
const MAX_NOISE_JSONRPC_MESSAGE_LEN: usize = 64 * 1024 * 1024;
pub(crate) const NOISE_RECORD_PLAINTEXT_LEN: usize = 60 * 1024;
/// Serialize one JSON-RPC message into the encrypted record byte stream.
///
@@ -16,9 +16,9 @@ pub(crate) fn frame_jsonrpc_message(message: &JSONRPCMessage) -> Result<Vec<u8>,
let mut framed = vec![0; LENGTH_PREFIX_BYTES];
serde_json::to_writer(&mut framed, message)?;
let message_len = framed.len() - LENGTH_PREFIX_BYTES;
if message_len > MAX_SECURE_JSONRPC_MESSAGE_LEN {
if message_len > MAX_NOISE_JSONRPC_MESSAGE_LEN {
return Err(ExecServerError::Protocol(
"secure relay JSON-RPC message exceeds maximum length".to_string(),
"Noise relay JSON-RPC message exceeds maximum length".to_string(),
));
}
framed[..LENGTH_PREFIX_BYTES].copy_from_slice(&(message_len as u32).to_be_bytes());
@@ -37,9 +37,9 @@ impl JsonRpcMessageDecoder {
&mut self,
plaintext_record: &[u8],
) -> Result<Vec<JSONRPCMessage>, ExecServerError> {
if plaintext_record.len() > SECURE_RECORD_PLAINTEXT_LEN {
if plaintext_record.len() > NOISE_RECORD_PLAINTEXT_LEN {
return Err(ExecServerError::Protocol(
"secure relay plaintext record exceeds maximum length".to_string(),
"Noise relay plaintext record exceeds maximum length".to_string(),
));
}
self.buffered.extend_from_slice(plaintext_record);
@@ -63,9 +63,9 @@ impl JsonRpcMessageDecoder {
}
// Even before a message is complete, keep reassembly memory bounded.
if self.buffered.len() > LENGTH_PREFIX_BYTES + MAX_SECURE_JSONRPC_MESSAGE_LEN {
if self.buffered.len() > LENGTH_PREFIX_BYTES + MAX_NOISE_JSONRPC_MESSAGE_LEN {
return Err(ExecServerError::Protocol(
"secure relay JSON-RPC reassembly buffer exceeds maximum length".to_string(),
"Noise relay JSON-RPC reassembly buffer exceeds maximum length".to_string(),
));
}
Ok(messages)
@@ -78,9 +78,9 @@ impl JsonRpcMessageDecoder {
let message_len = u32::from_be_bytes([prefix[0], prefix[1], prefix[2], prefix[3]]) as usize;
// Zero-length is never a JSON-RPC message, and rejecting an oversized
// declaration immediately avoids waiting for attacker-controlled data.
if message_len == 0 || message_len > MAX_SECURE_JSONRPC_MESSAGE_LEN {
if message_len == 0 || message_len > MAX_NOISE_JSONRPC_MESSAGE_LEN {
return Err(ExecServerError::Protocol(
"secure relay JSON-RPC message has invalid length".to_string(),
"Noise relay JSON-RPC message has invalid length".to_string(),
));
}
Ok(Some(message_len))
codex-rs/exec-server/src/secure_relay/message_framing_tests.rs → codex-rs/exec-server/src/noise_relay/message_framing_tests.rs
+7 -7
View File
@@ -3,8 +3,8 @@ use codex_app_server_protocol::JSONRPCNotification;
use pretty_assertions::assert_eq;
use super::JsonRpcMessageDecoder;
use super::MAX_SECURE_JSONRPC_MESSAGE_LEN;
use super::SECURE_RECORD_PLAINTEXT_LEN;
use super::MAX_NOISE_JSONRPC_MESSAGE_LEN;
use super::NOISE_RECORD_PLAINTEXT_LEN;
use super::frame_jsonrpc_message;
use crate::ExecServerError;
@@ -21,7 +21,7 @@ fn fragments_and_reassembles_large_jsonrpc_message() {
let mut decoder = JsonRpcMessageDecoder::default();
let mut decoded = Vec::new();
for record in framed.chunks(SECURE_RECORD_PLAINTEXT_LEN) {
for record in framed.chunks(NOISE_RECORD_PLAINTEXT_LEN) {
decoded.extend(decoder.push(record).unwrap());
}
@@ -31,12 +31,12 @@ fn fragments_and_reassembles_large_jsonrpc_message() {
#[test]
fn rejects_declared_message_length_above_limit_without_payload() {
let mut decoder = JsonRpcMessageDecoder::default();
let declared_len = (MAX_SECURE_JSONRPC_MESSAGE_LEN as u32 + 1).to_be_bytes();
let declared_len = (MAX_NOISE_JSONRPC_MESSAGE_LEN as u32 + 1).to_be_bytes();
assert!(matches!(
decoder.push(&declared_len),
Err(ExecServerError::Protocol(message))
if message == "secure relay JSON-RPC message has invalid length"
if message == "Noise relay JSON-RPC message has invalid length"
));
}
@@ -45,8 +45,8 @@ fn rejects_oversized_plaintext_record() {
let mut decoder = JsonRpcMessageDecoder::default();
assert!(matches!(
decoder.push(&vec![0; SECURE_RECORD_PLAINTEXT_LEN + 1]),
decoder.push(&vec![0; NOISE_RECORD_PLAINTEXT_LEN + 1]),
Err(ExecServerError::Protocol(message))
if message == "secure relay plaintext record exceeds maximum length"
if message == "Noise relay plaintext record exceeds maximum length"
));
}
codex-rs/exec-server/src/secure_relay/mod.rs → codex-rs/exec-server/src/noise_relay/mod.rs
+9 -9
View File
@@ -8,18 +8,18 @@ use tokio_tungstenite::tungstenite::protocol::WebSocketConfig;
use crate::ExecServerError;
pub(crate) use environment::HarnessKeyValidator;
pub(crate) use environment::run_secure_multiplexed_environment;
pub(crate) use harness::secure_harness_connection_from_websocket;
pub(crate) use environment::run_noise_multiplexed_environment;
pub(crate) use harness::noise_harness_connection_from_websocket;
// This bounds allocation in tungstenite before protobuf and secure-record
// This bounds allocation in tungstenite before protobuf and Noise record
// validation run. It comfortably fits one maximum Noise record plus metadata.
const MAX_SECURE_RELAY_WEBSOCKET_MESSAGE_SIZE: usize = 256 * 1024;
const MAX_NOISE_RELAY_WEBSOCKET_MESSAGE_SIZE: usize = 256 * 1024;
/// Return the websocket limits required by every secure relay endpoint.
pub(crate) fn secure_relay_websocket_config() -> WebSocketConfig {
/// Return the websocket limits required by every Noise relay endpoint.
pub(crate) fn noise_relay_websocket_config() -> WebSocketConfig {
WebSocketConfig::default()
.max_frame_size(Some(MAX_SECURE_RELAY_WEBSOCKET_MESSAGE_SIZE))
.max_message_size(Some(MAX_SECURE_RELAY_WEBSOCKET_MESSAGE_SIZE))
.max_frame_size(Some(MAX_NOISE_RELAY_WEBSOCKET_MESSAGE_SIZE))
.max_message_size(Some(MAX_NOISE_RELAY_WEBSOCKET_MESSAGE_SIZE))
}
fn take_next_sequence(next_seq: &mut u32) -> Result<u32, ExecServerError> {
@@ -27,7 +27,7 @@ fn take_next_sequence(next_seq: &mut u32) -> Result<u32, ExecServerError> {
// Noise nonce. Reusing zero after u32::MAX would be ambiguous and unsafe.
let seq = *next_seq;
*next_seq = next_seq.checked_add(1).ok_or_else(|| {
ExecServerError::Protocol("secure relay sequence number exhausted".to_string())
ExecServerError::Protocol("Noise relay sequence number exhausted".to_string())
})?;
Ok(seq)
}
codex-rs/exec-server/src/secure_relay/ordered_ciphertext.rs → codex-rs/exec-server/src/noise_relay/ordered_ciphertext.rs
+5 -5
View File
@@ -11,7 +11,7 @@ const MAX_PENDING_BYTES: usize = 1024 * 1024;
/// Relay delivery can be duplicated or reordered, but Noise transport nonces
/// are strictly ordered. This type absorbs only a small reliable-delivery
/// window and releases ciphertexts exactly once in nonce order. It must sit
/// before `SecureTransport::decrypt`; attempting to decrypt a future record
/// before `NoiseTransport::decrypt`; attempting to decrypt a future record
/// would advance or desynchronize cryptographic state.
#[derive(Default)]
pub(crate) struct OrderedCiphertextFrames {
@@ -39,7 +39,7 @@ impl OrderedCiphertextFrames {
// while forcing unbounded pre-decryption state.
if seq - self.next_seq > MAX_REORDER_DISTANCE {
return Err(ExecServerError::Protocol(
"secure relay ciphertext exceeds reorder window".to_string(),
"Noise relay ciphertext exceeds reorder window".to_string(),
));
}
let pending_bytes = self
@@ -47,12 +47,12 @@ impl OrderedCiphertextFrames {
.checked_add(payload.len())
.ok_or_else(|| {
ExecServerError::Protocol(
"secure relay pending ciphertext byte count overflowed".to_string(),
"Noise relay pending ciphertext byte count overflowed".to_string(),
)
})?;
if self.pending.len() >= MAX_PENDING_FRAMES || pending_bytes > MAX_PENDING_BYTES {
return Err(ExecServerError::Protocol(
"secure relay pending ciphertext buffer is full".to_string(),
"Noise relay pending ciphertext buffer is full".to_string(),
));
}
self.pending.insert(seq, payload);
@@ -74,7 +74,7 @@ impl OrderedCiphertextFrames {
fn advance(&mut self) -> Result<(), ExecServerError> {
self.next_seq = self.next_seq.checked_add(1).ok_or_else(|| {
ExecServerError::Protocol("secure relay sequence number exhausted".to_string())
ExecServerError::Protocol("Noise relay sequence number exhausted".to_string())
})?;
Ok(())
}
codex-rs/exec-server/src/secure_relay/ordered_ciphertext_tests.rs → codex-rs/exec-server/src/noise_relay/ordered_ciphertext_tests.rs
View File
codex-rs/exec-server/src/relay.rs
+2 -2
View File
@@ -94,7 +94,7 @@ impl RelayMessageFrame {
}
/// Validate cleartext routing metadata before a frame reaches either the
/// direct JSON-RPC parser or the secure relay state machine.
/// direct JSON-RPC parser or the Noise relay state machine.
///
/// The encrypted path intentionally leaves this metadata visible to
/// rendezvous, so it must be canonical and tightly bounded everywhere.
@@ -377,7 +377,7 @@ where
/// The physical websocket carries protobuf [`RelayMessageFrame`] values. Each
/// frame's `stream_id` selects an independent logical JSON-RPC connection, so a
/// single registered exec-server can serve multiple orchestrator sessions.
/// This runner intentionally preserves the protocol used before secure relay
/// This runner intentionally preserves the protocol used before Noise relay
/// support was introduced. Callers must select the separate secure runner when
/// Noise protection was explicitly negotiated during registration.
///
codex-rs/exec-server/src/remote.rs
+34 -34
View File
@@ -14,12 +14,12 @@ use codex_utils_rustls_provider::ensure_rustls_crypto_provider;
use crate::ExecServerError;
use crate::ExecServerRuntimePaths;
use crate::SecureChannelIdentity;
use crate::SecureChannelPublicKey;
use crate::NoiseChannelIdentity;
use crate::NoiseChannelPublicKey;
use crate::noise_relay::HarnessKeyValidator;
use crate::noise_relay::noise_relay_websocket_config;
use crate::noise_relay::run_noise_multiplexed_environment;
use crate::relay::run_multiplexed_environment;
use crate::secure_relay::HarnessKeyValidator;
use crate::secure_relay::run_secure_multiplexed_environment;
use crate::secure_relay::secure_relay_websocket_config;
use crate::server::ConnectionProcessor;
const ERROR_BODY_PREVIEW_BYTES: usize = 4096;
@@ -59,7 +59,7 @@ impl EnvironmentRegistryClient {
/// Register using the original body-less registry contract.
///
/// This remains the default so a new exec-server can still connect through
/// registries and harnesses that have not rolled out secure relay support.
/// registries and harnesses that have not rolled out Noise relay support.
async fn register_legacy_environment(
&self,
environment_id: &str,
@@ -76,15 +76,15 @@ impl EnvironmentRegistryClient {
self.parse_json_response(response).await
}
/// Register the exec-server's static Noise identity with a secure-aware registry.
/// Register the exec-server's static Noise identity with a Noise-aware registry.
///
/// Supplying this request body is the protocol-level opt in. Registries can
/// therefore distinguish secure registrations from the legacy body-less
/// therefore distinguish Noise registrations from the legacy body-less
/// contract without guessing based on binary version or rollout state.
async fn register_noise_environment(
&self,
environment_id: &str,
executor_public_key: &SecureChannelPublicKey,
executor_public_key: &NoiseChannelPublicKey,
) -> Result<EnvironmentRegistryNoiseRegistrationResponse, ExecServerError> {
let response = self
.http
@@ -94,7 +94,7 @@ impl EnvironmentRegistryClient {
))
.headers(self.auth_provider.to_auth_headers())
.json(&EnvironmentRegistryRegistrationRequest {
security_profile: SECURE_RELAY_SECURITY_PROFILE,
security_profile: NOISE_RELAY_SECURITY_PROFILE,
executor_public_key,
})
.send()
@@ -106,7 +106,7 @@ impl EnvironmentRegistryClient {
&self,
environment_id: &str,
executor_registration_id: &str,
harness_public_key: &SecureChannelPublicKey,
harness_public_key: &NoiseChannelPublicKey,
harness_key_authorization: &str,
) -> Result<(), ExecServerError> {
let response = self
@@ -149,12 +149,12 @@ impl EnvironmentRegistryClient {
}
}
const SECURE_RELAY_SECURITY_PROFILE: &str = "noise_hybrid_ik_v1";
const NOISE_RELAY_SECURITY_PROFILE: &str = "noise_hybrid_ik_v1";
#[derive(Serialize)]
struct EnvironmentRegistryRegistrationRequest<'a> {
security_profile: &'static str,
executor_public_key: &'a SecureChannelPublicKey,
executor_public_key: &'a NoiseChannelPublicKey,
}
#[derive(Debug, Clone, Eq, PartialEq, Deserialize)]
@@ -174,7 +174,7 @@ struct EnvironmentRegistryNoiseRegistrationResponse {
#[derive(Serialize)]
struct EnvironmentRegistryHarnessKeyValidationRequest<'a> {
executor_registration_id: &'a str,
harness_public_key: &'a SecureChannelPublicKey,
harness_public_key: &'a NoiseChannelPublicKey,
harness_key_authorization: &'a str,
}
@@ -191,7 +191,7 @@ struct RegistryHarnessKeyValidator {
impl HarnessKeyValidator for RegistryHarnessKeyValidator {
async fn validate_harness_key(
&self,
harness_public_key: &SecureChannelPublicKey,
harness_public_key: &NoiseChannelPublicKey,
authorization: &str,
) -> Result<(), ExecServerError> {
self.client
@@ -205,28 +205,28 @@ impl HarnessKeyValidator for RegistryHarnessKeyValidator {
}
}
/// Security mode used for an exec-server's registered remote relay.
/// Protocol used for an exec-server's registered remote relay.
///
/// Legacy is intentionally the default during rollout. Noise must be selected
/// explicitly so mixed-version deployments keep the original registry and relay
/// contract until both endpoints are ready for the secure protocol.
/// contract until both endpoints are ready for Noise.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub enum RemoteRelaySecurity {
pub enum RemoteRelayProtocol {
#[default]
Legacy,
Noise,
}
enum RemoteRelaySecurityState {
enum RemoteRelayProtocolState {
Legacy,
Noise(SecureChannelIdentity),
Noise(NoiseChannelIdentity),
}
enum RegisteredRemoteEnvironment {
Legacy(EnvironmentRegistryRegistrationResponse),
Noise {
response: EnvironmentRegistryNoiseRegistrationResponse,
identity: SecureChannelIdentity,
identity: NoiseChannelIdentity,
},
}
@@ -252,7 +252,7 @@ pub struct RemoteEnvironmentConfig {
pub base_url: String,
pub environment_id: String,
pub name: String,
pub relay_security: RemoteRelaySecurity,
pub relay_protocol: RemoteRelayProtocol,
auth_provider: SharedAuthProvider,
}
@@ -262,7 +262,7 @@ impl std::fmt::Debug for RemoteEnvironmentConfig {
.field("base_url", &self.base_url)
.field("environment_id", &self.environment_id)
.field("name", &self.name)
.field("relay_security", &self.relay_security)
.field("relay_protocol", &self.relay_protocol)
.field("auth_provider", &"<redacted>")
.finish()
}
@@ -279,7 +279,7 @@ impl RemoteEnvironmentConfig {
base_url,
environment_id,
name: "codex-exec-server".to_string(),
relay_security: RemoteRelaySecurity::Legacy,
relay_protocol: RemoteRelayProtocol::Legacy,
auth_provider,
})
}
@@ -295,26 +295,26 @@ pub async fn run_remote_environment(
let client =
EnvironmentRegistryClient::new(config.base_url.clone(), config.auth_provider.clone())?;
let processor = ConnectionProcessor::new(runtime_paths);
let security_state = match config.relay_security {
RemoteRelaySecurity::Legacy => RemoteRelaySecurityState::Legacy,
RemoteRelaySecurity::Noise => {
RemoteRelaySecurityState::Noise(SecureChannelIdentity::generate()?)
let protocol_state = match config.relay_protocol {
RemoteRelayProtocol::Legacy => RemoteRelayProtocolState::Legacy,
RemoteRelayProtocol::Noise => {
RemoteRelayProtocolState::Noise(NoiseChannelIdentity::generate()?)
}
};
let mut backoff = Duration::from_secs(1);
loop {
let registration = match &security_state {
RemoteRelaySecurityState::Legacy => RegisteredRemoteEnvironment::Legacy(
let registration = match &protocol_state {
RemoteRelayProtocolState::Legacy => RegisteredRemoteEnvironment::Legacy(
client
.register_legacy_environment(&config.environment_id)
.await?,
),
RemoteRelaySecurityState::Noise(identity) => {
RemoteRelayProtocolState::Noise(identity) => {
let response = client
.register_noise_environment(&config.environment_id, &identity.public_key())
.await?;
if response.security_profile != SECURE_RELAY_SECURITY_PROFILE {
if response.security_profile != NOISE_RELAY_SECURITY_PROFILE {
return Err(ExecServerError::Protocol(format!(
"environment registry returned unsupported security profile `{}`",
response.security_profile
@@ -337,7 +337,7 @@ pub async fn run_remote_environment(
);
let websocket_config = match &registration {
RegisteredRemoteEnvironment::Legacy(_) => None,
RegisteredRemoteEnvironment::Noise { .. } => Some(secure_relay_websocket_config()),
RegisteredRemoteEnvironment::Noise { .. } => Some(noise_relay_websocket_config()),
};
match timeout(
@@ -357,7 +357,7 @@ pub async fn run_remote_environment(
run_multiplexed_environment(websocket, processor.clone()).await;
}
RegisteredRemoteEnvironment::Noise { response, identity } => {
run_secure_multiplexed_environment(
run_noise_multiplexed_environment(
websocket,
processor.clone(),
response.environment_id,
codex-rs/exec-server/src/remote_tests.rs
+8 -8
View File
@@ -75,20 +75,20 @@ async fn register_environment_posts_with_auth_provider_headers() {
#[tokio::test]
async fn register_noise_environment_posts_security_profile_and_public_key() {
let server = MockServer::start().await;
let executor_public_key = SecureChannelIdentity::generate()
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": SECURE_RELAY_SECURITY_PROFILE,
"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/secure",
"security_profile": SECURE_RELAY_SECURITY_PROFILE,
"url": "wss://rendezvous.test/noise",
"security_profile": NOISE_RELAY_SECURITY_PROFILE,
"executor_registration_id": "registration-1",
})))
.mount(&server)
@@ -99,14 +99,14 @@ async fn register_noise_environment_posts_security_profile_and_public_key() {
let response = client
.register_noise_environment("environment-requested", &executor_public_key)
.await
.expect("register secure environment");
.expect("register Noise environment");
assert_eq!(
response,
EnvironmentRegistryNoiseRegistrationResponse {
environment_id: "environment-requested".to_string(),
url: "wss://rendezvous.test/secure".to_string(),
security_profile: SECURE_RELAY_SECURITY_PROFILE.to_string(),
url: "wss://rendezvous.test/noise".to_string(),
security_profile: NOISE_RELAY_SECURITY_PROFILE.to_string(),
executor_registration_id: "registration-1".to_string(),
}
);
@@ -174,7 +174,7 @@ fn remote_environment_config_accepts_cloud_environment_id() {
.expect("config");
assert_eq!(config.environment_id, environment_id);
assert_eq!(config.relay_security, RemoteRelaySecurity::Legacy);
assert_eq!(config.relay_protocol, RemoteRelayProtocol::Legacy);
}
#[test]
codex-rs/exec-server/tests/relay.rs
+39 -39
View File
@@ -22,14 +22,14 @@ use codex_exec_server::ExecServerClient;
use codex_exec_server::ExecServerError;
use codex_exec_server::ExecServerRuntimePaths;
use codex_exec_server::FsReadFileParams;
use codex_exec_server::NoiseChannelIdentity;
use codex_exec_server::NoiseChannelPublicKey;
use codex_exec_server::NoiseRendezvousConnectArgs;
use codex_exec_server::NoiseRendezvousConnectBundle;
use codex_exec_server::NoiseRendezvousConnectProvider;
use codex_exec_server::ProcessId;
use codex_exec_server::RemoteEnvironmentConfig;
use codex_exec_server::RemoteRelaySecurity;
use codex_exec_server::SecureChannelIdentity;
use codex_exec_server::SecureChannelPublicKey;
use codex_exec_server::SecureRendezvousConnectArgs;
use codex_exec_server::SecureRendezvousConnectBundle;
use codex_exec_server::SecureRendezvousConnectProvider;
use codex_exec_server::RemoteRelayProtocol;
use codex_utils_absolute_path::AbsolutePathBuf;
use futures::FutureExt;
use futures::SinkExt;
@@ -56,7 +56,7 @@ use wiremock::matchers::header;
use wiremock::matchers::method;
use wiremock::matchers::path;
const ENVIRONMENT_ID: &str = "env-secure-relay-test";
const ENVIRONMENT_ID: &str = "env-noise-relay-test";
const EXECUTOR_REGISTRATION_ID: &str = "registration-1";
const HARNESS_KEY_AUTHORIZATION: &str = "harness-key-authorization";
const REGISTRY_TOKEN: &str = "registry-token";
@@ -74,16 +74,16 @@ impl AuthProvider for StaticRegistryAuthProvider {
}
}
struct FailingSecureConnectProvider {
struct FailingNoiseConnectProvider {
attempts: Arc<AtomicUsize>,
}
impl SecureRendezvousConnectProvider for FailingSecureConnectProvider {
impl NoiseRendezvousConnectProvider for FailingNoiseConnectProvider {
fn environment_id(&self) -> &str {
ENVIRONMENT_ID
}
fn connect_args(&self) -> BoxFuture<'_, Result<SecureRendezvousConnectArgs, ExecServerError>> {
fn connect_args(&self) -> BoxFuture<'_, Result<NoiseRendezvousConnectArgs, ExecServerError>> {
self.attempts.fetch_add(1, Ordering::SeqCst);
async {
Err(ExecServerError::Protocol(
@@ -94,20 +94,20 @@ impl SecureRendezvousConnectProvider for FailingSecureConnectProvider {
}
}
struct WrongEnvironmentSecureConnectProvider {
harness_identity: SecureChannelIdentity,
executor_public_key: SecureChannelPublicKey,
struct WrongEnvironmentNoiseConnectProvider {
harness_identity: NoiseChannelIdentity,
executor_public_key: NoiseChannelPublicKey,
}
impl SecureRendezvousConnectProvider for WrongEnvironmentSecureConnectProvider {
impl NoiseRendezvousConnectProvider for WrongEnvironmentNoiseConnectProvider {
fn environment_id(&self) -> &str {
ENVIRONMENT_ID
}
fn connect_args(&self) -> BoxFuture<'_, Result<SecureRendezvousConnectArgs, ExecServerError>> {
fn connect_args(&self) -> BoxFuture<'_, Result<NoiseRendezvousConnectArgs, ExecServerError>> {
async move {
Ok(SecureRendezvousConnectArgs::new(
SecureRendezvousConnectBundle {
Ok(NoiseRendezvousConnectArgs::new(
NoiseRendezvousConnectBundle {
websocket_url: "ws://127.0.0.1:1".to_string(),
environment_id: "wrong-environment".to_string(),
executor_registration_id: EXECUTOR_REGISTRATION_ID.to_string(),
@@ -115,7 +115,7 @@ impl SecureRendezvousConnectProvider for WrongEnvironmentSecureConnectProvider {
harness_key_authorization: HARNESS_KEY_AUTHORIZATION.to_string(),
},
self.harness_identity.clone(),
"secure-relay-test".to_string(),
"noise-relay-test".to_string(),
))
}
.boxed()
@@ -150,7 +150,7 @@ async fn remote_environment_uses_legacy_relay_by_default() -> Result<()> {
ENVIRONMENT_ID.to_string(),
static_registry_auth_provider(),
)?;
assert_eq!(config.relay_security, RemoteRelaySecurity::Legacy);
assert_eq!(config.relay_protocol, RemoteRelayProtocol::Legacy);
let remote_environment = tokio::spawn(codex_exec_server::run_remote_environment(
config,
runtime_paths,
@@ -205,18 +205,18 @@ async fn remote_environment_uses_legacy_relay_by_default() -> Result<()> {
}
#[tokio::test]
async fn secure_environment_refreshes_bundle_for_each_connection_attempt() -> Result<()> {
async fn noise_environment_refreshes_bundle_for_each_connection_attempt() -> Result<()> {
let attempts = Arc::new(AtomicUsize::new(0));
let manager = EnvironmentManager::without_environments();
manager.upsert_secure_environment(
manager.upsert_noise_environment(
ENVIRONMENT_ID.to_string(),
Arc::new(FailingSecureConnectProvider {
Arc::new(FailingNoiseConnectProvider {
attempts: Arc::clone(&attempts),
}),
)?;
let backend = manager
.get_environment(ENVIRONMENT_ID)
.context("secure environment should be materialized")?
.context("Noise environment should be materialized")?
.get_exec_backend();
for attempt in 1..=2 {
@@ -244,18 +244,18 @@ async fn secure_environment_refreshes_bundle_for_each_connection_attempt() -> Re
}
#[tokio::test]
async fn secure_environment_rejects_provider_bundle_for_another_environment() -> Result<()> {
async fn noise_environment_rejects_provider_bundle_for_another_environment() -> Result<()> {
let manager = EnvironmentManager::without_environments();
manager.upsert_secure_environment(
manager.upsert_noise_environment(
ENVIRONMENT_ID.to_string(),
Arc::new(WrongEnvironmentSecureConnectProvider {
harness_identity: SecureChannelIdentity::generate()?,
executor_public_key: SecureChannelIdentity::generate()?.public_key(),
Arc::new(WrongEnvironmentNoiseConnectProvider {
harness_identity: NoiseChannelIdentity::generate()?,
executor_public_key: NoiseChannelIdentity::generate()?.public_key(),
}),
)?;
let backend = manager
.get_environment(ENVIRONMENT_ID)
.context("secure environment should be materialized")?
.context("Noise environment should be materialized")?
.get_exec_backend();
let result = backend
@@ -274,7 +274,7 @@ async fn secure_environment_rejects_provider_bundle_for_another_environment() ->
assert!(matches!(
result,
Err(ExecServerError::Protocol(ref message))
if message == "secure rendezvous provider returned a different environment id"
if message == "Noise rendezvous provider returned a different environment id"
));
Ok(())
}
@@ -313,7 +313,7 @@ async fn remote_environment_routes_encrypted_exec_server_rpc() -> Result<()> {
ENVIRONMENT_ID.to_string(),
static_registry_auth_provider(),
)?;
config.relay_security = RemoteRelaySecurity::Noise;
config.relay_protocol = RemoteRelayProtocol::Noise;
let remote_environment = tokio::spawn(codex_exec_server::run_remote_environment(
config,
runtime_paths,
@@ -321,9 +321,9 @@ async fn remote_environment_routes_encrypted_exec_server_rpc() -> Result<()> {
let environment_websocket = accept_websocket(&listener, "environment").await?;
let executor_public_key = registered_executor_public_key(&registry).await?;
let harness_identity = SecureChannelIdentity::generate()?;
let client_args = SecureRendezvousConnectArgs::new(
SecureRendezvousConnectBundle {
let harness_identity = NoiseChannelIdentity::generate()?;
let client_args = NoiseRendezvousConnectArgs::new(
NoiseRendezvousConnectBundle {
websocket_url: format!("{rendezvous_url}/relay?role=harness"),
environment_id: ENVIRONMENT_ID.to_string(),
executor_registration_id: EXECUTOR_REGISTRATION_ID.to_string(),
@@ -331,10 +331,10 @@ async fn remote_environment_routes_encrypted_exec_server_rpc() -> Result<()> {
harness_key_authorization: HARNESS_KEY_AUTHORIZATION.to_string(),
},
harness_identity,
"secure-relay-test".to_string(),
"noise-relay-test".to_string(),
);
let client_task =
tokio::spawn(async move { ExecServerClient::connect_secure_rendezvous(client_args).await });
tokio::spawn(async move { ExecServerClient::connect_noise_rendezvous(client_args).await });
let harness_websocket = accept_websocket(&listener, "harness").await?;
let captured_frames = Arc::new(Mutex::new(Vec::new()));
let relay_task = tokio::spawn(proxy_relay_frames(
@@ -344,7 +344,7 @@ async fn remote_environment_routes_encrypted_exec_server_rpc() -> Result<()> {
));
let client = timeout(TEST_TIMEOUT, client_task)
.await
.context("secure harness client should connect")???;
.context("Noise harness client should connect")???;
let response = client
.exec(ExecParams {
@@ -403,7 +403,7 @@ async fn accept_websocket(
.map_err(Into::into)
}
async fn registered_executor_public_key(registry: &MockServer) -> Result<SecureChannelPublicKey> {
async fn registered_executor_public_key(registry: &MockServer) -> Result<NoiseChannelPublicKey> {
let requests = registry
.received_requests()
.await
@@ -468,7 +468,7 @@ fn assert_relay_data_is_encrypted(captured_frames: &Mutex<Vec<Vec<u8>>>) -> Resu
let payload = String::from_utf8_lossy(&data.payload);
assert!(!payload.contains("initialize"));
assert!(!payload.contains("process/start"));
assert!(!payload.contains("secure-relay-test"));
assert!(!payload.contains("noise-relay-test"));
}
assert!(
data_frames >= 4,