From 1b65ab838f975475feb05e159d0551ffcba725ad Mon Sep 17 00:00:00 2001 From: colby-oai <228809017+colby-oai@users.noreply.github.com> Date: Tue, 28 Apr 2026 14:21:56 -0400 Subject: [PATCH] use fs abstraction --- codex-rs/Cargo.lock | 2 + codex-rs/codex-api/src/files.rs | 71 +++++++++++--- codex-rs/codex-api/src/lib.rs | 1 + codex-rs/core/src/mcp_openai_file.rs | 69 +++++++++++--- codex-rs/exec-server/Cargo.toml | 2 +- codex-rs/exec-server/src/fs_helper.rs | 72 ++++++++++++++ codex-rs/exec-server/src/fs_helper_main.rs | 7 ++ codex-rs/exec-server/src/fs_sandbox.rs | 93 +++++++++++++++++++ codex-rs/exec-server/src/lib.rs | 1 + codex-rs/exec-server/src/local_file_system.rs | 60 +++++++++++- .../exec-server/src/sandboxed_file_system.rs | 37 ++++++++ codex-rs/exec-server/tests/file_system.rs | 17 +++- codex-rs/file-system/Cargo.toml | 2 + codex-rs/file-system/src/lib.rs | 30 ++++++ 14 files changed, 438 insertions(+), 26 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index e3f2ac1a1f..eaee6ed294 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2729,8 +2729,10 @@ name = "codex-file-system" version = "0.0.0" dependencies = [ "async-trait", + "bytes", "codex-protocol", "codex-utils-absolute-path", + "futures", "serde", ] diff --git a/codex-rs/codex-api/src/files.rs b/codex-rs/codex-api/src/files.rs index d1e2840066..58b6779c59 100644 --- a/codex-rs/codex-api/src/files.rs +++ b/codex-rs/codex-api/src/files.rs @@ -27,7 +27,7 @@ pub struct UploadedOpenAiFile { pub file_name: String, pub file_size_bytes: u64, pub mime_type: Option, - pub path: PathBuf, + pub path: Option, } #[derive(Debug, thiserror::Error)] @@ -128,11 +128,64 @@ pub async fn upload_local_file( .and_then(|value| value.to_str()) .unwrap_or("file") .to_string(); + let upload_file = File::open(path) + .await + .map_err(|source| OpenAiFileError::ReadFile { + path: path.to_path_buf(), + source, + })?; + upload_file_body_with_source_path( + base_url, + auth, + file_name, + metadata.len(), + reqwest::Body::wrap_stream(ReaderStream::new(upload_file)), + Some(path.to_path_buf()), + ) + .await +} + +pub async fn upload_file_body( + base_url: &str, + auth: &dyn AuthProvider, + file_name: String, + file_size_bytes: u64, + body: reqwest::Body, +) -> Result { + upload_file_body_with_source_path( + base_url, + auth, + file_name, + file_size_bytes, + body, + /*source_path*/ None, + ) + .await +} + +async fn upload_file_body_with_source_path( + base_url: &str, + auth: &dyn AuthProvider, + file_name: String, + file_size_bytes: u64, + body: reqwest::Body, + source_path: Option, +) -> Result { + if file_size_bytes > OPENAI_FILE_UPLOAD_LIMIT_BYTES { + return Err(OpenAiFileError::FileTooLarge { + path: source_path + .clone() + .unwrap_or_else(|| PathBuf::from(file_name.clone())), + size_bytes: file_size_bytes, + limit_bytes: OPENAI_FILE_UPLOAD_LIMIT_BYTES, + }); + } + let create_url = format!("{}/files", base_url.trim_end_matches('/')); let create_response = authorized_request(auth, reqwest::Method::POST, &create_url) .json(&serde_json::json!({ "file_name": file_name, - "file_size": metadata.len(), + "file_size": file_size_bytes, "use_case": OPENAI_FILE_USE_CASE, })) .send() @@ -156,18 +209,12 @@ pub async fn upload_local_file( source, })?; - let upload_file = File::open(path) - .await - .map_err(|source| OpenAiFileError::ReadFile { - path: path.to_path_buf(), - source, - })?; let upload_response = build_reqwest_client() .put(&create_payload.upload_url) .timeout(OPENAI_FILE_REQUEST_TIMEOUT) .header("x-ms-blob-type", "BlockBlob") - .header(CONTENT_LENGTH, metadata.len()) - .body(reqwest::Body::wrap_stream(ReaderStream::new(upload_file))) + .header(CONTENT_LENGTH, file_size_bytes) + .body(body) .send() .await .map_err(|source| OpenAiFileError::Request { @@ -226,9 +273,9 @@ pub async fn upload_local_file( } })?, file_name: finalize_payload.file_name.unwrap_or(file_name), - file_size_bytes: metadata.len(), + file_size_bytes, mime_type: finalize_payload.mime_type, - path: path.to_path_buf(), + path: source_path, }); } "retry" => { diff --git a/codex-rs/codex-api/src/lib.rs b/codex-rs/codex-api/src/lib.rs index 0b8aee266b..df8301ecd7 100644 --- a/codex-rs/codex-api/src/lib.rs +++ b/codex-rs/codex-api/src/lib.rs @@ -57,6 +57,7 @@ pub use crate::endpoint::ResponsesWebsocketClient; pub use crate::endpoint::ResponsesWebsocketConnection; pub use crate::endpoint::session_update_session_json; pub use crate::error::ApiError; +pub use crate::files::upload_file_body; pub use crate::files::upload_local_file; pub use crate::provider::Provider; pub use crate::provider::RetryConfig; diff --git a/codex-rs/core/src/mcp_openai_file.rs b/codex-rs/core/src/mcp_openai_file.rs index 8a7c1c7bda..736646379a 100644 --- a/codex-rs/core/src/mcp_openai_file.rs +++ b/codex-rs/core/src/mcp_openai_file.rs @@ -12,7 +12,7 @@ use crate::session::session::Session; use crate::session::turn_context::TurnContext; -use codex_api::upload_local_file; +use codex_api::upload_file_body; use codex_login::CodexAuth; use serde_json::Value as JsonValue; @@ -39,9 +39,14 @@ pub(crate) async fn rewrite_mcp_tool_arguments_for_openai_files( let Some(value) = arguments.get(field_name) else { continue; }; - let Some(uploaded_value) = - rewrite_argument_value_for_openai_files(turn_context, auth.as_ref(), field_name, value) - .await? + let Some(uploaded_value) = rewrite_argument_value_for_openai_files( + sess, + turn_context, + auth.as_ref(), + field_name, + value, + ) + .await? else { continue; }; @@ -56,6 +61,7 @@ pub(crate) async fn rewrite_mcp_tool_arguments_for_openai_files( } async fn rewrite_argument_value_for_openai_files( + sess: &Session, turn_context: &TurnContext, auth: Option<&CodexAuth>, field_name: &str, @@ -64,6 +70,7 @@ async fn rewrite_argument_value_for_openai_files( match value { JsonValue::String(path_or_file_ref) => { let rewritten = build_uploaded_local_argument_value( + sess, turn_context, auth, field_name, @@ -80,6 +87,7 @@ async fn rewrite_argument_value_for_openai_files( return Ok(None); }; let rewritten = build_uploaded_local_argument_value( + sess, turn_context, auth, field_name, @@ -96,6 +104,7 @@ async fn rewrite_argument_value_for_openai_files( } async fn build_uploaded_local_argument_value( + sess: &Session, turn_context: &TurnContext, auth: Option<&CodexAuth>, field_name: &str, @@ -126,11 +135,33 @@ async fn build_uploaded_local_argument_value( "ChatGPT auth is required to upload local files for Codex Apps tools".to_string(), ); } + let sandbox = turn_context.file_system_sandbox_context(/*additional_permissions*/ None); + let file_system = turn_context + .environment + .as_ref() + .map(|environment| environment.get_filesystem()) + .unwrap_or_else(|| { + sess.services + .environment_manager + .local_environment() + .get_filesystem() + }); + let file_body = file_system + .read_file_body(&resolved_path, Some(&sandbox)) + .await + .map_err(|error| match index { + Some(index) => { + format!("failed to read `{file_path}` for `{field_name}[{index}]`: {error}") + } + None => format!("failed to read `{file_path}` for `{field_name}`: {error}"), + })?; let upload_auth = codex_model_provider::auth_provider_from_auth(auth); - let uploaded = upload_local_file( + let uploaded = upload_file_body( turn_context.config.chatgpt_base_url.trim_end_matches('/'), upload_auth.as_ref(), - &resolved_path, + file_body.file_name, + file_body.file_size_bytes, + reqwest::Body::wrap_stream(file_body.stream), ) .await .map_err(|error| match index { @@ -229,7 +260,7 @@ mod tests { .mount(&server) .await; - let (_, mut turn_context) = make_session_and_context().await; + let (session, mut turn_context) = make_session_and_context().await; let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing(); let dir = tempdir().expect("temp dir"); let local_path = dir.path().join("file_report.csv"); @@ -241,8 +272,13 @@ mod tests { let mut config = (*turn_context.config).clone(); config.chatgpt_base_url = format!("{}/backend-api", server.uri()); turn_context.config = Arc::new(config); + turn_context.permission_profile = PermissionProfile::from_runtime_permissions( + &FileSystemSandboxPolicy::unrestricted(), + NetworkSandboxPolicy::Enabled, + ); let rewritten = build_uploaded_local_argument_value( + &session, &turn_context, Some(&auth), "file", @@ -267,7 +303,7 @@ mod tests { #[tokio::test] async fn build_uploaded_local_argument_value_respects_filesystem_read_policy() { - let (_, mut turn_context) = make_session_and_context().await; + let (session, mut turn_context) = make_session_and_context().await; let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing(); let dir = tempdir().expect("temp dir"); let local_path = dir.path().join("blocked.csv"); @@ -297,6 +333,7 @@ mod tests { ); let error = build_uploaded_local_argument_value( + &session, &turn_context, Some(&auth), "file", @@ -357,7 +394,7 @@ mod tests { .mount(&server) .await; - let (_, mut turn_context) = make_session_and_context().await; + let (session, mut turn_context) = make_session_and_context().await; let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing(); let dir = tempdir().expect("temp dir"); let local_path = dir.path().join("file_report.csv"); @@ -369,7 +406,12 @@ mod tests { let mut config = (*turn_context.config).clone(); config.chatgpt_base_url = format!("{}/backend-api", server.uri()); turn_context.config = Arc::new(config); + turn_context.permission_profile = PermissionProfile::from_runtime_permissions( + &FileSystemSandboxPolicy::unrestricted(), + NetworkSandboxPolicy::Enabled, + ); let rewritten = rewrite_argument_value_for_openai_files( + &session, &turn_context, Some(&auth), "file", @@ -469,7 +511,7 @@ mod tests { .mount(&server) .await; - let (_, mut turn_context) = make_session_and_context().await; + let (session, mut turn_context) = make_session_and_context().await; let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing(); let dir = tempdir().expect("temp dir"); tokio::fs::write(dir.path().join("one.csv"), b"one") @@ -483,7 +525,12 @@ mod tests { let mut config = (*turn_context.config).clone(); config.chatgpt_base_url = format!("{}/backend-api", server.uri()); turn_context.config = Arc::new(config); + turn_context.permission_profile = PermissionProfile::from_runtime_permissions( + &FileSystemSandboxPolicy::unrestricted(), + NetworkSandboxPolicy::Enabled, + ); let rewritten = rewrite_argument_value_for_openai_files( + &session, &turn_context, Some(&auth), "files", @@ -532,7 +579,7 @@ mod tests { .await .expect_err("missing file should fail"); - assert!(error.contains("failed to upload")); + assert!(error.contains("failed to read")); assert!(error.contains("file")); } } diff --git a/codex-rs/exec-server/Cargo.toml b/codex-rs/exec-server/Cargo.toml index 5f31ca4329..4dc77f8552 100644 --- a/codex-rs/exec-server/Cargo.toml +++ b/codex-rs/exec-server/Cargo.toml @@ -38,7 +38,7 @@ tokio = { workspace = true, features = [ "sync", "time", ] } -tokio-util = { workspace = true, features = ["rt"] } +tokio-util = { workspace = true, features = ["io", "rt"] } tokio-tungstenite = { workspace = true } tracing = { workspace = true } uuid = { workspace = true, features = ["v4"] } diff --git a/codex-rs/exec-server/src/fs_helper.rs b/codex-rs/exec-server/src/fs_helper.rs index 8d91224803..d663557f8f 100644 --- a/codex-rs/exec-server/src/fs_helper.rs +++ b/codex-rs/exec-server/src/fs_helper.rs @@ -4,12 +4,14 @@ use codex_app_server_protocol::JSONRPCErrorError; use serde::Deserialize; use serde::Serialize; use tokio::io; +use tokio::io::AsyncWrite; use crate::CopyOptions; use crate::CreateDirectoryOptions; use crate::ExecutorFileSystem; use crate::RemoveOptions; use crate::local_file_system::DirectFileSystem; +use crate::local_file_system::MAX_READ_FILE_BYTES; use crate::protocol::FS_COPY_METHOD; use crate::protocol::FS_CREATE_DIRECTORY_METHOD; use crate::protocol::FS_GET_METADATA_METHOD; @@ -43,6 +45,10 @@ pub const CODEX_FS_HELPER_ARG1: &str = "--codex-run-as-fs-helper"; pub(crate) enum FsHelperRequest { #[serde(rename = "fs/readFile")] ReadFile(FsReadFileParams), + #[serde(rename = "fs/readFileInfo")] + ReadFileInfo(FsReadFileParams), + #[serde(rename = "fs/readFileStream")] + ReadFileStream(FsReadFileParams), #[serde(rename = "fs/writeFile")] WriteFile(FsWriteFileParams), #[serde(rename = "fs/createDirectory")] @@ -69,6 +75,8 @@ pub(crate) enum FsHelperResponse { pub(crate) enum FsHelperPayload { #[serde(rename = "fs/readFile")] ReadFile(FsReadFileResponse), + #[serde(rename = "fs/readFileInfo")] + ReadFileInfo(FsReadFileInfoResponse), #[serde(rename = "fs/writeFile")] WriteFile(FsWriteFileResponse), #[serde(rename = "fs/createDirectory")] @@ -83,10 +91,17 @@ pub(crate) enum FsHelperPayload { Copy(FsCopyResponse), } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct FsReadFileInfoResponse { + pub file_size_bytes: u64, +} + impl FsHelperPayload { fn operation(&self) -> &'static str { match self { Self::ReadFile(_) => FS_READ_FILE_METHOD, + Self::ReadFileInfo(_) => "fs/readFileInfo", Self::WriteFile(_) => FS_WRITE_FILE_METHOD, Self::CreateDirectory(_) => FS_CREATE_DIRECTORY_METHOD, Self::GetMetadata(_) => FS_GET_METADATA_METHOD, @@ -103,6 +118,13 @@ impl FsHelperPayload { } } + pub(crate) fn expect_read_file_info(self) -> Result { + match self { + Self::ReadFileInfo(response) => Ok(response), + other => Err(unexpected_response("fs/readFileInfo", other.operation())), + } + } + pub(crate) fn expect_write_file(self) -> Result { match self { Self::WriteFile(response) => Ok(response), @@ -179,6 +201,18 @@ pub(crate) async fn run_direct_request( data_base64: STANDARD.encode(data), })) } + FsHelperRequest::ReadFileInfo(params) => { + let metadata = tokio::fs::metadata(params.path.as_path()) + .await + .map_err(map_fs_error)?; + validate_read_file_metadata(params.path.as_path(), &metadata).map_err(map_fs_error)?; + Ok(FsHelperPayload::ReadFileInfo(FsReadFileInfoResponse { + file_size_bytes: metadata.len(), + })) + } + FsHelperRequest::ReadFileStream(_) => Err(invalid_request( + "fs/readFileStream is only supported by the streaming helper path".to_string(), + )), FsHelperRequest::WriteFile(params) => { let bytes = STANDARD.decode(params.data_base64).map_err(|err| { invalid_request(format!( @@ -266,6 +300,44 @@ pub(crate) async fn run_direct_request( } } +pub(crate) async fn run_direct_stream_request( + request: FsHelperRequest, + stdout: &mut (impl AsyncWrite + Unpin), +) -> Result<(), Box> { + match request { + FsHelperRequest::ReadFileStream(params) => { + let metadata = tokio::fs::metadata(params.path.as_path()).await?; + validate_read_file_metadata(params.path.as_path(), &metadata)?; + let mut file = tokio::fs::File::open(params.path.as_path()).await?; + tokio::io::copy(&mut file, stdout).await?; + Ok(()) + } + _ => Err(Box::new(io::Error::new( + io::ErrorKind::InvalidInput, + "streaming helper received a non-streaming request", + ))), + } +} + +fn validate_read_file_metadata( + path: &std::path::Path, + metadata: &std::fs::Metadata, +) -> io::Result<()> { + if !metadata.is_file() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("path `{}` is not a file", path.display()), + )); + } + if metadata.len() > MAX_READ_FILE_BYTES { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("file is too large to read: limit is {MAX_READ_FILE_BYTES} bytes"), + )); + } + Ok(()) +} + fn map_fs_error(err: io::Error) -> JSONRPCErrorError { match err.kind() { io::ErrorKind::NotFound => not_found(err.to_string()), diff --git a/codex-rs/exec-server/src/fs_helper_main.rs b/codex-rs/exec-server/src/fs_helper_main.rs index d9639dd1f8..25253be197 100644 --- a/codex-rs/exec-server/src/fs_helper_main.rs +++ b/codex-rs/exec-server/src/fs_helper_main.rs @@ -7,6 +7,7 @@ use tokio::io::AsyncWriteExt; use crate::fs_helper::FsHelperRequest; use crate::fs_helper::FsHelperResponse; use crate::fs_helper::run_direct_request; +use crate::fs_helper::run_direct_stream_request; pub fn main() -> ! { let exit_code = match tokio::runtime::Builder::new_current_thread() @@ -32,6 +33,12 @@ async fn run_main() -> Result<(), Box> { let mut input = Vec::new(); io::stdin().read_to_end(&mut input).await?; let request: FsHelperRequest = serde_json::from_slice(&input)?; + if matches!(request, FsHelperRequest::ReadFileStream(_)) { + let mut stdout = io::stdout(); + run_direct_stream_request(request, &mut stdout).await?; + stdout.shutdown().await?; + return Ok(()); + } let response = match run_direct_request(request).await { Ok(payload) => FsHelperResponse::Ok(payload), Err(error) => FsHelperResponse::Error(error), diff --git a/codex-rs/exec-server/src/fs_sandbox.rs b/codex-rs/exec-server/src/fs_sandbox.rs index 8f084a50e9..a1c46f29e7 100644 --- a/codex-rs/exec-server/src/fs_sandbox.rs +++ b/codex-rs/exec-server/src/fs_sandbox.rs @@ -1,6 +1,7 @@ use std::collections::HashMap; use codex_app_server_protocol::JSONRPCErrorError; +use codex_file_system::FileByteStream; use codex_protocol::models::PermissionProfile; use codex_protocol::permissions::FileSystemAccessMode; use codex_protocol::permissions::FileSystemPath; @@ -17,6 +18,7 @@ use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_absolute_path::canonicalize_preserving_symlinks; use tokio::io::AsyncWriteExt; use tokio::process::Command; +use tokio_util::io::ReaderStream; use crate::ExecServerRuntimePaths; use crate::FileSystemSandboxContext; @@ -69,6 +71,31 @@ impl FileSystemSandboxRunner { run_command(command, request_json).await } + pub(crate) async fn run_stream( + &self, + sandbox: &FileSystemSandboxContext, + request: FsHelperRequest, + ) -> Result { + let cwd = sandbox_cwd(sandbox)?; + let mut file_system_policy = sandbox.permissions.file_system_sandbox_policy(); + let helper_read_roots = if sandbox.use_legacy_landlock { + Vec::new() + } else { + helper_read_roots(&self.runtime_paths) + }; + add_helper_runtime_permissions(&mut file_system_policy, &helper_read_roots, cwd.as_path()); + normalize_file_system_policy_root_aliases(&mut file_system_policy); + let network_policy = NetworkSandboxPolicy::Restricted; + let permission_profile = PermissionProfile::from_runtime_permissions_with_enforcement( + sandbox.permissions.enforcement(), + &file_system_policy, + network_policy, + ); + let command = self.sandbox_exec_request(&permission_profile, &cwd, sandbox)?; + let request_json = serde_json::to_vec(&request).map_err(json_error)?; + run_command_stream(command, request_json).await + } + fn sandbox_exec_request( &self, permission_profile: &PermissionProfile, @@ -251,6 +278,72 @@ async fn run_command( } } +struct StreamCommandState { + stdout: ReaderStream, + child: tokio::process::Child, + stderr_task: tokio::task::JoinHandle>, +} + +async fn run_command_stream( + command: SandboxExecRequest, + request_json: Vec, +) -> Result { + let mut child = spawn_command(command)?; + let mut stdin = child + .stdin + .take() + .ok_or_else(|| internal_error("failed to open fs sandbox helper stdin".to_string()))?; + stdin.write_all(&request_json).await.map_err(io_error)?; + stdin.shutdown().await.map_err(io_error)?; + drop(stdin); + + let stdout = child + .stdout + .take() + .ok_or_else(|| internal_error("failed to open fs sandbox helper stdout".to_string()))?; + let stderr = child.stderr.take(); + let stderr_task = tokio::spawn(async move { + let Some(mut stderr) = stderr else { + return Vec::new(); + }; + let mut output = Vec::new(); + let _ = tokio::io::AsyncReadExt::read_to_end(&mut stderr, &mut output).await; + output + }); + let state = StreamCommandState { + stdout: ReaderStream::new(stdout), + child, + stderr_task, + }; + + Ok(Box::pin(futures::stream::try_unfold( + state, + |mut state| async move { + match futures::StreamExt::next(&mut state.stdout).await { + Some(Ok(bytes)) => Ok(Some((bytes, state))), + Some(Err(error)) => { + let _ = state.child.start_kill(); + Err(error) + } + None => { + let status = state.child.wait().await?; + let stderr = state + .stderr_task + .await + .map_err(|error| std::io::Error::other(error.to_string()))?; + if !status.success() { + return Err(std::io::Error::other(format!( + "fs sandbox helper failed with status {status}: {stderr}", + stderr = String::from_utf8_lossy(&stderr).trim() + ))); + } + Ok(None) + } + } + }, + ))) +} + fn spawn_command( SandboxExecRequest { command: argv, diff --git a/codex-rs/exec-server/src/lib.rs b/codex-rs/exec-server/src/lib.rs index e739b05644..ca56271146 100644 --- a/codex-rs/exec-server/src/lib.rs +++ b/codex-rs/exec-server/src/lib.rs @@ -28,6 +28,7 @@ pub use codex_file_system::CopyOptions; pub use codex_file_system::CreateDirectoryOptions; pub use codex_file_system::ExecutorFileSystem; pub use codex_file_system::FileMetadata; +pub use codex_file_system::FileReadBody; pub use codex_file_system::FileSystemResult; pub use codex_file_system::FileSystemSandboxContext; pub use codex_file_system::ReadDirectoryEntry; diff --git a/codex-rs/exec-server/src/local_file_system.rs b/codex-rs/exec-server/src/local_file_system.rs index c7d628ce8f..520939c14d 100644 --- a/codex-rs/exec-server/src/local_file_system.rs +++ b/codex-rs/exec-server/src/local_file_system.rs @@ -1,5 +1,7 @@ use async_trait::async_trait; +use bytes::Bytes; use codex_utils_absolute_path::AbsolutePathBuf; +use futures::StreamExt; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; @@ -13,13 +15,15 @@ use crate::CreateDirectoryOptions; use crate::ExecServerRuntimePaths; use crate::ExecutorFileSystem; use crate::FileMetadata; +use crate::FileReadBody; use crate::FileSystemResult; use crate::FileSystemSandboxContext; use crate::ReadDirectoryEntry; use crate::RemoveOptions; use crate::sandboxed_file_system::SandboxedFileSystem; +use tokio_util::io::ReaderStream; -const MAX_READ_FILE_BYTES: u64 = 512 * 1024 * 1024; +pub(crate) const MAX_READ_FILE_BYTES: u64 = 512 * 1024 * 1024; pub static LOCAL_FS: LazyLock> = LazyLock::new(|| -> Arc { Arc::new(LocalFileSystem::unsandboxed()) }); @@ -88,6 +92,15 @@ impl ExecutorFileSystem for LocalFileSystem { file_system.read_file(path, sandbox).await } + async fn read_file_body( + &self, + path: &AbsolutePathBuf, + sandbox: Option<&FileSystemSandboxContext>, + ) -> FileSystemResult { + let (file_system, sandbox) = self.file_system_for(sandbox)?; + file_system.read_file_body(path, sandbox).await + } + async fn write_file( &self, path: &AbsolutePathBuf, @@ -161,6 +174,17 @@ impl ExecutorFileSystem for UnsandboxedFileSystem { self.file_system.read_file(path, /*sandbox*/ None).await } + async fn read_file_body( + &self, + path: &AbsolutePathBuf, + sandbox: Option<&FileSystemSandboxContext>, + ) -> FileSystemResult { + reject_platform_sandbox_context(sandbox)?; + self.file_system + .read_file_body(path, /*sandbox*/ None) + .await + } + async fn write_file( &self, path: &AbsolutePathBuf, @@ -254,6 +278,33 @@ impl ExecutorFileSystem for DirectFileSystem { tokio::fs::read(path.as_path()).await } + async fn read_file_body( + &self, + path: &AbsolutePathBuf, + sandbox: Option<&FileSystemSandboxContext>, + ) -> FileSystemResult { + reject_sandbox_context(sandbox)?; + let metadata = tokio::fs::metadata(path.as_path()).await?; + if !metadata.is_file() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("path `{}` is not a file", path.display()), + )); + } + if metadata.len() > MAX_READ_FILE_BYTES { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("file is too large to read: limit is {MAX_READ_FILE_BYTES} bytes"), + )); + } + let file = tokio::fs::File::open(path.as_path()).await?; + Ok(FileReadBody { + file_name: file_name_from_path(path.as_path()), + file_size_bytes: metadata.len(), + stream: Box::pin(ReaderStream::new(file).map(|chunk| chunk.map(Bytes::from))), + }) + } + async fn write_file( &self, path: &AbsolutePathBuf, @@ -417,6 +468,13 @@ fn reject_platform_sandbox_context(sandbox: Option<&FileSystemSandboxContext>) - Ok(()) } +pub(crate) fn file_name_from_path(path: &Path) -> String { + path.file_name() + .and_then(|value| value.to_str()) + .unwrap_or("file") + .to_string() +} + fn copy_dir_recursive(source: &Path, target: &Path) -> io::Result<()> { std::fs::create_dir_all(target)?; for entry in std::fs::read_dir(source)? { diff --git a/codex-rs/exec-server/src/sandboxed_file_system.rs b/codex-rs/exec-server/src/sandboxed_file_system.rs index 133a3b7327..a895584e95 100644 --- a/codex-rs/exec-server/src/sandboxed_file_system.rs +++ b/codex-rs/exec-server/src/sandboxed_file_system.rs @@ -10,6 +10,7 @@ use crate::CreateDirectoryOptions; use crate::ExecServerRuntimePaths; use crate::ExecutorFileSystem; use crate::FileMetadata; +use crate::FileReadBody; use crate::FileSystemResult; use crate::FileSystemSandboxContext; use crate::ReadDirectoryEntry; @@ -17,6 +18,7 @@ use crate::RemoveOptions; use crate::fs_helper::FsHelperPayload; use crate::fs_helper::FsHelperRequest; use crate::fs_sandbox::FileSystemSandboxRunner; +use crate::local_file_system::file_name_from_path; use crate::protocol::FsCopyParams; use crate::protocol::FsCreateDirectoryParams; use crate::protocol::FsGetMetadataParams; @@ -76,6 +78,41 @@ impl ExecutorFileSystem for SandboxedFileSystem { }) } + async fn read_file_body( + &self, + path: &AbsolutePathBuf, + sandbox: Option<&FileSystemSandboxContext>, + ) -> FileSystemResult { + let sandbox = require_platform_sandbox(sandbox)?; + let file_info = self + .run_sandboxed( + sandbox, + FsHelperRequest::ReadFileInfo(FsReadFileParams { + path: path.clone(), + sandbox: None, + }), + ) + .await? + .expect_read_file_info() + .map_err(map_sandbox_error)?; + let stream = self + .sandbox_runner + .run_stream( + sandbox, + FsHelperRequest::ReadFileStream(FsReadFileParams { + path: path.clone(), + sandbox: None, + }), + ) + .await + .map_err(map_sandbox_error)?; + Ok(FileReadBody { + file_name: file_name_from_path(path.as_path()), + file_size_bytes: file_info.file_size_bytes, + stream, + }) + } + async fn write_file( &self, path: &AbsolutePathBuf, diff --git a/codex-rs/exec-server/tests/file_system.rs b/codex-rs/exec-server/tests/file_system.rs index c42159a6dd..f82b645547 100644 --- a/codex-rs/exec-server/tests/file_system.rs +++ b/codex-rs/exec-server/tests/file_system.rs @@ -32,6 +32,7 @@ use codex_protocol::permissions::NetworkSandboxPolicy; use codex_sandboxing::policy_transforms::effective_file_system_sandbox_policy; use codex_sandboxing::policy_transforms::effective_network_sandbox_policy; use codex_utils_absolute_path::AbsolutePathBuf; +use futures::StreamExt; use pretty_assertions::assert_eq; use tempfile::TempDir; use test_case::test_case; @@ -521,11 +522,25 @@ async fn file_system_sandboxed_read_allows_readable_root(use_remote: bool) -> Re let sandbox = read_only_sandbox(allowed_dir); let contents = file_system - .read_file(&absolute_path(file_path), Some(&sandbox)) + .read_file(&absolute_path(file_path.clone()), Some(&sandbox)) .await .with_context(|| format!("mode={use_remote}"))?; assert_eq!(contents, b"sandboxed hello"); + let body = file_system + .read_file_body(&absolute_path(file_path), Some(&sandbox)) + .await + .with_context(|| format!("mode={use_remote}"))?; + assert_eq!(body.file_name, "note.txt"); + assert_eq!(body.file_size_bytes, "sandboxed hello".len() as u64); + let chunks = body + .stream + .collect::>() + .await + .into_iter() + .collect::>>()?; + assert_eq!(chunks.concat(), b"sandboxed hello"); + Ok(()) } diff --git a/codex-rs/file-system/Cargo.toml b/codex-rs/file-system/Cargo.toml index 63eaccffd2..a7c5d94c76 100644 --- a/codex-rs/file-system/Cargo.toml +++ b/codex-rs/file-system/Cargo.toml @@ -9,6 +9,8 @@ workspace = true [dependencies] async-trait = { workspace = true } +bytes = { workspace = true } codex-protocol = { workspace = true } codex-utils-absolute-path = { workspace = true } +futures = { workspace = true } serde = { workspace = true, features = ["derive"] } diff --git a/codex-rs/file-system/src/lib.rs b/codex-rs/file-system/src/lib.rs index 026523eb2f..571d10174f 100644 --- a/codex-rs/file-system/src/lib.rs +++ b/codex-rs/file-system/src/lib.rs @@ -1,4 +1,5 @@ use async_trait::async_trait; +use bytes::Bytes; use codex_protocol::config_types::WindowsSandboxLevel; use codex_protocol::models::PermissionProfile; use codex_protocol::models::SandboxEnforcement; @@ -9,8 +10,10 @@ use codex_protocol::permissions::FileSystemSpecialPath; use codex_protocol::permissions::NetworkSandboxPolicy; use codex_protocol::protocol::SandboxPolicy; use codex_utils_absolute_path::AbsolutePathBuf; +use futures::Stream; use std::io; use std::path::Path; +use std::pin::Pin; #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub struct CreateDirectoryOptions { @@ -37,6 +40,14 @@ pub struct FileMetadata { pub modified_at_ms: i64, } +pub type FileByteStream = Pin> + Send + 'static>>; + +pub struct FileReadBody { + pub file_name: String, + pub file_size_bytes: u64, + pub stream: FileByteStream, +} + #[derive(Clone, Debug, Eq, PartialEq)] pub struct ReadDirectoryEntry { pub file_name: String, @@ -139,6 +150,25 @@ pub trait ExecutorFileSystem: Send + Sync { sandbox: Option<&FileSystemSandboxContext>, ) -> FileSystemResult>; + async fn read_file_body( + &self, + path: &AbsolutePathBuf, + sandbox: Option<&FileSystemSandboxContext>, + ) -> FileSystemResult { + let bytes = self.read_file(path, sandbox).await?; + let file_size_bytes = bytes.len() as u64; + let file_name = path + .file_name() + .and_then(|value| value.to_str()) + .unwrap_or("file") + .to_string(); + Ok(FileReadBody { + file_name, + file_size_bytes, + stream: Box::pin(futures::stream::once(async move { Ok(Bytes::from(bytes)) })), + }) + } + /// Reads a file and decodes it as UTF-8 text. async fn read_file_text( &self,