diff --git a/lib/apps/fabro-cli/src/commands/run/petri_worker.rs b/lib/apps/fabro-cli/src/commands/run/petri_worker.rs index cc491e007..ba2a834ff 100644 --- a/lib/apps/fabro-cli/src/commands/run/petri_worker.rs +++ b/lib/apps/fabro-cli/src/commands/run/petri_worker.rs @@ -241,36 +241,16 @@ pub(super) async fn execute(worker: PetriWorker<'_>) -> Result<()> { .with_source(source) .with_publisher(publisher) .with_test_gates(test_checkpoint_gates()); + let environment = &worker.run_state.spec.settings.run.environment; let request = RunRequest { run_id: run_id.to_string(), run_dir: worker.run_dir.join("petri"), execution, store, runtime, - provider: worker - .run_state - .spec - .settings - .run - .environment - .provider - .clone(), - resources: worker - .run_state - .spec - .settings - .run - .environment - .resources - .clone(), - network: worker - .run_state - .spec - .settings - .run - .environment - .network - .clone(), + provider: environment.provider.clone(), + resources: environment.resources.clone(), + network: environment.network.clone(), cancel: cancel_token.clone(), controls: controls.clone(), interviewer: Arc::new(petri_interviewer), diff --git a/lib/apps/fabro-server/src/server/petri_runs.rs b/lib/apps/fabro-server/src/server/petri_runs.rs index eafe55510..99ce39575 100644 --- a/lib/apps/fabro-server/src/server/petri_runs.rs +++ b/lib/apps/fabro-server/src/server/petri_runs.rs @@ -140,9 +140,9 @@ fn settings_layer_toml(state: &AppState) -> Option { /// under `daytona`, and `env`. The rest is the platform's (`cwd`, /// `network`, `lifecycle`, `labels`, `image.dockerfile`, resources the host /// and Docker providers run without, an image the host runs without) and -/// stays with the server's own resolution. The resolved network policy is -/// passed through `RunRequest` at execution; handing it to the frontend would +/// stays with the server's own resolution; handing it to Petri here would /// only warn `ignored.workflow_toml.environments..` on every admit. +/// The resolved network policy reaches Petri through `RunRequest` instead. fn petri_environments(catalog: &MergeMap) -> MergeMap { MergeMap( catalog diff --git a/lib/apps/fabro-server/tests/it/scenario/petri.rs b/lib/apps/fabro-server/tests/it/scenario/petri.rs index a9706bea3..03b28cbeb 100644 --- a/lib/apps/fabro-server/tests/it/scenario/petri.rs +++ b/lib/apps/fabro-server/tests/it/scenario/petri.rs @@ -33,6 +33,7 @@ use fabro_server::test_support::{ use fabro_static::EnvVars; use fabro_store::platform_records::{PlatformRecord, PlatformRecordKind, PlatformRecordStore}; use fabro_test::{TwinScenario, TwinScenarios, twin_openai}; +use fabro_types::settings::run::EnvironmentNetworkMode; use fabro_types::{RunId, WorkflowPath, WorkflowVersion}; use tower::ServiceExt; @@ -1018,7 +1019,13 @@ async fn the_server_attaches_to_the_container_petri_created() { .vault_entries([(EnvVars::OPENAI_API_KEY, namespace.clone())]) .build(); let app = test_app_with_scheduler(Arc::clone(&state)); - create_docker_environment(&app, "docker", CATALOG_IMAGE).await; + create_docker_environment( + &app, + "docker", + CATALOG_IMAGE, + EnvironmentNetworkMode::AllowAll, + ) + .await; let version_id = register_version(&app, &[ ("workflow.fabro", COMMAND_DOT), @@ -1228,15 +1235,11 @@ fn get(path: &str) -> Request { const CATALOG_IMAGE: &str = "ghcr.io/lithoscomputer/ubuntu-22.04:slim"; /// A Docker environment in the server's catalog, with the image it runs. -async fn create_docker_environment(app: &axum::Router, id: &str, image: &str) { - create_docker_environment_with_network(app, id, image, "allow_all").await; -} - -async fn create_docker_environment_with_network( +async fn create_docker_environment( app: &axum::Router, id: &str, image: &str, - mode: &str, + mode: EnvironmentNetworkMode, ) { let environment = serde_json::json!({ "id": id, @@ -1340,21 +1343,22 @@ async fn admitted_root_graph(app: &axum::Router, run_id: &str) -> serde_json::Va /// a Docker daemon, the run's container runs that image. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn a_bundle_naming_a_catalog_environment_runs_on_docker_with_its_image() { - assert_catalog_environment_runs("allow_all", "bridge").await; + assert_catalog_environment_runs(EnvironmentNetworkMode::AllowAll, "bridge").await; } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[ignore = "requires Docker; creates and deletes a container"] async fn a_catalog_network_block_reaches_the_created_docker_container() { + // The shared body returns early without Docker; an explicit run must not. assert!(fabro_test::docker_available()); - assert_catalog_environment_runs("block", "none").await; + assert_catalog_environment_runs(EnvironmentNetworkMode::Block, "none").await; } -async fn assert_catalog_environment_runs(mode: &str, network_mode: &str) { +async fn assert_catalog_environment_runs(mode: EnvironmentNetworkMode, network_mode: &str) { let settings = settings_from_toml("_version = 1\n\n[run.environment]\nid = \"local\"\n"); let state = test_app_state_with_options(settings, 5); let app = test_app_with_scheduler(Arc::clone(&state)); - create_docker_environment_with_network(&app, "docker-small", CATALOG_IMAGE, mode).await; + create_docker_environment(&app, "docker-small", CATALOG_IMAGE, mode).await; let version_id = register_version(&app, &[ ("workflow.fabro", COMMAND_DOT), @@ -1433,7 +1437,9 @@ async fn assert_catalog_environment_runs(mode: &str, network_mode: &str) { } assert!(inspection.status.success(), "{inspection:?}"); assert_eq!( - String::from_utf8(inspection.stdout).unwrap().trim(), + String::from_utf8(inspection.stdout) + .expect("Docker inspect prints UTF-8") + .trim(), network_mode ); } @@ -1445,7 +1451,13 @@ async fn a_bundles_own_environment_table_overrides_the_servers() { let settings = settings_from_toml("_version = 1\n\n[run.environment]\nid = \"local\"\n"); let state = test_app_state_with_options(settings, 5); let app = test_app_with_scheduler(Arc::clone(&state)); - create_docker_environment(&app, "docker-small", CATALOG_IMAGE).await; + create_docker_environment( + &app, + "docker-small", + CATALOG_IMAGE, + EnvironmentNetworkMode::AllowAll, + ) + .await; let version_id = register_version(&app, &[ ("workflow.fabro", COMMAND_DOT), diff --git a/lib/components/fabro-petri/src/engine.rs b/lib/components/fabro-petri/src/engine.rs index 8820fd276..369149be5 100644 --- a/lib/components/fabro-petri/src/engine.rs +++ b/lib/components/fabro-petri/src/engine.rs @@ -207,7 +207,9 @@ pub async fn run(request: RunRequest) -> Result { options.run_key = Some(key.clone()); options.retention = RETENTION; options.sandbox.backend = backend; - if !request.runtime.dry_run { + // The host provider manages no networking and refuses any policy but its + // default, so only container backends receive the run's policy. + if backend != SandboxBackend::Host { options.sandbox.network = network_policy(&request.network); } if backend == SandboxBackend::Daytona { diff --git a/lib/components/fabro-petri/tests/network.rs b/lib/components/fabro-petri/tests/network.rs index 5a424b69b..8d6d3b128 100644 --- a/lib/components/fabro-petri/tests/network.rs +++ b/lib/components/fabro-petri/tests/network.rs @@ -3,18 +3,20 @@ mod support; use std::env; +use std::path::Path; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; use fabro_petri::check::Launch; -use fabro_petri::engine::{self, RunStatus}; -use fabro_petri::providers::SandboxProviderConfig; +use fabro_petri::engine::{self, RunRequest, RunStatus}; +use fabro_petri::providers::{self, DaytonaCredentials, SandboxProviderConfig}; use fabro_petri::prune::{self, PruneRequest}; use fabro_petri::runtime::RuntimeSpec; use fabro_types::settings::run::{EnvironmentNetworkMode, EnvironmentNetworkSettings}; use fabro_types::{RunId, SandboxProviderKind}; use petri_store::MemoryRunStore; +use sandbox_driver::{NetworkPolicy, SandboxFilter, SandboxState, SnapshotFilter}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; use tokio::process::Command; @@ -39,49 +41,24 @@ async fn docker_block_prevents_canary_access_while_allow_all_preserves_it() { })); for (mode, network_mode, expected_hits) in [ (EnvironmentNetworkMode::AllowAll, "bridge", 1), - (EnvironmentNetworkMode::Block, "none", 1), + (EnvironmentNetworkMode::Block, "none", 0), ] { let root = tempfile::tempdir().unwrap(); let run_id = RunId::new().to_string(); let store = Arc::new(MemoryRunStore::new()); - let runtime = RuntimeSpec::default(); let curl = format!( "curl --noproxy '*' --fail --silent --max-time 2 http://host.docker.internal:{port}/canary" ); - let script = if mode == EnvironmentNetworkMode::Block { - format!("if {curl}; then exit 19; fi") - } else { - format!("test $({curl}) = network-canary") - }; - let workflow = format!( - r#"digraph Network {{ - start [shape=Mdiamond] - probe [shape=parallelogram, script="{script}"] - exit [shape=Msquare] - start -> probe -> exit - }}"# - ); - let graphs = support::admit( - &[ - ("workflow.toml", support::SETTINGS), - ("workflow.fabro", &workflow), - ], - Launch::default(), - &runtime, - ); - let mut request = support::run_request( + let request = probe_request( &run_id, root.path(), - graphs, - store.clone(), - runtime, - support::no_questions(Arc::new(support::Silent)), - ); - request.provider = SandboxProviderKind::DOCKER; - request.network = EnvironmentNetworkSettings { + &store, + RuntimeSpec::default(), + SandboxProviderKind::DOCKER, mode, - allow: Vec::new(), - }; + &format!("test $({curl}) = network-canary"), + &curl, + ); let outcome = engine::run(request).await; // Inspect Docker independently of Fabro's settings and driver status. let containers = Command::new("docker") @@ -122,7 +99,7 @@ async fn docker_block_prevents_canary_access_while_allow_all_preserves_it() { String::from_utf8(inspected.stdout).unwrap().trim(), network_mode ); - assert_eq!(hits.load(Ordering::SeqCst), expected_hits, "{mode}"); + assert_eq!(hits.swap(0, Ordering::SeqCst), expected_hits, "{mode}"); } } @@ -130,9 +107,6 @@ async fn docker_block_prevents_canary_access_while_allow_all_preserves_it() { /// runner snapshot and deletes both task-owned sandboxes through Petri. #[fabro_macros::e2e_test(live("DAYTONA_API_KEY"), live("FABRO_TEST_DAYTONA_RUNNER_SNAPSHOT"))] async fn daytona_block_prevents_outbound_https_while_allow_all_preserves_it() { - use fabro_petri::providers::{self, DaytonaCredentials}; - use sandbox_driver::{NetworkPolicy, SandboxFilter, SandboxState, SnapshotFilter}; - let credentials = DaytonaCredentials::from_api_key( fabro_test::require_env("DAYTONA_API_KEY").expect("guard checked the credential"), provider_env, @@ -158,46 +132,21 @@ async fn daytona_block_prevents_outbound_https_while_allow_all_preserves_it() { let root = tempfile::tempdir().unwrap(); let run_id = RunId::new().to_string(); let store = Arc::new(MemoryRunStore::new()); - let runtime = RuntimeSpec { - sandbox: config.clone(), - ..Default::default() - }; let curl = "curl --noproxy '*' --fail --silent --max-time 5 https://example.com/ -o /dev/null"; - let script = if mode == EnvironmentNetworkMode::Block { - format!("if {curl}; then exit 19; fi") - } else { - curl.to_string() - }; - let workflow = format!( - r#"digraph Network {{ - start [shape=Mdiamond] - probe [shape=parallelogram, script="{script}"] - exit [shape=Msquare] - start -> probe -> exit - }}"# - ); - let graphs = support::admit( - &[ - ("workflow.toml", support::SETTINGS), - ("workflow.fabro", &workflow), - ], - Launch::default(), - &runtime, - ); - let mut request = support::run_request( + let request = probe_request( &run_id, root.path(), - graphs, - store.clone(), - runtime, - support::no_questions(Arc::new(support::Silent)), - ); - request.provider = SandboxProviderKind::DAYTONA; - request.network = EnvironmentNetworkSettings { + &store, + RuntimeSpec { + sandbox: config.clone(), + ..Default::default() + }, + SandboxProviderKind::DAYTONA, mode, - allow: Vec::new(), - }; + curl, + curl, + ); let outcome = engine::run(request).await; let mut filter = SandboxFilter::default(); filter @@ -237,6 +186,59 @@ async fn daytona_block_prevents_outbound_https_while_allow_all_preserves_it() { } } +/// A one-stage run whose probe script must succeed under `AllowAll`, and +/// whose `reach` command must fail under `Block`. +#[expect( + clippy::too_many_arguments, + reason = "each live test varies every input" +)] +fn probe_request( + run_id: &str, + root: &Path, + store: &Arc, + runtime: RuntimeSpec, + provider: SandboxProviderKind, + mode: EnvironmentNetworkMode, + allowed_probe: &str, + reach: &str, +) -> RunRequest { + let script = if mode == EnvironmentNetworkMode::Block { + format!("if {reach}; then exit 19; fi") + } else { + allowed_probe.to_string() + }; + let workflow = format!( + r#"digraph Network {{ + start [shape=Mdiamond] + probe [shape=parallelogram, script="{script}"] + exit [shape=Msquare] + start -> probe -> exit + }}"# + ); + let graphs = support::admit( + &[ + ("workflow.toml", support::SETTINGS), + ("workflow.fabro", &workflow), + ], + Launch::default(), + &runtime, + ); + let mut request = support::run_request( + run_id, + root, + graphs, + store.clone(), + runtime, + support::no_questions(Arc::new(support::Silent)), + ); + request.provider = provider; + request.network = EnvironmentNetworkSettings { + mode, + allow: Vec::new(), + }; + request +} + #[expect( clippy::disallowed_methods, reason = "live tests explicitly pass the operator's provider configuration"