Route local-only app-server gating through processors (#23551)

## Summary
- move local-only app-server gating out of `MessageProcessor`
- let `fs/*`, `command/exec`, and `process/spawn` resolve local
availability inside their owning processors
- keep `fs/*` mounted for the future environment-param path while
preserving current no-local error behavior

## Validation
- not run locally per Codex repo guidance
This commit is contained in:
starr-openai authored and GitHub committed 2026-05-19 14:38:03 -07:00
1 parent 954a9c8579
commit 1509ae6d8d
4 files changed
+73 -64

No files matched your search

+25 -53
View File
@@ -7,7 +7,6 @@ use std::sync::atomic::AtomicBool;
use crate::attestation::app_server_attestation_provider; use crate::attestation::app_server_attestation_provider;
use crate::config_manager::ConfigManager; use crate::config_manager::ConfigManager;
use crate::connection_rpc_gate::ConnectionRpcGate; use crate::connection_rpc_gate::ConnectionRpcGate;
use crate::error_code::internal_error;
use crate::error_code::invalid_request; use crate::error_code::invalid_request;
use crate::extensions::guardian_agent_spawner; use crate::extensions::guardian_agent_spawner;
use crate::extensions::thread_extensions; use crate::extensions::thread_extensions;
@@ -166,11 +165,10 @@ pub(crate) struct MessageProcessor {
command_exec_processor: CommandExecRequestProcessor, command_exec_processor: CommandExecRequestProcessor,
process_exec_processor: ProcessExecRequestProcessor, process_exec_processor: ProcessExecRequestProcessor,
config_processor: ConfigRequestProcessor, config_processor: ConfigRequestProcessor,
environment_manager: Arc<EnvironmentManager>,
environment_processor: EnvironmentRequestProcessor, environment_processor: EnvironmentRequestProcessor,
external_agent_config_processor: ExternalAgentConfigRequestProcessor, external_agent_config_processor: ExternalAgentConfigRequestProcessor,
feedback_processor: FeedbackRequestProcessor, feedback_processor: FeedbackRequestProcessor,
fs_processor: Option<FsRequestProcessor>, fs_processor: FsRequestProcessor,
git_processor: GitRequestProcessor, git_processor: GitRequestProcessor,
initialize_processor: InitializeRequestProcessor, initialize_processor: InitializeRequestProcessor,
marketplace_processor: MarketplaceRequestProcessor, marketplace_processor: MarketplaceRequestProcessor,
@@ -274,23 +272,6 @@ pub(crate) struct MessageProcessorArgs {
} }
impl MessageProcessor { impl MessageProcessor {
fn fs_processor(&self) -> Result<&FsRequestProcessor, JSONRPCErrorError> {
self.fs_processor
.as_ref()
.ok_or_else(|| internal_error("local filesystem is not configured"))
}
fn require_local_environment(&self) -> Result<(), JSONRPCErrorError> {
// CCA filters these local-only RPCs before they reach app-server, but
// keep a Codex-side backstop so no-local app-server modes fail safely
// if a client still invokes one directly.
self.environment_manager
.try_local_environment()
.is_some()
.then_some(())
.ok_or_else(|| internal_error("local environment is not configured"))
}
/// Create a new `MessageProcessor`, retaining a handle to the outgoing /// Create a new `MessageProcessor`, retaining a handle to the outgoing
/// `Sender` so handlers can enqueue messages to be written to stdout. /// `Sender` so handlers can enqueue messages to be written to stdout.
pub(crate) fn new(args: MessageProcessorArgs) -> Self { pub(crate) fn new(args: MessageProcessorArgs) -> Self {
@@ -375,8 +356,12 @@ impl MessageProcessor {
Arc::clone(&config), Arc::clone(&config),
outgoing.clone(), outgoing.clone(),
config_manager.clone(), config_manager.clone(),
Arc::clone(&environment_manager_for_requests),
);
let process_exec_processor = ProcessExecRequestProcessor::new(
outgoing.clone(),
Arc::clone(&environment_manager_for_requests),
); );
let process_exec_processor = ProcessExecRequestProcessor::new(outgoing.clone());
let feedback_processor = FeedbackRequestProcessor::new( let feedback_processor = FeedbackRequestProcessor::new(
auth_manager.clone(), auth_manager.clone(),
Arc::clone(&thread_manager), Arc::clone(&thread_manager),
@@ -480,17 +465,10 @@ impl MessageProcessor {
); );
let environment_processor = let environment_processor =
EnvironmentRequestProcessor::new(thread_manager.environment_manager()); EnvironmentRequestProcessor::new(thread_manager.environment_manager());
// `fs/*` is a local-host filesystem surface. Do not construct it when let fs_processor = FsRequestProcessor::new(
// the manager intentionally has no local environment. Arc::clone(&environment_manager_for_requests),
let fs_processor = thread_manager FsWatchManager::new(outgoing.clone()),
.environment_manager() );
.try_local_environment()
.map(|environment| {
FsRequestProcessor::new(
environment.get_filesystem(),
FsWatchManager::new(outgoing.clone()),
)
});
let windows_sandbox_processor = WindowsSandboxRequestProcessor::new( let windows_sandbox_processor = WindowsSandboxRequestProcessor::new(
outgoing.clone(), outgoing.clone(),
Arc::clone(&config), Arc::clone(&config),
@@ -505,7 +483,6 @@ impl MessageProcessor {
command_exec_processor, command_exec_processor,
process_exec_processor, process_exec_processor,
config_processor, config_processor,
environment_manager: environment_manager_for_requests,
environment_processor, environment_processor,
external_agent_config_processor, external_agent_config_processor,
feedback_processor, feedback_processor,
@@ -729,9 +706,7 @@ impl MessageProcessor {
) { ) {
session_state.rpc_gate.shutdown().await; session_state.rpc_gate.shutdown().await;
self.outgoing.connection_closed(connection_id).await; self.outgoing.connection_closed(connection_id).await;
if let Some(fs_processor) = &self.fs_processor { self.fs_processor.connection_closed(connection_id).await;
fs_processor.connection_closed(connection_id).await;
}
self.command_exec_processor self.command_exec_processor
.connection_closed(connection_id) .connection_closed(connection_id)
.await; .await;
@@ -937,47 +912,47 @@ impl MessageProcessor {
self.environment_processor.environment_add(params).await self.environment_processor.environment_add(params).await
} }
ClientRequest::FsReadFile { params, .. } => self ClientRequest::FsReadFile { params, .. } => self
.fs_processor()? .fs_processor
.read_file(params) .read_file(params)
.await .await
.map(|response| Some(response.into())), .map(|response| Some(response.into())),
ClientRequest::FsWriteFile { params, .. } => self ClientRequest::FsWriteFile { params, .. } => self
.fs_processor()? .fs_processor
.write_file(params) .write_file(params)
.await .await
.map(|response| Some(response.into())), .map(|response| Some(response.into())),
ClientRequest::FsCreateDirectory { params, .. } => self ClientRequest::FsCreateDirectory { params, .. } => self
.fs_processor()? .fs_processor
.create_directory(params) .create_directory(params)
.await .await
.map(|response| Some(response.into())), .map(|response| Some(response.into())),
ClientRequest::FsGetMetadata { params, .. } => self ClientRequest::FsGetMetadata { params, .. } => self
.fs_processor()? .fs_processor
.get_metadata(params) .get_metadata(params)
.await .await
.map(|response| Some(response.into())), .map(|response| Some(response.into())),
ClientRequest::FsReadDirectory { params, .. } => self ClientRequest::FsReadDirectory { params, .. } => self
.fs_processor()? .fs_processor
.read_directory(params) .read_directory(params)
.await .await
.map(|response| Some(response.into())), .map(|response| Some(response.into())),
ClientRequest::FsRemove { params, .. } => self ClientRequest::FsRemove { params, .. } => self
.fs_processor()? .fs_processor
.remove(params) .remove(params)
.await .await
.map(|response| Some(response.into())), .map(|response| Some(response.into())),
ClientRequest::FsCopy { params, .. } => self ClientRequest::FsCopy { params, .. } => self
.fs_processor()? .fs_processor
.copy(params) .copy(params)
.await .await
.map(|response| Some(response.into())), .map(|response| Some(response.into())),
ClientRequest::FsWatch { params, .. } => self ClientRequest::FsWatch { params, .. } => self
.fs_processor()? .fs_processor
.watch(connection_id, params) .watch(connection_id, params)
.await .await
.map(|response| Some(response.into())), .map(|response| Some(response.into())),
ClientRequest::FsUnwatch { params, .. } => self ClientRequest::FsUnwatch { params, .. } => self
.fs_processor()? .fs_processor
.unwatch(connection_id, params) .unwatch(connection_id, params)
.await .await
.map(|response| Some(response.into())), .map(|response| Some(response.into())),
@@ -1306,7 +1281,6 @@ impl MessageProcessor {
.await .await
.map(|response| Some(response.into())), .map(|response| Some(response.into())),
ClientRequest::OneOffCommandExec { params, .. } => { ClientRequest::OneOffCommandExec { params, .. } => {
self.require_local_environment()?;
self.command_exec_processor self.command_exec_processor
.one_off_command_exec(&request_id, params) .one_off_command_exec(&request_id, params)
.await .await
@@ -1326,13 +1300,11 @@ impl MessageProcessor {
.command_exec_terminate(request_id.clone(), params) .command_exec_terminate(request_id.clone(), params)
.await .await
} }
ClientRequest::ProcessSpawn { params, .. } => { ClientRequest::ProcessSpawn { params, .. } => self
self.require_local_environment()?; .process_exec_processor
self.process_exec_processor .process_spawn(request_id.clone(), params)
.process_spawn(request_id.clone(), params) .await
.await .map(|()| None),
.map(|()| None)
}
ClientRequest::ProcessWriteStdin { params, .. } => { ClientRequest::ProcessWriteStdin { params, .. } => {
self.process_exec_processor self.process_exec_processor
.process_write_stdin(request_id.clone(), params) .process_write_stdin(request_id.clone(), params)
@@ -6,6 +6,7 @@ pub(crate) struct CommandExecRequestProcessor {
config: Arc<Config>, config: Arc<Config>,
outgoing: Arc<OutgoingMessageSender>, outgoing: Arc<OutgoingMessageSender>,
config_manager: ConfigManager, config_manager: ConfigManager,
environment_manager: Arc<EnvironmentManager>,
command_exec_manager: CommandExecManager, command_exec_manager: CommandExecManager,
} }
@@ -15,12 +16,14 @@ impl CommandExecRequestProcessor {
config: Arc<Config>, config: Arc<Config>,
outgoing: Arc<OutgoingMessageSender>, outgoing: Arc<OutgoingMessageSender>,
config_manager: ConfigManager, config_manager: ConfigManager,
environment_manager: Arc<EnvironmentManager>,
) -> Self { ) -> Self {
Self { Self {
arg0_paths, arg0_paths,
config, config,
outgoing, outgoing,
config_manager, config_manager,
environment_manager,
command_exec_manager: CommandExecManager::default(), command_exec_manager: CommandExecManager::default(),
} }
} }
@@ -30,6 +33,7 @@ impl CommandExecRequestProcessor {
request_id: &ConnectionRequestId, request_id: &ConnectionRequestId,
params: CommandExecParams, params: CommandExecParams,
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
self.require_local_environment()?;
self.exec_one_off_command(request_id, params) self.exec_one_off_command(request_id, params)
.await .await
.map(|()| None) .map(|()| None)
@@ -74,6 +78,14 @@ impl CommandExecRequestProcessor {
.await; .await;
} }
fn require_local_environment(&self) -> Result<(), JSONRPCErrorError> {
self.environment_manager
.try_local_environment()
.is_some()
.then_some(())
.ok_or_else(|| internal_error("local environment is not configured"))
}
async fn exec_one_off_command( async fn exec_one_off_command(
&self, &self,
request_id: &ConnectionRequestId, request_id: &ConnectionRequestId,
@@ -26,6 +26,7 @@ use codex_app_server_protocol::FsWriteFileResponse;
use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::JSONRPCErrorError;
use codex_exec_server::CopyOptions; use codex_exec_server::CopyOptions;
use codex_exec_server::CreateDirectoryOptions; use codex_exec_server::CreateDirectoryOptions;
use codex_exec_server::EnvironmentManager;
use codex_exec_server::ExecutorFileSystem; use codex_exec_server::ExecutorFileSystem;
use codex_exec_server::RemoveOptions; use codex_exec_server::RemoveOptions;
use std::io; use std::io;
@@ -33,21 +34,28 @@ use std::sync::Arc;
#[derive(Clone)] #[derive(Clone)]
pub(crate) struct FsRequestProcessor { pub(crate) struct FsRequestProcessor {
file_system: Arc<dyn ExecutorFileSystem>, environment_manager: Arc<EnvironmentManager>,
fs_watch_manager: FsWatchManager, fs_watch_manager: FsWatchManager,
} }
impl FsRequestProcessor { impl FsRequestProcessor {
pub(crate) fn new( pub(crate) fn new(
file_system: Arc<dyn ExecutorFileSystem>, environment_manager: Arc<EnvironmentManager>,
fs_watch_manager: FsWatchManager, fs_watch_manager: FsWatchManager,
) -> Self { ) -> Self {
Self { Self {
file_system, environment_manager,
fs_watch_manager, fs_watch_manager,
} }
} }
fn file_system(&self) -> Result<Arc<dyn ExecutorFileSystem>, JSONRPCErrorError> {
self.environment_manager
.try_local_environment()
.map(|environment| environment.get_filesystem())
.ok_or_else(|| internal_error("local filesystem is not configured"))
}
pub(crate) async fn connection_closed(&self, connection_id: ConnectionId) { pub(crate) async fn connection_closed(&self, connection_id: ConnectionId) {
self.fs_watch_manager.connection_closed(connection_id).await; self.fs_watch_manager.connection_closed(connection_id).await;
} }
@@ -57,7 +65,7 @@ impl FsRequestProcessor {
params: FsReadFileParams, params: FsReadFileParams,
) -> Result<FsReadFileResponse, JSONRPCErrorError> { ) -> Result<FsReadFileResponse, JSONRPCErrorError> {
let bytes = self let bytes = self
.file_system .file_system()?
.read_file(&params.path, /*sandbox*/ None) .read_file(&params.path, /*sandbox*/ None)
.await .await
.map_err(map_fs_error)?; .map_err(map_fs_error)?;
@@ -75,7 +83,7 @@ impl FsRequestProcessor {
"fs/writeFile requires valid base64 dataBase64: {err}" "fs/writeFile requires valid base64 dataBase64: {err}"
)) ))
})?; })?;
self.file_system self.file_system()?
.write_file(&params.path, bytes, /*sandbox*/ None) .write_file(&params.path, bytes, /*sandbox*/ None)
.await .await
.map_err(map_fs_error)?; .map_err(map_fs_error)?;
@@ -86,7 +94,7 @@ impl FsRequestProcessor {
&self, &self,
params: FsCreateDirectoryParams, params: FsCreateDirectoryParams,
) -> Result<FsCreateDirectoryResponse, JSONRPCErrorError> { ) -> Result<FsCreateDirectoryResponse, JSONRPCErrorError> {
self.file_system self.file_system()?
.create_directory( .create_directory(
&params.path, &params.path,
CreateDirectoryOptions { CreateDirectoryOptions {
@@ -104,7 +112,7 @@ impl FsRequestProcessor {
params: FsGetMetadataParams, params: FsGetMetadataParams,
) -> Result<FsGetMetadataResponse, JSONRPCErrorError> { ) -> Result<FsGetMetadataResponse, JSONRPCErrorError> {
let metadata = self let metadata = self
.file_system .file_system()?
.get_metadata(&params.path, /*sandbox*/ None) .get_metadata(&params.path, /*sandbox*/ None)
.await .await
.map_err(map_fs_error)?; .map_err(map_fs_error)?;
@@ -122,7 +130,7 @@ impl FsRequestProcessor {
params: FsReadDirectoryParams, params: FsReadDirectoryParams,
) -> Result<FsReadDirectoryResponse, JSONRPCErrorError> { ) -> Result<FsReadDirectoryResponse, JSONRPCErrorError> {
let entries = self let entries = self
.file_system .file_system()?
.read_directory(&params.path, /*sandbox*/ None) .read_directory(&params.path, /*sandbox*/ None)
.await .await
.map_err(map_fs_error)?; .map_err(map_fs_error)?;
@@ -142,7 +150,7 @@ impl FsRequestProcessor {
&self, &self,
params: FsRemoveParams, params: FsRemoveParams,
) -> Result<FsRemoveResponse, JSONRPCErrorError> { ) -> Result<FsRemoveResponse, JSONRPCErrorError> {
self.file_system self.file_system()?
.remove( .remove(
&params.path, &params.path,
RemoveOptions { RemoveOptions {
@@ -160,7 +168,7 @@ impl FsRequestProcessor {
&self, &self,
params: FsCopyParams, params: FsCopyParams,
) -> Result<FsCopyResponse, JSONRPCErrorError> { ) -> Result<FsCopyResponse, JSONRPCErrorError> {
self.file_system self.file_system()?
.copy( .copy(
&params.source_path, &params.source_path,
&params.destination_path, &params.destination_path,
@@ -179,6 +187,7 @@ impl FsRequestProcessor {
connection_id: ConnectionId, connection_id: ConnectionId,
params: FsWatchParams, params: FsWatchParams,
) -> Result<FsWatchResponse, JSONRPCErrorError> { ) -> Result<FsWatchResponse, JSONRPCErrorError> {
self.file_system()?;
self.fs_watch_manager.watch(connection_id, params).await self.fs_watch_manager.watch(connection_id, params).await
} }
@@ -187,6 +196,7 @@ impl FsRequestProcessor {
connection_id: ConnectionId, connection_id: ConnectionId,
params: FsUnwatchParams, params: FsUnwatchParams,
) -> Result<FsUnwatchResponse, JSONRPCErrorError> { ) -> Result<FsUnwatchResponse, JSONRPCErrorError> {
self.file_system()?;
self.fs_watch_manager.unwatch(connection_id, params).await self.fs_watch_manager.unwatch(connection_id, params).await
} }
} }
@@ -23,6 +23,7 @@ use codex_app_server_protocol::ServerNotification;
use codex_core::exec::ExecExpiration; use codex_core::exec::ExecExpiration;
use codex_core::exec::ExecExpirationOutcome; use codex_core::exec::ExecExpirationOutcome;
use codex_core::exec::IO_DRAIN_TIMEOUT_MS; use codex_core::exec::IO_DRAIN_TIMEOUT_MS;
use codex_exec_server::EnvironmentManager;
use codex_protocol::exec_output::bytes_to_string_smart; use codex_protocol::exec_output::bytes_to_string_smart;
use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_pty::DEFAULT_OUTPUT_BYTES_CAP; use codex_utils_pty::DEFAULT_OUTPUT_BYTES_CAP;
@@ -48,13 +49,18 @@ const OUTPUT_CHUNK_SIZE_HINT: usize = 64 * 1024;
#[derive(Clone)] #[derive(Clone)]
pub(crate) struct ProcessExecRequestProcessor { pub(crate) struct ProcessExecRequestProcessor {
outgoing: Arc<OutgoingMessageSender>, outgoing: Arc<OutgoingMessageSender>,
environment_manager: Arc<EnvironmentManager>,
process_exec_manager: ProcessExecManager, process_exec_manager: ProcessExecManager,
} }
impl ProcessExecRequestProcessor { impl ProcessExecRequestProcessor {
pub(crate) fn new(outgoing: Arc<OutgoingMessageSender>) -> Self { pub(crate) fn new(
outgoing: Arc<OutgoingMessageSender>,
environment_manager: Arc<EnvironmentManager>,
) -> Self {
Self { Self {
outgoing, outgoing,
environment_manager,
process_exec_manager: ProcessExecManager::default(), process_exec_manager: ProcessExecManager::default(),
} }
} }
@@ -64,6 +70,7 @@ impl ProcessExecRequestProcessor {
request_id: ConnectionRequestId, request_id: ConnectionRequestId,
params: ProcessSpawnParams, params: ProcessSpawnParams,
) -> Result<(), JSONRPCErrorError> { ) -> Result<(), JSONRPCErrorError> {
self.require_local_environment()?;
let ProcessSpawnParams { let ProcessSpawnParams {
command, command,
process_handle, process_handle,
@@ -173,6 +180,14 @@ impl ProcessExecRequestProcessor {
.connection_closed(connection_id) .connection_closed(connection_id)
.await; .await;
} }
fn require_local_environment(&self) -> Result<(), JSONRPCErrorError> {
self.environment_manager
.try_local_environment()
.is_some()
.then_some(())
.ok_or_else(|| internal_error("local environment is not configured"))
}
} }
#[derive(Clone, Default)] #[derive(Clone, Default)]