eval: feat: no network access in the evaluator

backend/gradient-db/src/metrics/worker_samples.rs

@@ -78,7 +78,7 @@ fn board_fleet_sql(window_hours: i64, project_filter: Option<&str>) -> String {
                 count(DISTINCT worker_id) AS connected, \
                 count(DISTINCT worker_id) FILTER (WHERE state = {draining}) AS draining, \
                 count(DISTINCT worker_id) FILTER (WHERE (capabilities->>'eval')::boolean) AS ev, \
-                count(DISTINCT worker_id) FILTER (WHERE (capabilities->>'fetch')::boolean) AS ft, \
+                count(DISTINCT worker_id) FILTER (WHERE (capabilities->>'network')::boolean) AS nw, \
                 count(DISTINCT worker_id) FILTER (WHERE (capabilities->>'build')::boolean) AS bd \
          FROM worker_sample \
          WHERE at >= (now() AT TIME ZONE 'UTC') - interval '{window_hours} hours'",
@@ -103,7 +103,7 @@ pub struct FleetCount {
     pub connected: i64,
     pub draining: i64,
     pub eval: i64,
-    pub fetch: i64,
+    pub network: i64,
     pub build: i64,
 }
 
@@ -123,7 +123,7 @@ pub async fn fleet_per_hour<C: ConnectionTrait>(
             connected: r.try_get("", "connected").unwrap_or(0),
             draining: r.try_get("", "draining").unwrap_or(0),
             eval: r.try_get("", "ev").unwrap_or(0),
-            fetch: r.try_get("", "ft").unwrap_or(0),
+            network: r.try_get("", "nw").unwrap_or(0),
             build: r.try_get("", "bd").unwrap_or(0),
         })
         .collect())

backend/gradient-db/src/teams/demand.rs

@@ -15,7 +15,8 @@ use std::collections::BTreeMap;
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, utoipa::ToSchema)]
 pub struct TeamDemand {
     pub builds: Vec<UnmetRequirement>,
-    pub fetch: u32,
+    pub network: u32,
+    pub secrets: u32,
     pub eval: u32,
 }
 
@@ -63,7 +64,8 @@ fn aggregate(reasons: impl IntoIterator<Item = WaitingReason>) -> TeamDemand {
                 }
             }
             WaitingReason::EvalWorkers { capability, .. } => match capability {
-                EvalCapability::Fetch => demand.fetch += 1,
+                EvalCapability::Network => demand.network += 1,
+                EvalCapability::Secrets => demand.secrets += 1,
                 EvalCapability::Eval => demand.eval += 1,
             },
             _ => {}
@@ -147,13 +149,14 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn counts_eval_and_fetch_gaps() {
+    async fn counts_eval_network_and_secrets_gaps() {
         let db = mock_db()
             .append_query_results([vec![grant(ProjectId::now_v7())]])
             .append_query_results([vec![
                 waiting(WaitingReason::eval_workers(EvalCapability::Eval, 0)),
                 waiting(WaitingReason::eval_workers(EvalCapability::Eval, 2)),
-                waiting(WaitingReason::eval_workers(EvalCapability::Fetch, 0)),
+                waiting(WaitingReason::eval_workers(EvalCapability::Network, 0)),
+                waiting(WaitingReason::eval_workers(EvalCapability::Secrets, 1)),
                 waiting(WaitingReason::Draining),
             ]])
             .into_connection();
@@ -162,7 +165,7 @@ mod tests {
             .await
             .expect("demand");
 
-        assert_eq!((demand.eval, demand.fetch), (2, 1));
+        assert_eq!((demand.eval, demand.network, demand.secrets), (2, 1, 1));
         assert!(demand.builds.is_empty());
     }
 

backend/gradient-entity/src/team_worker.rs

@@ -24,7 +24,9 @@ pub struct Model {
     pub url: Option<String>,
     pub display_name: String,
     pub gradient_ci: bool,
-    pub enable_fetch: bool,
+    pub enable_network: bool,
+    pub enable_secrets: bool,
+    pub enable_exec: bool,
     pub enable_eval: bool,
     pub enable_build: bool,
     pub active: bool,

backend/gradient-entity/src/worker_registration.rs

@@ -25,7 +25,9 @@ pub struct Model {
     /// connection.
     pub url: Option<String>,
     pub active: bool,
-    pub enable_fetch: bool,
+    pub enable_network: bool,
+    pub enable_secrets: bool,
+    pub enable_exec: bool,
     pub enable_eval: bool,
     pub enable_build: bool,
     pub display_name: String,

backend/gradient-eval/src/eval_worker.rs

@@ -14,7 +14,7 @@ use crate::frames::Frames;
 use crate::ipc::{
     EVAL_IPC_VERSION, EvalRequest, EvalResponse, ResolvedItem, decode_request, read_frame,
 };
-use crate::nix_eval::{NixEvaluator, RealiseHook, StatsReader};
+use crate::nix_eval::{FetchHook, NixEvaluator, RealiseHook, StatsReader};
 
 pub(crate) struct ParentPipe<R, W> {
     reader: Mutex<R>,
@@ -34,45 +34,72 @@ impl<R: Read, W: Write> ParentPipe<R, W> {
     pub(crate) fn read_frame(&self) -> std::io::Result<Option<Vec<u8>>> {
         if self.broken.load(Ordering::Acquire) {
             return Err(std::io::Error::other(
-                "eval pipe broke during a build request",
+                "eval pipe broke during a request to the parent",
             ));
         }
         read_frame(&mut *self.reader.lock().unwrap_or_else(PoisonError::into_inner))
     }
 
     pub(crate) fn request_build(&self, derived_paths: &[String]) -> Result<(), String> {
+        let need = EvalResponse::NeedsBuild {
+            derived_paths: derived_paths.to_vec(),
+        };
+        match self.ask(&need, "a build")? {
+            EvalRequest::BuildDone { error: None } => Ok(()),
+            EvalRequest::BuildDone { error: Some(error) } => Err(error),
+            other => Err(self.break_on(Some(other), "a build")),
+        }
+    }
+
+    pub(crate) fn request_input(&self, request: &str) -> Result<String, String> {
+        let need = EvalResponse::NeedsInput {
+            request: request.to_owned(),
+        };
+        match self.ask(&need, "an input")? {
+            EvalRequest::InputDone {
+                response: Some(response),
+                error: None,
+            } => Ok(response),
+            EvalRequest::InputDone {
+                error: Some(error), ..
+            } => Err(error),
+            other => Err(self.break_on(Some(other), "an input")),
+        }
+    }
+
+    fn ask(&self, need: &EvalResponse, what: &str) -> Result<EvalRequest, String> {
         self.frames
-            .send(&EvalResponse::NeedsBuild {
-                derived_paths: derived_paths.to_vec(),
-            })
-            .map_err(|e| format!("eval pipe closed while asking for a build: {e}"))?;
+            .send(need)
+            .map_err(|e| format!("eval pipe closed while asking for {what}: {e}"))?;
 
         let answer = match self.read_frame() {
             Ok(Some(payload)) => decode_request(&payload).ok(),
             Ok(None) | Err(_) => None,
         };
-        match answer {
-            Some(EvalRequest::BuildDone { error: None }) => Ok(()),
-            Some(EvalRequest::BuildDone { error: Some(error) }) => Err(error),
-            other => {
-                self.broken.store(true, Ordering::Release);
-                error!(?other, "eval worker: expected a build result");
-                Err("eval pipe closed while waiting for a build".to_string())
-            }
-        }
+        answer.ok_or_else(|| self.break_on(None, what))
+    }
+
+    fn break_on(&self, answer: Option<EvalRequest>, what: &str) -> String {
+        self.broken.store(true, Ordering::Release);
+        error!(?answer, "eval worker: expected the answer for {what}");
+        format!("eval pipe closed while waiting for {what}")
     }
 }
 
 pub fn run_eval_worker(collect_stats: bool) -> std::io::Result<()> {
     let mut stdout = std::io::stdout();
     stdout.write_all(&[EVAL_IPC_VERSION])?;
     stdout.flush()?;
+    let mut fetch_through_parent = [0u8; 1];
+    std::io::stdin().read_exact(&mut fetch_through_parent)?;
+    let fetch_through_parent = fetch_through_parent[0] == 1;
 
     let pipe = Arc::new(ParentPipe::new(
         std::io::stdin(),
         Frames::new(stdout, nix_bindings::EvalStats::default()),
     ));
-    let evaluator = match NixEvaluator::new(Some(realise_through(Arc::clone(&pipe)))) {
+    let fetch_hook = fetch_through_parent.then(|| fetch_through(Arc::clone(&pipe)));
+    let evaluator = match NixEvaluator::new(Some(realise_through(Arc::clone(&pipe))), fetch_hook) {
         Ok(e) => Some(e),
         Err(e) => {
             error!(
@@ -113,6 +140,14 @@ where
     Box::new(move |derived_paths| pipe.request_build(derived_paths))
 }
 
+fn fetch_through<R, W>(pipe: Arc<ParentPipe<R, W>>) -> FetchHook
+where
+    R: Read + Send + 'static,
+    W: Write + Send + 'static,
+{
+    Box::new(move |request| pipe.request_input(request))
+}
+
 const STATS_TICK: std::time::Duration = std::time::Duration::from_secs(1);
 
 fn tick_stats<W: Write>(
@@ -168,8 +203,8 @@ fn answer_requests<R: Read, W: Write>(
                 trace!("eval worker shutting down on request");
                 return Ok(());
             }
-            EvalRequest::BuildDone { .. } => EvalResponse::Err {
-                message: "a build result arrived outside a build request".to_string(),
+            EvalRequest::BuildDone { .. } | EvalRequest::InputDone { .. } => EvalResponse::Err {
+                message: "an answer arrived outside a request to the parent".to_string(),
             },
             EvalRequest::Plan {
                 repository,
@@ -218,12 +253,12 @@ fn answer_requests<R: Read, W: Write>(
                 )
             }),
             EvalRequest::FetchInput {
-                locked,
+                request,
                 git_ssh_command,
             } => with_evaluator(evaluator, |ev| {
                 or_err(
-                    ev.fetch_tree(&locked, git_ssh_command.as_deref())
-                        .map(|store_path| EvalResponse::FetchOk { store_path }),
+                    ev.fetch_tree(&request, git_ssh_command.as_deref())
+                        .map(|response| EvalResponse::FetchOk { response }),
                 )
             }),
             EvalRequest::Fingerprint {
@@ -336,11 +371,12 @@ fn response_kind(resp: &EvalResponse) -> String {
             format!("FingerprintOk({})", fingerprint.is_some())
         }
         EvalResponse::CheckpointOk => "CheckpointOk".to_string(),
-        EvalResponse::FetchOk { store_path } => format!("FetchOk({store_path})"),
+        EvalResponse::FetchOk { response } => format!("FetchOk({response})"),
         EvalResponse::Stats { delta } => format!("Stats({} thunks)", delta.nr_thunks),
         EvalResponse::NeedsBuild { derived_paths } => {
             format!("NeedsBuild({} paths)", derived_paths.len())
         }
+        EvalResponse::NeedsInput { request } => format!("NeedsInput({request})"),
         EvalResponse::Err { message } => format!("Err({message})"),
     }
 }
@@ -528,6 +564,44 @@ mod tests {
         );
     }
 
+    #[test]
+    fn request_input_returns_the_parent_answer() {
+        let request = r#"{"builtin":"fetchTree","args":{"type":"tarball","url":"https://x"}}"#;
+        let pipe = pipe_over(&[EvalRequest::InputDone {
+            response: Some(r#"{"storePath":"/nix/store/aaaa-source"}"#.into()),
+            error: None,
+        }]);
+
+        let fetched = pipe.request_input(request);
+
+        assert_eq!(
+            fetched,
+            Ok(r#"{"storePath":"/nix/store/aaaa-source"}"#.to_string())
+        );
+        let sent = sent(pipe);
+        assert!(
+            matches!(&sent[..], [EvalResponse::NeedsInput { request: r }] if r == request),
+            "{sent:?}"
+        );
+    }
+
+    #[test]
+    fn a_wrong_answer_to_an_input_request_breaks_the_pipe() {
+        let pipe = pipe_over(&[
+            EvalRequest::BuildDone { error: None },
+            EvalRequest::Shutdown,
+        ]);
+
+        assert_eq!(
+            pipe.request_input("{}"),
+            Err("eval pipe closed while waiting for an input".to_string())
+        );
+        assert!(
+            pipe.read_frame().is_err(),
+            "a broken pipe ends the request loop"
+        );
+    }
+
     #[test]
     fn parse_warnings_keeps_multiline_warning() {
         let captured = "warning: the following insecure packages:\n  - foo-1.2.3\nKnown issues:\n  - CVE-1234\ntrace: unrelated\n";