brokerd: consult config, secrets, loopback URL

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>

crates/brokerd/Cargo.toml

@@ -9,7 +9,7 @@ publish.workspace = true
 workspace = true
 
 [dependencies]
-bxcore.workspace = true
+bxcore = { workspace = true, features = ["secrets"] }
 serde.workspace = true
 serde_json.workspace = true
 toml.workspace = true

crates/brokerd/src/config.rs

@@ -5,8 +5,11 @@
 //! owner believes is set and is not.
 
 use serde::Deserialize;
+use std::collections::BTreeMap;
 use std::path::{Path, PathBuf};
 
+use bxcore::secrets::SecretSpec;
+
 #[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
 #[serde(deny_unknown_fields, default)]
 pub struct Paths {
@@ -155,6 +158,10 @@ pub struct Config {
     pub gateway: Option<Gateway>,
     #[serde(default)]
     pub memory: Memory,
+    #[serde(default)]
+    pub consult: Option<crate::consult::Consult>,
+    #[serde(default)]
+    pub secrets: BTreeMap<String, SecretSpec>,
 }
 
 /// The longest an approval may wait: a day, the longest `loopd` waits after a pending frame.
@@ -272,6 +279,11 @@ impl Config {
                 ));
             }
         }
+        if let Some(consult) = &config.consult
+            && let Err(why) = crate::consult::validate(consult)
+        {
+            return Err(ConfigError::Invalid(path.to_path_buf(), why));
+        }
         Ok(config)
     }
     pub fn broker_socket(&self) -> PathBuf {

crates/brokerd/src/consult.rs

@@ -0,0 +1,188 @@
+//! `consult`: an allowlisted chat-completions POST decided by the grant's `max_taint`.
+//! Nothing here names a vendor (M6 spec, section 4).
+
+use std::ffi::OsString;
+use std::time::Duration;
+
+use bxcore::secrets::{Loaded, SecretError, SecretSpec};
+use serde::{Deserialize, Deserializer};
+
+use crate::config::Config;
+
+pub const CONSULT: &str = "consult";
+/// Default wait for the remote completion, milliseconds.
+pub const DEFAULT_TIMEOUT_MS: u64 = 180_000;
+/// Default sampling temperature as thousandths (0.6).
+pub const DEFAULT_TEMPERATURE_MILLI: u16 = 600;
+/// Start failure when the named secret cannot be read.
+pub const SECRET_POINTER: &str = "see docs/runbook.md#consult-secret-unreadable";
+
+/// `[consult]` in `brokerd.toml`. Unknown keys are errors.
+#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Consult {
+    pub url: String,
+    pub model: String,
+    #[serde(
+        rename = "temperature",
+        default = "default_temperature_milli",
+        deserialize_with = "de_temperature_milli"
+    )]
+    pub temperature_milli: u16,
+    #[serde(default = "default_timeout_ms")]
+    pub timeout_ms: u64,
+    pub secret: String,
+}
+
+fn default_temperature_milli() -> u16 {
+    DEFAULT_TEMPERATURE_MILLI
+}
+
+fn default_timeout_ms() -> u64 {
+    DEFAULT_TIMEOUT_MS
+}
+
+fn de_temperature_milli<'de, D: Deserializer<'de>>(deserializer: D) -> Result<u16, D::Error> {
+    #[derive(Deserialize)]
+    #[serde(untagged)]
+    enum Num {
+        F(f64),
+        I(i64),
+    }
+    let v = match Num::deserialize(deserializer).map_err(serde::de::Error::custom)? {
+        Num::F(f) => f,
+        Num::I(0) => 0.0,
+        Num::I(_) => {
+            return Err(serde::de::Error::custom(
+                "[consult] temperature must be from 0 to 0.6",
+            ));
+        }
+    };
+    if !v.is_finite() || v < 0.0 || v > 0.6 {
+        return Err(serde::de::Error::custom(
+            "[consult] temperature must be from 0 to 0.6",
+        ));
+    }
+    let milli = (v * 1000.0).round();
+    let text = format!("{milli:.0}");
+    text.parse::<u16>().map_err(serde::de::Error::custom)
+}
+
+impl Consult {
+    /// The temperature written in the JSON body.
+    pub fn temperature(&self) -> f64 {
+        f64::from(self.temperature_milli) / 1000.0
+    }
+
+    pub fn timeout(&self) -> Duration {
+        Duration::from_millis(self.timeout_ms)
+    }
+}
+
+/// Why a `[consult]` table cannot be used.
+pub fn validate(consult: &Consult) -> Result<(), String> {
+    parse_loopback_http(&consult.url).map(|_| ())?;
+    if consult.model.is_empty() {
+        return Err("[consult] model must not be empty".to_string());
+    }
+    if consult.timeout_ms == 0 {
+        return Err("[consult] timeout_ms must be at least 1".to_string());
+    }
+    if consult.secret.is_empty() {
+        return Err("[consult] secret must not be empty".to_string());
+    }
+    if consult.temperature_milli > DEFAULT_TEMPERATURE_MILLI {
+        return Err("[consult] temperature must be from 0 to 0.6".to_string());
+    }
+    Ok(())
+}
+
+/// Host, port and path of a loopback `http://` URL. M6 stays on loopback.
+pub fn parse_loopback_http(url: &str) -> Result<(String, u16, String), String> {
+    let rest = match url.strip_prefix("http://") {
+        Some(rest) => rest,
+        None => {
+            return Err("[consult] url must be http://127.0.0.1 or http://localhost".to_string());
+        }
+    };
+    if rest.contains('@') {
+        return Err("[consult] url must not include a userinfo".to_string());
+    }
+    let (hostport, path) = match rest.find('/') {
+        Some(i) => match rest.split_at_checked(i) {
+            Some(pair) => pair,
+            None => {
+                return Err("[consult] url is not a valid http URL".to_string());
+            }
+        },
+        None => (rest, "/"),
+    };
+    let path = if path.is_empty() { "/" } else { path };
+    let (host, port) = match hostport.split_once(':') {
+        Some((host, port)) => {
+            let port: u16 = port
+                .parse()
+                .map_err(|_| "[consult] url port is not a number".to_string())?;
+            if port == 0 {
+                return Err("[consult] url port must be from 1 to 65535".to_string());
+            }
+            (host, port)
+        }
+        None => (hostport, 80),
+    };
+    if host != "127.0.0.1" && host != "localhost" {
+        return Err("[consult] url must be http://127.0.0.1 or http://localhost".to_string());
+    }
+    Ok((host.to_string(), port, path.to_string()))
+}
+
+/// Why `brokerd` will not start with `[consult]` set.
+#[derive(Debug)]
+pub enum SecretStartError {
+    Missing(String),
+    Spec(String, String),
+    Load(SecretError),
+}
+
+impl std::fmt::Display for SecretStartError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            SecretStartError::Missing(name) => {
+                write!(f, "secret {name} is not in [secrets]; {SECRET_POINTER}")
+            }
+            SecretStartError::Spec(name, why) => {
+                write!(f, "[secrets.{name}]: {why}; {SECRET_POINTER}")
+            }
+            SecretStartError::Load(err) => write!(f, "{err}\n{SECRET_POINTER}"),
+        }
+    }
+}
+
+impl std::error::Error for SecretStartError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            SecretStartError::Load(err) => Some(err),
+            SecretStartError::Missing(_) | SecretStartError::Spec(_, _) => None,
+        }
+    }
+}
+
+/// Load the bearer for `[consult]`, or `Ok(None)` when the table is absent.
+pub fn load_secret(
+    cfg: &Config,
+    env: &dyn Fn(&str) -> Option<OsString>,
+) -> Result<Option<Loaded>, SecretStartError> {
+    let Some(consult) = &cfg.consult else {
+        return Ok(None);
+    };
+    let spec: &SecretSpec = match cfg.secrets.get(&consult.secret) {
+        Some(spec) => spec,
+        None => return Err(SecretStartError::Missing(consult.secret.clone())),
+    };
+    let source = spec
+        .source()
+        .map_err(|why| SecretStartError::Spec(consult.secret.clone(), why))?;
+    let loaded = bxcore::secrets::load("brokerd", &consult.secret, &source, env)
+        .map_err(SecretStartError::Load)?;
+    Ok(Some(loaded))
+}

crates/brokerd/src/lib.rs

@@ -7,6 +7,7 @@ pub mod audit;
 pub mod broker;
 pub mod builtin;
 pub mod config;
+pub mod consult;
 pub mod container;
 pub mod courier;
 pub mod grants;

crates/brokerd/tests/config_consult.rs

@@ -0,0 +1,84 @@
+//! `[consult]` in `brokerd.toml`. Do not edit.
+
+#[path = "support/tmp.rs"]
+mod tmp;
+
+use brokerd::config::{Config, ConfigError};
+use brokerd::consult::{DEFAULT_TEMPERATURE_MILLI, DEFAULT_TIMEOUT_MS};
+use tmp::TempDir;
+
+fn load(text: &str) -> Result<Config, ConfigError> {
+    let dir = TempDir::new("cfg-consult");
+    let path = dir.write("brokerd.toml", text);
+    Config::load(&path)
+}
+
+const TABLE: &str = r#"
+[consult]
+url = "http://127.0.0.1:9/v1/chat/completions"
+model = "test-model"
+secret = "consult"
+"#;
+
+#[test]
+fn no_consult_table_means_none() {
+    let c = load("").unwrap();
+    assert!(c.consult.is_none());
+    assert!(c.secrets.is_empty());
+}
+
+#[test]
+fn a_consult_table_loads_defaults() {
+    let c = load(TABLE).unwrap();
+    let consult = c.consult.expect("consult");
+    assert_eq!(consult.url, "http://127.0.0.1:9/v1/chat/completions");
+    assert_eq!(consult.model, "test-model");
+    assert_eq!(consult.secret, "consult");
+    assert_eq!(consult.temperature_milli, DEFAULT_TEMPERATURE_MILLI);
+    assert_eq!(consult.timeout_ms, DEFAULT_TIMEOUT_MS);
+    assert!((consult.temperature() - 0.6).abs() < 1e-9);
+}
+
+#[test]
+fn temperature_above_zero_point_six_is_invalid() {
+    let text = format!("{TABLE}temperature = 0.7\n");
+    match load(&text) {
+        Err(ConfigError::Parse(_, err)) => {
+            let msg = err.to_string();
+            assert!(msg.contains("0.6"), "{msg}");
+        }
+        other => panic!("{other:?}"),
+    }
+}
+
+#[test]
+fn a_non_loopback_url_is_invalid() {
+    let text = r#"
+[consult]
+url = "https://example.com/v1/chat/completions"
+model = "x"
+secret = "consult"
+"#;
+    match load(text) {
+        Err(ConfigError::Invalid(_, why)) => {
+            assert!(
+                why.contains("http://127.0.0.1") || why.contains("localhost"),
+                "{why}"
+            );
+        }
+        other => panic!("{other:?}"),
+    }
+}
+
+#[test]
+fn unknown_consult_keys_are_errors() {
+    let text = format!("{TABLE}vendor = \"x\"\n");
+    assert!(matches!(load(&text), Err(ConfigError::Parse(_, _))));
+}
+
+#[test]
+fn a_secret_spec_loads_under_secrets() {
+    let text = format!("{TABLE}\n[secrets.consult]\nenv = \"CONSULT_TOKEN\"\n");
+    let c = load(&text).unwrap();
+    assert!(c.secrets.contains_key("consult"));
+}

crates/brokerd/tests/consult_secret.rs

@@ -0,0 +1,82 @@
+//! `[consult]` without a readable secret: `brokerd` does not start. Do not edit.
+
+#[path = "support/tmp.rs"]
+mod tmp;
+
+use std::ffi::OsString;
+
+use brokerd::config::Config;
+use brokerd::consult::{self, SECRET_POINTER};
+use tmp::TempDir;
+
+#[test]
+fn a_missing_secrets_entry_is_consult_secret_unreadable() {
+    let dir = TempDir::new("consult-secret-missing");
+    let path = dir.write(
+        "brokerd.toml",
+        r#"
+[consult]
+url = "http://127.0.0.1:9/v1/chat/completions"
+model = "test-model"
+secret = "consult"
+"#,
+    );
+    let cfg = Config::load(&path).unwrap();
+    let err = consult::load_secret(&cfg, &|_| None).unwrap_err();
+    let text = err.to_string();
+    assert!(text.contains(SECRET_POINTER), "{text}");
+    assert!(text.contains("consult"), "{text}");
+}
+
+#[test]
+fn an_unreadable_env_secret_names_both_pointers() {
+    let dir = TempDir::new("consult-secret-env");
+    let path = dir.write(
+        "brokerd.toml",
+        r#"
+[consult]
+url = "http://127.0.0.1:9/v1/chat/completions"
+model = "test-model"
+secret = "consult"
+
+[secrets.consult]
+env = "CONSULT_TOKEN_FOR_THIS_TEST_UNSET"
+"#,
+    );
+    let cfg = Config::load(&path).unwrap();
+    let err = consult::load_secret(&cfg, &|_| None).unwrap_err();
+    let text = err.to_string();
+    assert!(text.contains(SECRET_POINTER), "{text}");
+    assert!(
+        text.contains("docs/runbook.md#secret-unavailable"),
+        "{text}"
+    );
+}
+
+#[test]
+fn a_present_env_secret_loads() {
+    let dir = TempDir::new("consult-secret-ok");
+    let path = dir.write(
+        "brokerd.toml",
+        r#"
+[consult]
+url = "http://127.0.0.1:9/v1/chat/completions"
+model = "test-model"
+secret = "consult"
+
+[secrets.consult]
+env = "CONSULT_TOKEN_FOR_THIS_TEST_SET"
+"#,
+    );
+    let cfg = Config::load(&path).unwrap();
+    let loaded = consult::load_secret(&cfg, &|k| {
+        if k == "CONSULT_TOKEN_FOR_THIS_TEST_SET" {
+            Some(OsString::from("bearer-value"))
+        } else {
+            None
+        }
+    })
+    .unwrap()
+    .expect("loaded");
+    assert_eq!(loaded.secret.expose(), "bearer-value");
+}

crates/brokerd/tests/support/rig.rs

@@ -43,6 +43,8 @@ impl Rig {
             runner: None,
             gateway: None,
             memory: Default::default(),
+            consult: None,
+            secrets: Default::default(),
         };
         Rig {
             dir,