brokerd: memory calls to memoryd with the session's state; labels only stricter

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)

crates/brokerd/src/config.rs

@@ -53,6 +53,14 @@ pub struct Gateway {
     pub socket: PathBuf,
 }
 
+/// `[memory]`: where `memoryd` takes the decided `recall` and `memory` calls. The memory tools are
+/// always on, so the section is optional; an empty path means the default under `home`.
+#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Default)]
+#[serde(deny_unknown_fields, default)]
+pub struct Memory {
+    pub socket: PathBuf,
+}
+
 fn default_podman() -> PathBuf {
     PathBuf::from("podman")
 }
@@ -141,6 +149,8 @@ pub struct Config {
     pub runner: Option<Runner>,
     #[serde(default)]
     pub gateway: Option<Gateway>,
+    #[serde(default)]
+    pub memory: Memory,
 }
 
 /// The longest an approval may wait: a day, the longest `loopd` waits after a pending frame.
@@ -283,6 +293,14 @@ impl Config {
             Some(gateway.socket.clone())
         }
     }
+    /// `memoryd`'s socket for `brokerd`.
+    pub fn memory_socket(&self) -> PathBuf {
+        if self.memory.socket.as_os_str().is_empty() {
+            self.paths.home.join("run/broker-memory/memory.sock")
+        } else {
+            self.memory.socket.clone()
+        }
+    }
     pub fn audit_dir(&self) -> PathBuf {
         self.paths.home.join("audit")
     }

crates/brokerd/src/container.rs

@@ -34,6 +34,9 @@ pub const EGRESS_POLL: Duration = Duration::from_millis(20);
 /// A log sink. Call it as `(self.log)("a line")`.
 pub type Log = Arc<dyn Fn(&str) + Send + Sync>;
 
+/// A memory tool that reached the container runtime: fixed, it reaches the model.
+pub const IN_MEMORYD: &str = "the memory tools run in memoryd, not in a container";
+
 pub struct Podman {
     runner: Runner,
     egress_dir: PathBuf,
@@ -240,6 +243,10 @@ impl Runtime for Podman {
     /// an egress starts the proxy first and gives the tool the directory. `run_container` is step 3
     /// to 7 for either. Already written.
     fn run(&self, spec: &RunSpec) -> Result<RunOutput, RunError> {
+        // `Routed` sends the memory tools to `memoryd`; one here is refused, never run.
+        if spec.tool().is_memory() {
+            return Err(RunError::Unavailable(IN_MEMORYD.to_string()));
+        }
         let n = self.next.fetch_add(1, Ordering::SeqCst);
         let name = podman::container_name(spec.session(), spec.call(), n);
         let input = spec.arguments().canonical_json();

crates/brokerd/src/lib.rs

@@ -11,6 +11,7 @@ pub mod container;
 pub mod courier;
 pub mod grants;
 pub mod ledger;
+pub mod memory_port;
 pub mod pipes;
 pub mod podman;
 pub mod policy;

crates/brokerd/src/main.rs

@@ -8,6 +8,7 @@ use std::sync::Arc;
 use brokerd::audit::{AuditError, RECOVERED_NOTICE};
 use brokerd::config::Config;
 use brokerd::container::Podman;
+use brokerd::memory_port::{MEMORY_WAIT, MemoryPort, Routed};
 use brokerd::runner::{Refusing, Runtime};
 use brokerd::serve::{self, ServeError};
 
@@ -66,6 +67,12 @@ fn main() -> ExitCode {
         }
     }
 
+    // The memory tools go to `memoryd`, whatever runs the others (M5a spec, section 8).
+    let runtime: Box<dyn Runtime> = Box::new(Routed {
+        memory: MemoryPort::new(cfg.memory_socket(), MEMORY_WAIT),
+        other: runtime,
+    });
+
     let started = match serve::start(cfg, accept_break, runtime, log) {
         Ok(started) => started,
         Err(ServeError::Audit(e @ AuditError::NothingToAccept)) => {

crates/brokerd/src/memory_port.rs

@@ -0,0 +1,119 @@
+//! Where decided `recall` and `memory` calls go: to `memoryd` over its `brokerd` socket, with the
+//! session's state, instead of a container (M5a spec, section 8). `Routed` sends every other call
+//! to the configured runtime.
+
+use std::os::unix::net::UnixStream;
+use std::path::PathBuf;
+use std::time::Duration;
+
+use proto::{
+    Envelope, MemoryCall, Message, PROTOCOL_VERSION, ToolRequest, ToolResponse, read_frame,
+    write_frame,
+};
+
+use crate::policy::Label;
+use crate::runner::{RunError, RunOutput, RunSpec, Runtime};
+
+/// The sentence when `memoryd` cannot be reached or answers wrongly: fixed, it reaches the model.
+pub const UNREACHABLE: &str =
+    "memory is unavailable: memoryd did not answer (see docs/runbook.md#memoryd-unreachable)";
+
+/// How long `brokerd` waits for `memoryd` to answer one call.
+pub const MEMORY_WAIT: Duration = Duration::from_secs(30);
+
+/// A client of `memoryd`'s `brokerd` socket. One connection per call.
+pub struct MemoryPort {
+    socket: PathBuf,
+    wait: Duration,
+}
+
+impl MemoryPort {
+    pub fn new(socket: PathBuf, wait: Duration) -> MemoryPort {
+        MemoryPort { socket, wait }
+    }
+
+    fn ask(&self, spec: &RunSpec) -> Option<ToolResponse> {
+        // `None` at the first thing that fails. 1. A MemoryCall: request with the spec's session,
+        // call, `spec.tool().as_str()`, and `spec.arguments().canonical_json()`; taint and
+        // untrusted from `spec.state()`. 2. `UnixStream::connect(&self.socket)`, then read and
+        // write timeouts of `self.wait`. 3. `write_frame` an Envelope { v: PROTOCOL_VERSION, id:
+        // spec.call().0, final: true, msg: Message::MemoryCall(call) }. 4. `read_frame`; a
+        // Message::ToolResponse is Some(it), any other message None.
+        let call = MemoryCall {
+            request: ToolRequest {
+                session: spec.session().clone(),
+                call: spec.call(),
+                tool: spec.tool().as_str().to_string(),
+                arguments: spec.arguments().canonical_json(),
+            },
+            taint: spec.state().taint,
+            untrusted: spec.state().untrusted,
+        };
+        let mut stream = UnixStream::connect(&self.socket).ok()?;
+        stream.set_read_timeout(Some(self.wait)).ok()?;
+        stream.set_write_timeout(Some(self.wait)).ok()?;
+        write_frame(
+            &mut stream,
+            &Envelope {
+                v: PROTOCOL_VERSION,
+                id: spec.call().0,
+                r#final: true,
+                msg: Message::MemoryCall(call),
+            },
+        )
+        .ok()?;
+        let env = read_frame(&mut stream).ok()?;
+        match env.msg {
+            Message::ToolResponse(response) => Some(response),
+            _ => None,
+        }
+    }
+}
+
+impl Runtime for MemoryPort {
+    fn run(&self, spec: &RunSpec) -> Result<RunOutput, RunError> {
+        self.run_labelled(spec).map(|(output, _)| output)
+    }
+
+    fn run_labelled(&self, spec: &RunSpec) -> Result<(RunOutput, Option<Label>), RunError> {
+        // `self.ask(spec)`: Some(Result { content, class, untrusted, truncated }) -> Ok((RunOutput
+        // { content, truncated }, Some(Label { class, untrusted }))); Some(Failed { message }) ->
+        // Err(RunError::Failed(message)); anything else, None included ->
+        // Err(RunError::Unavailable(UNREACHABLE.to_string())).
+        match self.ask(spec) {
+            Some(ToolResponse::Result {
+                content,
+                class,
+                untrusted,
+                truncated,
+            }) => Ok((
+                RunOutput { content, truncated },
+                Some(Label { class, untrusted }),
+            )),
+            Some(ToolResponse::Failed { message }) => Err(RunError::Failed(message)),
+            _ => Err(RunError::Unavailable(UNREACHABLE.to_string())),
+        }
+    }
+}
+
+/// The memory tools to `memoryd`, everything else to `other`.
+pub struct Routed {
+    pub memory: MemoryPort,
+    pub other: Box<dyn Runtime>,
+}
+
+impl Runtime for Routed {
+    fn run(&self, spec: &RunSpec) -> Result<RunOutput, RunError> {
+        self.run_labelled(spec).map(|(output, _)| output)
+    }
+
+    fn run_labelled(&self, spec: &RunSpec) -> Result<(RunOutput, Option<Label>), RunError> {
+        // `spec.tool().is_memory()` -> `self.memory.run_labelled(spec)`; otherwise
+        // `self.other.run_labelled(spec)`.
+        if spec.tool().is_memory() {
+            self.memory.run_labelled(spec)
+        } else {
+            self.other.run_labelled(spec)
+        }
+    }
+}

crates/brokerd/src/policy/mod.rs

@@ -151,6 +151,8 @@ pub struct Decision {
     request: ToolRequest,
     args: ToolArgs,
     matched: Matched,
+    /// The session's state when the call was decided; `memoryd` needs it (M5a spec, section 8).
+    state: SessionState,
 }
 
 /// A call the broker left for the owner: built only in this module, re-decided by `redecide`.
@@ -165,6 +167,9 @@ impl Decision {
     pub fn request(&self) -> &ToolRequest {
         &self.request
     }
+    pub fn state(&self) -> SessionState {
+        self.state
+    }
     pub fn args(&self) -> &ToolArgs {
         &self.args
     }
@@ -278,6 +283,7 @@ pub fn decide(
             request,
             args,
             matched,
+            state,
         }),
     }
 }
@@ -315,6 +321,7 @@ pub fn redecide(
             request: ask.request().clone(),
             args: ask.args().clone(),
             matched,
+            state,
         }),
     }
 }

crates/brokerd/src/runner.rs

@@ -21,7 +21,7 @@
 //! ```
 
 use crate::args::{ToolArgs, ToolName};
-use crate::policy::Decision;
+use crate::policy::{Decision, Label, SessionState};
 use proto::{CallId, SessionId, ToolResponse};
 
 /// A directory mounted for one call: a path and whether the runtime may write to it.
@@ -42,6 +42,7 @@ pub struct RunSpec {
     arguments: ToolArgs,
     mounts: Vec<Mount>,
     egress: Option<Vec<String>>,
+    state: SessionState,
 }
 
 impl RunSpec {
@@ -63,6 +64,10 @@ impl RunSpec {
     pub fn egress(&self) -> Option<&[String]> {
         self.egress.as_deref()
     }
+    /// The session's state when the call was decided.
+    pub fn state(&self) -> SessionState {
+        self.state
+    }
 }
 
 /// A `Runtime`'s answer to one call. Its text reaches the model as a `failed` result labelled
@@ -83,6 +88,13 @@ pub struct RunOutput {
 /// Runs one call. `Send + Sync` so one `Runtime` can serve several brokers at once.
 pub trait Runtime: Send + Sync {
     fn run(&self, spec: &RunSpec) -> Result<RunOutput, RunError>;
+
+    /// `run`, and a label for what came back when the runtime knows one (`memoryd` does: a
+    /// `recall` hit from an untrusted session). The runner keeps the stricter of this and the
+    /// grant's label. Container runtimes know none.
+    fn run_labelled(&self, spec: &RunSpec) -> Result<(RunOutput, Option<Label>), RunError> {
+        self.run(spec).map(|output| (output, None))
+    }
 }
 
 /// The production runtime for M3a. It runs nothing; every call is `Unavailable`.
@@ -126,14 +138,25 @@ pub fn run(decision: Decision, runtime: &dyn Runtime) -> proto::ToolResponse {
         arguments: args.clone(),
         mounts,
         egress,
+        state: decision.state(),
     };
-    match runtime.run(&spec) {
-        Ok(output) => ToolResponse::Result {
-            content: output.content,
-            class: label.class,
-            untrusted: label.untrusted,
-            truncated: output.truncated,
-        },
+    match runtime.run_labelled(&spec) {
+        Ok((output, found)) => {
+            // Labels only get stricter: the higher class, and untrusted if either says so.
+            let (class, untrusted) = match found {
+                Some(found) => (
+                    label.class.max(found.class),
+                    label.untrusted || found.untrusted,
+                ),
+                None => (label.class, label.untrusted),
+            };
+            ToolResponse::Result {
+                content: output.content,
+                class,
+                untrusted,
+                truncated: output.truncated,
+            }
+        }
         Err(RunError::Failed(message) | RunError::Unavailable(message)) => {
             ToolResponse::Failed { message }
         }