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

use fs abstraction

Browse Source
This commit is contained in:
colby-oai
2026-04-28 14:21:56 -04:00
parent 0d5a38b145
commit 1b65ab838f
14 changed files with 438 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+2
View File
@@ -2729,8 +2729,10 @@ name = "codex-file-system"
version = "0.0.0"
dependencies = [
"async-trait",
"bytes",
"codex-protocol",
"codex-utils-absolute-path",
"futures",
"serde",
]
codex-rs/codex-api/src/files.rs
+59 -12
View File
@@ -27,7 +27,7 @@ pub struct UploadedOpenAiFile {
pub file_name: String,
pub file_size_bytes: u64,
pub mime_type: Option<String>,
pub path: PathBuf,
pub path: Option<PathBuf>,
}
#[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<UploadedOpenAiFile, OpenAiFileError> {
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<PathBuf>,
) -> Result<UploadedOpenAiFile, OpenAiFileError> {
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" => {
codex-rs/codex-api/src/lib.rs
+1
View File
@@ -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;
codex-rs/core/src/mcp_openai_file.rs
+58 -11
View File
@@ -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"));
}
}
codex-rs/exec-server/Cargo.toml
+1 -1
View File
@@ -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"] }
codex-rs/exec-server/src/fs_helper.rs
+72
View File
@@ -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<FsReadFileInfoResponse, JSONRPCErrorError> {
match self {
Self::ReadFileInfo(response) => Ok(response),
other => Err(unexpected_response("fs/readFileInfo", other.operation())),
}
}
pub(crate) fn expect_write_file(self) -> Result<FsWriteFileResponse, JSONRPCErrorError> {
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<dyn std::error::Error + Send + Sync>> {
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()),
codex-rs/exec-server/src/fs_helper_main.rs
+7
View File
@@ -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<dyn Error + Send + Sync>> {
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),
codex-rs/exec-server/src/fs_sandbox.rs
+93
View File
@@ -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<FileByteStream, JSONRPCErrorError> {
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<tokio::process::ChildStdout>,
child: tokio::process::Child,
stderr_task: tokio::task::JoinHandle<Vec<u8>>,
}
async fn run_command_stream(
command: SandboxExecRequest,
request_json: Vec<u8>,
) -> Result<FileByteStream, JSONRPCErrorError> {
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,
codex-rs/exec-server/src/lib.rs
+1
View File
@@ -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;
codex-rs/exec-server/src/local_file_system.rs
+59 -1
View File
@@ -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<Arc<dyn ExecutorFileSystem>> =
LazyLock::new(|| -> Arc<dyn ExecutorFileSystem> { 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<FileReadBody> {
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<FileReadBody> {
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<FileReadBody> {
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)? {
codex-rs/exec-server/src/sandboxed_file_system.rs
+37
View File
@@ -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<FileReadBody> {
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,
codex-rs/exec-server/tests/file_system.rs
+16 -1
View File
@@ -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::<Vec<_>>()
.await
.into_iter()
.collect::<std::io::Result<Vec<_>>>()?;
assert_eq!(chunks.concat(), b"sandboxed hello");
Ok(())
}
codex-rs/file-system/Cargo.toml
+2
View File
@@ -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"] }
codex-rs/file-system/src/lib.rs
+30
View File
@@ -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<Box<dyn Stream<Item = io::Result<Bytes>> + 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<Vec<u8>>;
async fn read_file_body(
&self,
path: &AbsolutePathBuf,
sandbox: Option<&FileSystemSandboxContext>,
) -> FileSystemResult<FileReadBody> {
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,