mirror of
https://github.com/openai/codex.git
synced 2026-11-10 21:00:47 +00:00
Harden cloud runtime local access boundaries
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
co-authored by
Codex
parent
326e31ab65
commit
af2e9b3b04
@@ -464,8 +464,8 @@ impl MessageProcessor {
|
||||
let fs_processor = FsRequestProcessor::new(
|
||||
thread_manager
|
||||
.environment_manager()
|
||||
.local_environment()
|
||||
.get_filesystem(),
|
||||
.try_local_environment()
|
||||
.map(|environment| environment.get_filesystem()),
|
||||
fs_watch_manager,
|
||||
);
|
||||
let windows_sandbox_processor = WindowsSandboxRequestProcessor::new(
|
||||
|
||||
@@ -33,13 +33,13 @@ use std::sync::Arc;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct FsRequestProcessor {
|
||||
file_system: Arc<dyn ExecutorFileSystem>,
|
||||
file_system: Option<Arc<dyn ExecutorFileSystem>>,
|
||||
fs_watch_manager: FsWatchManager,
|
||||
}
|
||||
|
||||
impl FsRequestProcessor {
|
||||
pub(crate) fn new(
|
||||
file_system: Arc<dyn ExecutorFileSystem>,
|
||||
file_system: Option<Arc<dyn ExecutorFileSystem>>,
|
||||
fs_watch_manager: FsWatchManager,
|
||||
) -> Self {
|
||||
Self {
|
||||
@@ -48,6 +48,13 @@ impl FsRequestProcessor {
|
||||
}
|
||||
}
|
||||
|
||||
fn file_system(&self) -> Result<Arc<dyn ExecutorFileSystem>, JSONRPCErrorError> {
|
||||
self.file_system
|
||||
.as_ref()
|
||||
.map(Arc::clone)
|
||||
.ok_or_else(|| invalid_request("local filesystem access is disabled for this runtime"))
|
||||
}
|
||||
|
||||
pub(crate) async fn connection_closed(&self, connection_id: ConnectionId) {
|
||||
self.fs_watch_manager.connection_closed(connection_id).await;
|
||||
}
|
||||
@@ -57,7 +64,7 @@ impl FsRequestProcessor {
|
||||
params: FsReadFileParams,
|
||||
) -> Result<FsReadFileResponse, JSONRPCErrorError> {
|
||||
let bytes = self
|
||||
.file_system
|
||||
.file_system()?
|
||||
.read_file(¶ms.path, /*sandbox*/ None)
|
||||
.await
|
||||
.map_err(map_fs_error)?;
|
||||
@@ -75,7 +82,7 @@ impl FsRequestProcessor {
|
||||
"fs/writeFile requires valid base64 dataBase64: {err}"
|
||||
))
|
||||
})?;
|
||||
self.file_system
|
||||
self.file_system()?
|
||||
.write_file(¶ms.path, bytes, /*sandbox*/ None)
|
||||
.await
|
||||
.map_err(map_fs_error)?;
|
||||
@@ -86,7 +93,7 @@ impl FsRequestProcessor {
|
||||
&self,
|
||||
params: FsCreateDirectoryParams,
|
||||
) -> Result<FsCreateDirectoryResponse, JSONRPCErrorError> {
|
||||
self.file_system
|
||||
self.file_system()?
|
||||
.create_directory(
|
||||
¶ms.path,
|
||||
CreateDirectoryOptions {
|
||||
@@ -104,7 +111,7 @@ impl FsRequestProcessor {
|
||||
params: FsGetMetadataParams,
|
||||
) -> Result<FsGetMetadataResponse, JSONRPCErrorError> {
|
||||
let metadata = self
|
||||
.file_system
|
||||
.file_system()?
|
||||
.get_metadata(¶ms.path, /*sandbox*/ None)
|
||||
.await
|
||||
.map_err(map_fs_error)?;
|
||||
@@ -122,7 +129,7 @@ impl FsRequestProcessor {
|
||||
params: FsReadDirectoryParams,
|
||||
) -> Result<FsReadDirectoryResponse, JSONRPCErrorError> {
|
||||
let entries = self
|
||||
.file_system
|
||||
.file_system()?
|
||||
.read_directory(¶ms.path, /*sandbox*/ None)
|
||||
.await
|
||||
.map_err(map_fs_error)?;
|
||||
@@ -142,7 +149,7 @@ impl FsRequestProcessor {
|
||||
&self,
|
||||
params: FsRemoveParams,
|
||||
) -> Result<FsRemoveResponse, JSONRPCErrorError> {
|
||||
self.file_system
|
||||
self.file_system()?
|
||||
.remove(
|
||||
¶ms.path,
|
||||
RemoveOptions {
|
||||
@@ -160,7 +167,7 @@ impl FsRequestProcessor {
|
||||
&self,
|
||||
params: FsCopyParams,
|
||||
) -> Result<FsCopyResponse, JSONRPCErrorError> {
|
||||
self.file_system
|
||||
self.file_system()?
|
||||
.copy(
|
||||
¶ms.source_path,
|
||||
¶ms.destination_path,
|
||||
@@ -179,6 +186,7 @@ impl FsRequestProcessor {
|
||||
connection_id: ConnectionId,
|
||||
params: FsWatchParams,
|
||||
) -> Result<FsWatchResponse, JSONRPCErrorError> {
|
||||
self.file_system()?;
|
||||
self.fs_watch_manager.watch(connection_id, params).await
|
||||
}
|
||||
|
||||
@@ -198,3 +206,67 @@ fn map_fs_error(err: io::Error) -> JSONRPCErrorError {
|
||||
internal_error(err.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::error_code::INVALID_REQUEST_ERROR_CODE;
|
||||
use crate::outgoing_message::OutgoingMessageSender;
|
||||
use codex_analytics::AnalyticsEventsClient;
|
||||
use codex_utils_absolute_path::AbsolutePathBuf;
|
||||
use std::path::PathBuf;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
fn absolute_path(path: &str) -> AbsolutePathBuf {
|
||||
AbsolutePathBuf::try_from(PathBuf::from(path)).expect("absolute path")
|
||||
}
|
||||
|
||||
fn processor_without_file_system() -> FsRequestProcessor {
|
||||
let (tx, _rx) = mpsc::channel(1);
|
||||
let outgoing = Arc::new(OutgoingMessageSender::new(
|
||||
tx,
|
||||
AnalyticsEventsClient::disabled(),
|
||||
));
|
||||
FsRequestProcessor::new(/*file_system*/ None, FsWatchManager::new(outgoing))
|
||||
}
|
||||
|
||||
fn assert_local_filesystem_disabled(err: JSONRPCErrorError) {
|
||||
assert_eq!(err.code, INVALID_REQUEST_ERROR_CODE);
|
||||
assert_eq!(
|
||||
err.message,
|
||||
"local filesystem access is disabled for this runtime"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn read_file_fails_when_local_filesystem_is_disabled() {
|
||||
let processor = processor_without_file_system();
|
||||
|
||||
let err = processor
|
||||
.read_file(FsReadFileParams {
|
||||
path: absolute_path("/tmp/disabled-read"),
|
||||
})
|
||||
.await
|
||||
.expect_err("read should fail");
|
||||
|
||||
assert_local_filesystem_disabled(err);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn watch_fails_when_local_filesystem_is_disabled() {
|
||||
let processor = processor_without_file_system();
|
||||
|
||||
let err = processor
|
||||
.watch(
|
||||
ConnectionId(1),
|
||||
FsWatchParams {
|
||||
watch_id: "disabled-watch".to_string(),
|
||||
path: absolute_path("/tmp/disabled-watch"),
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect_err("watch should fail");
|
||||
|
||||
assert_local_filesystem_disabled(err);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -205,17 +205,17 @@ impl McpRequestProcessor {
|
||||
.await;
|
||||
let auth = self.auth_manager.auth().await;
|
||||
let environment_manager = self.thread_manager.environment_manager();
|
||||
let runtime_environment = match environment_manager.default_environment() {
|
||||
Some(environment) => {
|
||||
// Status listing has no turn cwd. This fallback is used only
|
||||
// by executor-backed stdio MCPs whose config omits `cwd`.
|
||||
McpRuntimeEnvironment::new(environment, config.cwd.to_path_buf())
|
||||
}
|
||||
None => McpRuntimeEnvironment::new(
|
||||
environment_manager.local_environment(),
|
||||
config.cwd.to_path_buf(),
|
||||
),
|
||||
};
|
||||
// Status listing has no turn cwd. This fallback is used only by MCPs
|
||||
// whose config omits `cwd`; when the runtime has no local environment,
|
||||
// stdio MCP startup will fail closed while HTTP MCPs can still be
|
||||
// queried.
|
||||
let runtime_environment = environment_manager
|
||||
.default_environment()
|
||||
.or_else(|| environment_manager.try_local_environment())
|
||||
.map_or_else(
|
||||
|| McpRuntimeEnvironment::without_environment(config.cwd.to_path_buf()),
|
||||
|environment| McpRuntimeEnvironment::new(environment, config.cwd.to_path_buf()),
|
||||
);
|
||||
|
||||
tokio::spawn(async move {
|
||||
Self::list_mcp_server_status_task(
|
||||
@@ -371,12 +371,14 @@ impl McpRequestProcessor {
|
||||
let auth = self.auth_manager.auth().await;
|
||||
let runtime_environment = {
|
||||
let environment_manager = self.thread_manager.environment_manager();
|
||||
let environment = environment_manager
|
||||
let fallback_cwd = config.cwd.to_path_buf();
|
||||
environment_manager
|
||||
.default_environment()
|
||||
.unwrap_or_else(|| environment_manager.local_environment());
|
||||
// Resource reads without a thread have no turn cwd. This fallback
|
||||
// is used only by executor-backed stdio MCPs whose config omits `cwd`.
|
||||
McpRuntimeEnvironment::new(environment, config.cwd.to_path_buf())
|
||||
.or_else(|| environment_manager.try_local_environment())
|
||||
.map_or_else(
|
||||
|| McpRuntimeEnvironment::without_environment(fallback_cwd.clone()),
|
||||
|environment| McpRuntimeEnvironment::new(environment, fallback_cwd.clone()),
|
||||
)
|
||||
};
|
||||
let request_id = request_id.clone();
|
||||
|
||||
|
||||
@@ -574,7 +574,12 @@ async fn make_rmcp_client(
|
||||
let remote_environment = match experimental_environment.as_deref() {
|
||||
None | Some("local") => false,
|
||||
Some("remote") => {
|
||||
if !runtime_environment.environment().is_remote() {
|
||||
let Some(environment) = runtime_environment.environment() else {
|
||||
return Err(StartupOutcomeError::from(anyhow!(
|
||||
"remote MCP server `{server_name}` requires an environment"
|
||||
)));
|
||||
};
|
||||
if !environment.is_remote() {
|
||||
return Err(StartupOutcomeError::from(anyhow!(
|
||||
"remote MCP server `{server_name}` requires a remote environment"
|
||||
)));
|
||||
@@ -604,11 +609,21 @@ async fn make_rmcp_client(
|
||||
.collect::<HashMap<_, _>>()
|
||||
});
|
||||
let launcher = if remote_environment {
|
||||
let Some(environment) = runtime_environment.environment() else {
|
||||
return Err(StartupOutcomeError::from(anyhow!(
|
||||
"remote MCP server `{server_name}` requires an environment"
|
||||
)));
|
||||
};
|
||||
Arc::new(ExecutorStdioServerLauncher::new(
|
||||
runtime_environment.environment().get_exec_backend(),
|
||||
environment.get_exec_backend(),
|
||||
runtime_environment.fallback_cwd(),
|
||||
))
|
||||
} else {
|
||||
if runtime_environment.environment().is_none() {
|
||||
return Err(StartupOutcomeError::from(anyhow!(
|
||||
"local stdio MCP server `{server_name}` is disabled because no local environment is available"
|
||||
)));
|
||||
}
|
||||
Arc::new(LocalStdioServerLauncher::new(
|
||||
runtime_environment.fallback_cwd(),
|
||||
)) as Arc<dyn StdioServerLauncher>
|
||||
@@ -628,7 +643,12 @@ async fn make_rmcp_client(
|
||||
bearer_token_env_var,
|
||||
} => {
|
||||
let http_client: Arc<dyn HttpClient> = if remote_environment {
|
||||
runtime_environment.environment().get_http_client()
|
||||
let Some(environment) = runtime_environment.environment() else {
|
||||
return Err(StartupOutcomeError::from(anyhow!(
|
||||
"remote MCP server `{server_name}` requires an environment"
|
||||
)));
|
||||
};
|
||||
environment.get_http_client()
|
||||
} else {
|
||||
Arc::new(ReqwestHttpClient)
|
||||
};
|
||||
@@ -659,6 +679,32 @@ mod tests {
|
||||
use rmcp::model::JsonObject;
|
||||
use rmcp::model::Meta;
|
||||
|
||||
fn stdio_server_config(command: &str) -> McpServerConfig {
|
||||
McpServerConfig {
|
||||
transport: McpServerTransportConfig::Stdio {
|
||||
command: command.to_string(),
|
||||
args: Vec::new(),
|
||||
env: None,
|
||||
env_vars: Vec::new(),
|
||||
cwd: None,
|
||||
},
|
||||
experimental_environment: None,
|
||||
enabled: true,
|
||||
required: false,
|
||||
supports_parallel_tool_calls: false,
|
||||
disabled_reason: None,
|
||||
startup_timeout_sec: None,
|
||||
tool_timeout_sec: None,
|
||||
default_tools_approval_mode: None,
|
||||
enabled_tools: None,
|
||||
disabled_tools: None,
|
||||
scopes: None,
|
||||
oauth: None,
|
||||
oauth_resource: None,
|
||||
tools: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn tool_with_connector_meta() -> RmcpTool {
|
||||
RmcpTool {
|
||||
name: "capture_file_upload".to_string().into(),
|
||||
@@ -724,6 +770,32 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stdio_mcp_fails_closed_without_local_environment() {
|
||||
let server = EffectiveMcpServer::configured(stdio_server_config("echo"));
|
||||
let runtime_environment = McpRuntimeEnvironment::without_environment(
|
||||
std::env::current_dir().expect("current dir"),
|
||||
);
|
||||
|
||||
let result = make_rmcp_client(
|
||||
"local-test",
|
||||
server,
|
||||
OAuthCredentialsStoreMode::File,
|
||||
runtime_environment,
|
||||
/*runtime_auth_provider*/ None,
|
||||
)
|
||||
.await;
|
||||
let Err(err) = result else {
|
||||
panic!("stdio MCP should be disabled without an environment");
|
||||
};
|
||||
|
||||
assert!(err.to_string().contains("local stdio MCP server"));
|
||||
assert!(
|
||||
err.to_string()
|
||||
.contains("no local environment is available")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn codex_apps_connector_metadata_is_preserved() {
|
||||
let mut tool = tool_with_connector_meta();
|
||||
|
||||
@@ -38,20 +38,27 @@ pub struct SandboxState {
|
||||
/// process working directory.
|
||||
#[derive(Clone)]
|
||||
pub struct McpRuntimeEnvironment {
|
||||
environment: Arc<Environment>,
|
||||
environment: Option<Arc<Environment>>,
|
||||
fallback_cwd: PathBuf,
|
||||
}
|
||||
|
||||
impl McpRuntimeEnvironment {
|
||||
pub fn new(environment: Arc<Environment>, fallback_cwd: PathBuf) -> Self {
|
||||
Self {
|
||||
environment,
|
||||
environment: Some(environment),
|
||||
fallback_cwd,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn environment(&self) -> Arc<Environment> {
|
||||
Arc::clone(&self.environment)
|
||||
pub fn without_environment(fallback_cwd: PathBuf) -> Self {
|
||||
Self {
|
||||
environment: None,
|
||||
fallback_cwd,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn environment(&self) -> Option<Arc<Environment>> {
|
||||
self.environment.as_ref().map(Arc::clone)
|
||||
}
|
||||
|
||||
pub(crate) fn fallback_cwd(&self) -> PathBuf {
|
||||
|
||||
@@ -261,9 +261,13 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_environment_manager(
|
||||
let (tx_event, rx_event) = unbounded();
|
||||
drop(rx_event);
|
||||
|
||||
let environment = environment_manager
|
||||
let runtime_environment = environment_manager
|
||||
.default_environment()
|
||||
.unwrap_or_else(|| environment_manager.local_environment());
|
||||
.or_else(|| environment_manager.try_local_environment())
|
||||
.map_or_else(
|
||||
|| McpRuntimeEnvironment::without_environment(config.cwd.to_path_buf()),
|
||||
|environment| McpRuntimeEnvironment::new(environment, config.cwd.to_path_buf()),
|
||||
);
|
||||
|
||||
let (mut mcp_connection_manager, cancel_token) = McpConnectionManager::new(
|
||||
&mcp_servers,
|
||||
@@ -273,7 +277,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_environment_manager(
|
||||
INITIAL_SUBMIT_ID.to_owned(),
|
||||
tx_event,
|
||||
PermissionProfile::default(),
|
||||
McpRuntimeEnvironment::new(environment, config.cwd.to_path_buf()),
|
||||
runtime_environment,
|
||||
config.codex_home.to_path_buf(),
|
||||
codex_apps_tools_cache_key(auth.as_ref()),
|
||||
host_owned_codex_apps_enabled,
|
||||
|
||||
@@ -294,14 +294,18 @@ impl Session {
|
||||
Arc::clone(&turn_environment.environment),
|
||||
turn_environment.cwd.to_path_buf(),
|
||||
),
|
||||
None => McpRuntimeEnvironment::new(
|
||||
None => {
|
||||
#[allow(deprecated)]
|
||||
let fallback_cwd = turn_context.cwd.to_path_buf();
|
||||
self.services
|
||||
.environment_manager
|
||||
.default_environment()
|
||||
.unwrap_or_else(|| self.services.environment_manager.local_environment()),
|
||||
#[allow(deprecated)]
|
||||
turn_context.cwd.to_path_buf(),
|
||||
),
|
||||
.or_else(|| self.services.environment_manager.try_local_environment())
|
||||
.map_or_else(
|
||||
|| McpRuntimeEnvironment::without_environment(fallback_cwd.clone()),
|
||||
|environment| McpRuntimeEnvironment::new(environment, fallback_cwd.clone()),
|
||||
)
|
||||
}
|
||||
};
|
||||
{
|
||||
let mut guard = self.services.mcp_startup_cancellation_token.lock().await;
|
||||
|
||||
@@ -1045,13 +1045,24 @@ impl Session {
|
||||
Arc::clone(&turn_environment.environment),
|
||||
turn_environment.cwd.to_path_buf(),
|
||||
),
|
||||
None => McpRuntimeEnvironment::new(
|
||||
sess.services
|
||||
.environment_manager
|
||||
.default_environment()
|
||||
.unwrap_or_else(|| sess.services.environment_manager.local_environment()),
|
||||
session_configuration.cwd.to_path_buf(),
|
||||
),
|
||||
None => sess
|
||||
.services
|
||||
.environment_manager
|
||||
.default_environment()
|
||||
.or_else(|| sess.services.environment_manager.try_local_environment())
|
||||
.map_or_else(
|
||||
|| {
|
||||
McpRuntimeEnvironment::without_environment(
|
||||
session_configuration.cwd.to_path_buf(),
|
||||
)
|
||||
},
|
||||
|environment| {
|
||||
McpRuntimeEnvironment::new(
|
||||
environment,
|
||||
session_configuration.cwd.to_path_buf(),
|
||||
)
|
||||
},
|
||||
),
|
||||
};
|
||||
let (mcp_connection_manager, cancel_token) = McpConnectionManager::new(
|
||||
&mcp_servers,
|
||||
|
||||
@@ -33,7 +33,9 @@ pub const CODEX_EXEC_SERVER_URL_ENV_VAR: &str = "CODEX_EXEC_SERVER_URL";
|
||||
/// the default environment unset while still keeping an explicit local
|
||||
/// environment available through `local_environment()`. Callers use
|
||||
/// `default_environment().is_some()` as the signal for model-facing
|
||||
/// shell/filesystem tool availability.
|
||||
/// shell/filesystem tool availability. Cloud-hosted runtimes can use
|
||||
/// `disabled_without_local()` to make local access absent even for internal
|
||||
/// fallback paths.
|
||||
///
|
||||
/// Remote environments create remote filesystem and execution backends that
|
||||
/// lazy-connect to the configured exec-server on first use. The remote
|
||||
@@ -42,7 +44,8 @@ pub const CODEX_EXEC_SERVER_URL_ENV_VAR: &str = "CODEX_EXEC_SERVER_URL";
|
||||
pub struct EnvironmentManager {
|
||||
default_environment: Option<String>,
|
||||
environments: RwLock<HashMap<String, Arc<Environment>>>,
|
||||
local_environment: Arc<Environment>,
|
||||
local_environment: Option<Arc<Environment>>,
|
||||
local_runtime_paths: Option<ExecServerRuntimePaths>,
|
||||
}
|
||||
|
||||
pub const LOCAL_ENVIRONMENT_ID: &str = "local";
|
||||
@@ -64,22 +67,40 @@ impl EnvironmentManagerArgs {
|
||||
impl EnvironmentManager {
|
||||
/// Builds a test-only manager without configured sandbox helper paths.
|
||||
pub fn default_for_tests() -> Self {
|
||||
let local_environment = Arc::new(Environment::default_for_tests());
|
||||
Self {
|
||||
default_environment: Some(LOCAL_ENVIRONMENT_ID.to_string()),
|
||||
environments: RwLock::new(HashMap::from([(
|
||||
LOCAL_ENVIRONMENT_ID.to_string(),
|
||||
Arc::new(Environment::default_for_tests()),
|
||||
local_environment.clone(),
|
||||
)])),
|
||||
local_environment: Arc::new(Environment::default_for_tests()),
|
||||
local_environment: Some(local_environment),
|
||||
local_runtime_paths: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Builds a test-only manager with environment access disabled.
|
||||
pub fn disabled_for_tests(local_runtime_paths: ExecServerRuntimePaths) -> Self {
|
||||
let local_environment = Arc::new(Environment::local(local_runtime_paths.clone()));
|
||||
Self {
|
||||
default_environment: None,
|
||||
environments: RwLock::new(HashMap::new()),
|
||||
local_environment: Arc::new(Environment::local(local_runtime_paths)),
|
||||
local_environment: Some(local_environment),
|
||||
local_runtime_paths: Some(local_runtime_paths),
|
||||
}
|
||||
}
|
||||
|
||||
/// Builds a manager with no default or local environment available.
|
||||
///
|
||||
/// This is intended for cloud-hosted orchestrator runtimes that must not
|
||||
/// have a worker-local filesystem or process fallback. Remote environments
|
||||
/// may still be added explicitly.
|
||||
pub fn disabled_without_local() -> Self {
|
||||
Self {
|
||||
default_environment: None,
|
||||
environments: RwLock::new(HashMap::new()),
|
||||
local_environment: None,
|
||||
local_runtime_paths: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -171,7 +192,7 @@ impl EnvironmentManager {
|
||||
} = snapshot;
|
||||
let mut environment_map =
|
||||
HashMap::with_capacity(environments.len() + usize::from(include_local));
|
||||
let local_environment = Arc::new(Environment::local(local_runtime_paths));
|
||||
let local_environment = Arc::new(Environment::local(local_runtime_paths.clone()));
|
||||
if include_local {
|
||||
environment_map.insert(
|
||||
LOCAL_ENVIRONMENT_ID.to_string(),
|
||||
@@ -212,7 +233,8 @@ impl EnvironmentManager {
|
||||
Ok(Self {
|
||||
default_environment,
|
||||
environments: RwLock::new(environment_map),
|
||||
local_environment,
|
||||
local_environment: Some(local_environment),
|
||||
local_runtime_paths: Some(local_runtime_paths),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -250,7 +272,15 @@ impl EnvironmentManager {
|
||||
|
||||
/// Returns the local environment instance used for internal runtime work.
|
||||
pub fn local_environment(&self) -> Arc<Environment> {
|
||||
Arc::clone(&self.local_environment)
|
||||
let Some(environment) = self.try_local_environment() else {
|
||||
panic!("local environment is disabled for this runtime");
|
||||
};
|
||||
environment
|
||||
}
|
||||
|
||||
/// Returns the local environment if this runtime is allowed to access it.
|
||||
pub fn try_local_environment(&self) -> Option<Arc<Environment>> {
|
||||
self.local_environment.as_ref().map(Arc::clone)
|
||||
}
|
||||
|
||||
/// Returns a named environment instance.
|
||||
@@ -274,6 +304,11 @@ impl EnvironmentManager {
|
||||
"environment id cannot be empty".to_string(),
|
||||
));
|
||||
}
|
||||
if environment_id == LOCAL_ENVIRONMENT_ID {
|
||||
return Err(ExecServerError::Protocol(format!(
|
||||
"environment id `{LOCAL_ENVIRONMENT_ID}` is reserved for EnvironmentManager"
|
||||
)));
|
||||
}
|
||||
let (exec_server_url, disabled) = normalize_exec_server_url(Some(exec_server_url));
|
||||
if disabled {
|
||||
return Err(ExecServerError::Protocol(
|
||||
@@ -285,10 +320,8 @@ impl EnvironmentManager {
|
||||
"remote environment requires an exec-server url".to_string(),
|
||||
));
|
||||
};
|
||||
let environment = Environment::remote_inner(
|
||||
exec_server_url,
|
||||
self.local_environment.local_runtime_paths.clone(),
|
||||
);
|
||||
let environment =
|
||||
Environment::remote_inner(exec_server_url, self.local_runtime_paths.clone());
|
||||
self.environments
|
||||
.write()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
@@ -512,11 +545,23 @@ mod tests {
|
||||
|
||||
assert!(manager.default_environment().is_none());
|
||||
assert_eq!(manager.default_environment_id(), None);
|
||||
assert!(manager.try_local_environment().is_some());
|
||||
assert!(!manager.local_environment().is_remote());
|
||||
assert!(manager.get_environment(LOCAL_ENVIRONMENT_ID).is_none());
|
||||
assert!(manager.get_environment(REMOTE_ENVIRONMENT_ID).is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn disabled_without_local_environment_removes_worker_local_fallback() {
|
||||
let manager = EnvironmentManager::disabled_without_local();
|
||||
|
||||
assert!(manager.default_environment().is_none());
|
||||
assert_eq!(manager.default_environment_id(), None);
|
||||
assert!(manager.try_local_environment().is_none());
|
||||
assert!(manager.get_environment(LOCAL_ENVIRONMENT_ID).is_none());
|
||||
assert!(manager.get_environment(REMOTE_ENVIRONMENT_ID).is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn environment_manager_reports_remote_url() {
|
||||
let manager = EnvironmentManager::create_for_tests(
|
||||
@@ -765,6 +810,7 @@ mod tests {
|
||||
assert_eq!(manager.default_environment_id(), None);
|
||||
assert!(manager.get_environment(LOCAL_ENVIRONMENT_ID).is_none());
|
||||
assert!(manager.get_environment(REMOTE_ENVIRONMENT_ID).is_none());
|
||||
assert!(manager.try_local_environment().is_some());
|
||||
assert!(!manager.local_environment().is_remote());
|
||||
}
|
||||
|
||||
@@ -800,6 +846,22 @@ mod tests {
|
||||
assert!(!Arc::ptr_eq(&first, &second));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn environment_manager_without_local_can_upsert_named_remote_environment() {
|
||||
let manager = EnvironmentManager::disabled_without_local();
|
||||
|
||||
manager
|
||||
.upsert_environment("executor-a".to_string(), "ws://127.0.0.1:8765".to_string())
|
||||
.expect("remote environment");
|
||||
let environment = manager
|
||||
.get_environment("executor-a")
|
||||
.expect("remote environment");
|
||||
|
||||
assert!(environment.is_remote());
|
||||
assert_eq!(environment.exec_server_url(), Some("ws://127.0.0.1:8765"));
|
||||
assert_eq!(environment.local_runtime_paths(), None);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn environment_manager_rejects_empty_remote_environment_url() {
|
||||
let manager = EnvironmentManager::disabled_for_tests(test_runtime_paths());
|
||||
@@ -814,6 +876,23 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn environment_manager_rejects_upserted_local_environment_id() {
|
||||
let manager = EnvironmentManager::disabled_without_local();
|
||||
|
||||
let err = manager
|
||||
.upsert_environment(
|
||||
LOCAL_ENVIRONMENT_ID.to_string(),
|
||||
"ws://127.0.0.1:8765".to_string(),
|
||||
)
|
||||
.expect_err("local environment id should fail");
|
||||
|
||||
assert_eq!(
|
||||
err.to_string(),
|
||||
"exec-server protocol error: environment id `local` is reserved for EnvironmentManager"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn default_environment_has_ready_local_executor() {
|
||||
let environment = Environment::default_for_tests();
|
||||
|
||||
Reference in New Issue
Block a user