From 89440418798328e6143552ca7cccbdc8ce60fd33 Mon Sep 17 00:00:00 2001 From: Jordan Mecom Date: Tue, 18 Aug 2026 16:51:52 -0700 Subject: [PATCH] Expose IFC policy to external agent harnesses Signed-off-by: Jordan Mecom --- ARCHITECTURE.md | 7 +- Cargo.lock | 18 + Cargo.toml | 1 + README.md | 2 +- crates/buzz-acp/README.md | 4 +- crates/buzz-ifc-broker/Cargo.toml | 22 + crates/buzz-ifc-broker/README.md | 77 ++ crates/buzz-ifc-broker/src/lib.rs | 842 ++++++++++++++++++ crates/buzz-ifc-broker/src/main.rs | 69 ++ crates/buzz-ifc-broker/tests/conformance.rs | 22 + .../tests/fixtures/domain-golden.json | 37 + crates/buzz-ifc-broker/tests/stdio.rs | 35 + 12 files changed, 1131 insertions(+), 5 deletions(-) create mode 100644 crates/buzz-ifc-broker/Cargo.toml create mode 100644 crates/buzz-ifc-broker/README.md create mode 100644 crates/buzz-ifc-broker/src/lib.rs create mode 100644 crates/buzz-ifc-broker/src/main.rs create mode 100644 crates/buzz-ifc-broker/tests/conformance.rs create mode 100644 crates/buzz-ifc-broker/tests/fixtures/domain-golden.json create mode 100644 crates/buzz-ifc-broker/tests/stdio.rs diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 84913e1d65..a44a360ad2 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -87,7 +87,8 @@ buzz-core (zero I/O — types, verification, filter matching, kind registry) └── buzz-relay (ties everything together — the server) buzz-ifc (zero I/O — execution-domain derivation and IFC rules) - └── buzz-acp (relay/ACP adapter and worker routing) + ├── buzz-acp (relay/ACP adapter and worker routing) + └── buzz-ifc-broker (stateful JSON-RPC policy process for non-Rust harnesses) buzz-sdk (typed Nostr event builders — used by buzz-acp and buzz-cli) buzz-media (Blossom/S3 media storage) buzz-cli (agent-first CLI) @@ -677,11 +678,11 @@ Buzz Relay ──WS──→ buzz-acp ──stdio (ACP/JSON-RPC)──→ Agent --- -### buzz-ifc — Shared Agent Policy +### buzz-ifc and buzz-ifc-broker — Shared Agent Policy `buzz-ifc` is a zero-I/O policy crate. Given facts already verified by a trusted Buzz adapter, it derives the invocation's audience, retained-state context, membership epoch, and effective capabilities. It also evaluates reads, calls, publications, and process reuse while retaining a conservative label for each process. -`buzz-acp` links the crate directly. The surrounding adapter remains responsible for signed-event and membership verification, process lifecycle, credential mediation, and OS or VM confinement for confidential domains. +`buzz-acp` links the crate directly. Other harnesses can run `buzz-ifc-broker` and use its bounded JSON-RPC protocol over stdio. Despite the executable name, this is the policy process inside a larger trusted agent gateway, not the complete product broker. It retains policy state for workers on that connection. The surrounding adapter remains responsible for signed-event and membership verification, process lifecycle, credential mediation, and OS or VM confinement for confidential domains. --- diff --git a/Cargo.lock b/Cargo.lock index 41316db889..bffd39314e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1132,6 +1132,24 @@ dependencies = [ "uuid", ] +[[package]] +name = "buzz-ifc-broker" +version = "0.1.0" +dependencies = [ + "anyhow", + "buzz-ifc", + "clap", + "futures-util", + "hex", + "serde", + "serde_json", + "tokio", + "tokio-util", + "tracing", + "tracing-subscriber", + "uuid", +] + [[package]] name = "buzz-media" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 3960338be6..9571bd8965 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,6 +11,7 @@ members = [ "crates/buzz-audit", "crates/buzz-acp", "crates/buzz-ifc", + "crates/buzz-ifc-broker", "crates/buzz-agent", "crates/sprig", "crates/buzz-test-client", diff --git a/README.md b/README.md index c165d4027f..efd23ec8b0 100644 --- a/README.md +++ b/README.md @@ -227,7 +227,7 @@ A Rust workspace of focused crates. Single source of truth: the relay. See [ARCH **Services** — `buzz-db` (Postgres) · `buzz-auth` (NIP-42/98 Schnorr auth, rate limiting) · `buzz-pubsub` (Redis, presence, typing) · `buzz-search` (Postgres FTS) · `buzz-audit` (hash-chain log). Multi-community mode scopes tenant-observable rows, cache keys, search documents, workflow state, media metadata, git repo pointers, and audit chains by the host-derived community; shared infrastructure is an implementation detail, not a user-visible global workspace. -**Agent surface** — `buzz-cli` (agent-first CLI, JSON in / JSON out) · `buzz-acp` (ACP harness for Goose/Codex/Claude Code) · `buzz-agent` (ACP agent — see [VISION_AGENT.md](VISION_AGENT.md)) · `buzz-dev-mcp` (shell + file-edit tools) · `buzz-ifc` (shared audience-scoped agent policy) · `buzz-workflow` (YAML automation) · `buzz-persona` (agent persona packs) +**Agent surface** — `buzz-cli` (agent-first CLI, JSON in / JSON out) · `buzz-acp` (ACP harness for Goose/Codex/Claude Code) · `buzz-agent` (ACP agent — see [VISION_AGENT.md](VISION_AGENT.md)) · `buzz-dev-mcp` (shell + file-edit tools) · `buzz-ifc` / `buzz-ifc-broker` (shared audience-scoped agent policy) · `buzz-workflow` (YAML automation) · `buzz-persona` (agent persona packs) **Git & pairing** — `git-sign-nostr` / `git-credential-nostr` (nostr-signed git) · `buzz-pair-relay` / `buzz-pairing-cli` (relay pairing) diff --git a/crates/buzz-acp/README.md b/crates/buzz-acp/README.md index 727e0b93cc..60dfe71f30 100644 --- a/crates/buzz-acp/README.md +++ b/crates/buzz-acp/README.md @@ -153,7 +153,9 @@ state. Its logs report those remaining gaps explicitly. The deterministic label, domain, capability, reuse, and confinement rules live in the runtime-independent `buzz-ifc` crate. `buzz-acp` supplies the trusted Buzz-specific half: signed-event admission, relay-signed membership resolution, -and ACP worker replacement. +and ACP worker replacement. A non-Rust harness can use the same rules through +the bounded JSON-RPC interface provided by `buzz-ifc-broker`; see that crate's +README and conformance fixture for the wire contract. ### Parallel Agents & Heartbeat diff --git a/crates/buzz-ifc-broker/Cargo.toml b/crates/buzz-ifc-broker/Cargo.toml new file mode 100644 index 0000000000..7901ea437f --- /dev/null +++ b/crates/buzz-ifc-broker/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "buzz-ifc-broker" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +description = "Language-neutral policy broker for audience-scoped Buzz agents" + +[dependencies] +anyhow = { workspace = true } +buzz-ifc = { workspace = true } +clap = { version = "4", features = ["derive", "env"] } +futures-util = { workspace = true } +hex = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +tokio = { workspace = true, features = ["io-std"] } +tokio-util = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } +uuid = { workspace = true } diff --git a/crates/buzz-ifc-broker/README.md b/crates/buzz-ifc-broker/README.md new file mode 100644 index 0000000000..f369d3b6f4 --- /dev/null +++ b/crates/buzz-ifc-broker/README.md @@ -0,0 +1,77 @@ +# buzz-ifc-broker + +`buzz-ifc-broker` exposes the shared `buzz-ifc` policy over bounded JSON-RPC +2.0 on stdin/stdout. It exists for trusted agent harnesses that cannot link the +Rust crate directly, such as a Go kgoose integration. + +The broker owns process-level IFC state. The caller supplies facts only after +it has verified the triggering event and authoritative Buzz membership. The +caller does not choose an audience or context: `buzz-ifc` derives both from the +verified conversation kind, roster, requesters, executing agent, and owner. + +The agent runtime must not receive raw Buzz credentials or another route around +the trusted adapter. Despite its executable name, this is the policy process +inside the agent gateway, not the complete product broker. It does not launch, +sandbox, or terminate workers and cannot force its caller to obey a decision. + +Every derived domain includes a compartment profile: + +- `shared_public` may reuse the realm's normal public worker and public state. +- `domain_confined` requires a worker, writable state, and output paths dedicated + to that exact restricted or owner-private domain. + +The trusted adapter implements that placement. A local sandbox, container, or VM +supplies the final confinement boundary for `domain_confined` work. + +Run in compatibility mode: + +```bash +buzz-ifc-broker --mode audit +``` + +Run when the trusted adapter will treat denied decisions as blocking: + +```bash +buzz-ifc-broker --mode enforce +``` + +Each request and response is one JSON line. Start by binding a concrete worker +to the domain derived from verified invocation facts: + +```json +{"jsonrpc":"2.0","id":1,"method":"worker/enter","params":{"worker_id":"kgoose-session-42","invocation":{"realm_url":"wss://buzz.example","channel_id":"00000000-0000-0000-0000-000000000001","conversation_kind":"restricted","epoch":"membership:","members":["","",""],"executing_agent":"","requesters":[""],"owner":"","bot_capabilities":["buzz.read.current","buzz.publish.current","email.read"],"conversation_capabilities":["buzz.read.current","buzz.publish.current"]}}} +``` + +The result includes the opaque `domain_id`; `details.replace_worker` tells the +adapter when the named process already contains another domain. The adapter +must retire that process before delivering the new request. + +Call `worker/observe` before labeled data enters the process and `worker/call` +before invoking a mediated operation. If audit mode proceeds after a denied +call, its result must still be reported with `worker/observe`. Before Buzz signs +or sends a response, call `worker/publish` with the exact content digest and the +actual destination. Without a verified declassification grant, output is bound +to the source context. The current protocol intentionally exposes no +declassification method. + +Supported methods: + +- `broker/info` +- `domain/derive` +- `worker/enter` +- `worker/observe` +- `worker/call` +- `worker/publish` +- `worker/retire` + +Logs are written to stderr. Stdout contains protocol frames only. + +The broker connection is part of the worker lifecycle. `worker/retire` removes +policy state only after the adapter has terminated the corresponding process. +If the broker exits or the connection is replaced, the adapter must terminate +all workers tracked by that broker. A fresh broker has no evidence that an +existing process is safe to reuse. + +`tests/fixtures/domain-golden.json` is the language-neutral conformance vector. +Adapters should replay it in their own test suite rather than duplicating the +domain hashing algorithm. diff --git a/crates/buzz-ifc-broker/src/lib.rs b/crates/buzz-ifc-broker/src/lib.rs new file mode 100644 index 0000000000..9bbbd52d0a --- /dev/null +++ b/crates/buzz-ifc-broker/src/lib.rs @@ -0,0 +1,842 @@ +//! Stateful, language-neutral access to [`buzz_ifc`] policy decisions. +//! +//! The broker accepts bounded JSON-RPC requests from a trusted Buzz adapter. +//! That adapter remains responsible for authenticating events and resolving +//! authoritative membership. The broker derives the execution domain from +//! those verified facts, retains process-level confinement state, and returns +//! decisions that a local ACP bridge or remote harness can enforce consistently. + +use std::collections::{BTreeSet, HashMap}; +use std::fmt; + +use buzz_ifc::{ + derive_execution_domain, CapabilityPolicy, CapabilitySet, ConfidentialityLabel, + ConversationKind, DomainContext, DomainFacts, ExecutionDomain, MembershipEpoch, Principal, + ProcessState, RealmId, ResourceLabel, RuleDecision, RuleEvaluator, +}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use uuid::Uuid; + +/// Version of the JSON-RPC contract implemented by this broker. +pub const PROTOCOL_VERSION: &str = "buzz-ifc-broker/1"; + +/// Whether denied decisions are observational or enforced by the caller. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum EnforcementMode { + /// Return and log decisions while conservatively tracking inputs that + /// entered the worker despite a denial. + Audit, + /// Require the caller to stop denied reads, calls, publications, and + /// cross-domain worker reuse. + Enforce, +} + +impl EnforcementMode { + fn enforced(self) -> bool { + matches!(self, Self::Enforce) + } +} + +impl fmt::Display for EnforcementMode { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Audit => f.write_str("audit"), + Self::Enforce => f.write_str("enforce"), + } + } +} + +/// Stateful policy broker. One instance serves one trusted adapter stream. +pub struct Broker { + mode: EnforcementMode, + workers: HashMap, +} + +impl Broker { + /// Construct an empty broker in the selected enforcement mode. + pub fn new(mode: EnforcementMode) -> Self { + Self { + mode, + workers: HashMap::new(), + } + } + + /// Parse one JSON-RPC line and return one JSON-RPC response line. + pub fn handle_line(&mut self, line: &str) -> String { + let response = match serde_json::from_str::(line) { + Ok(request) => self.handle_request(request), + Err(error) => RpcResponse::error( + Value::Null, + -32700, + "parse error", + Some(json!({ "detail": error.to_string() })), + ), + }; + match serde_json::to_string(&response) { + Ok(encoded) => encoded, + Err(_) => { + "{\"jsonrpc\":\"2.0\",\"id\":null,\"error\":{\"code\":-32603,\"message\":\"response encoding failed\"}}".to_string() + } + } + } + + /// Return a JSON-RPC error for an input frame rejected before JSON parsing. + pub fn frame_error(message: &str) -> String { + let response = RpcResponse::error(Value::Null, -32600, message, None); + match serde_json::to_string(&response) { + Ok(encoded) => encoded, + Err(_) => { + "{\"jsonrpc\":\"2.0\",\"id\":null,\"error\":{\"code\":-32603,\"message\":\"internal error\"}}".to_string() + } + } + } + + fn handle_request(&mut self, request: RpcRequest) -> RpcResponse { + let id = match request.id { + Some(id) => id, + None => { + return RpcResponse::error( + Value::Null, + -32600, + "security decisions require a request id", + None, + ) + } + }; + if request.jsonrpc != "2.0" { + return RpcResponse::error(id, -32600, "jsonrpc must equal 2.0", None); + } + + let method = request.method.clone(); + let outcome = self.dispatch(&request.method, request.params); + match outcome { + Ok(result) => { + tracing::info!( + target: "buzz_ifc_broker::decision", + method, + mode = %self.mode, + outcome = "ok", + "IFC broker request evaluated" + ); + RpcResponse::success(id, result) + } + Err(error) => { + tracing::info!( + target: "buzz_ifc_broker::decision", + method, + mode = %self.mode, + outcome = "error", + code = error.code, + reason = error.message, + "IFC broker request rejected" + ); + RpcResponse::error(id, error.code, error.message, error.data) + } + } + } + + fn dispatch(&mut self, method: &str, params: Value) -> Result { + match method { + "broker/info" => Ok(json!({ + "protocol_version": PROTOCOL_VERSION, + "mode": self.mode.to_string(), + "worker_count": self.workers.len(), + })), + "domain/derive" => { + let params: DeriveParams = parse_params(params)?; + let domain = params.invocation.resolve()?; + Ok(domain_description(&domain)) + } + "worker/enter" => { + let params: EnterParams = parse_params(params)?; + validate_worker_id(¶ms.worker_id)?; + let domain = params.invocation.resolve()?; + let domain_id = domain.id(); + let compartment = domain.compartment_profile().as_str(); + let record = self.workers.entry(params.worker_id.clone()).or_default(); + let decision = record.state.enter(&domain); + let replace_worker = !decision.allowed(); + if replace_worker { + record.state.mark_unknown(); + if self.mode.enforced() { + record.poisoned = true; + } else { + record.domain = Some(domain); + } + } else { + record.domain = Some(domain); + } + Ok(decision_value( + self.mode, + "reuse", + &decision, + Some(¶ms.worker_id), + Some(&domain_id), + json!({ + "replace_worker": replace_worker, + "compartment": compartment, + }), + )) + } + "worker/observe" => { + let params: ObserveParams = parse_params(params)?; + let mode = self.mode; + let record = active_worker_mut(&mut self.workers, ¶ms.worker_id)?; + if record.poisoned { + return Ok(poisoned_decision(mode, ¶ms.worker_id, record)); + } + let domain = record.domain.as_ref().ok_or_else(|| { + BrokerError::state("worker has not entered an execution domain") + })?; + let resource = params.resource.resolve(domain)?; + let decision = RuleEvaluator::read(domain, &resource); + if decision.allowed() || !mode.enforced() { + record.state.observe(&resource); + } + Ok(decision_value( + mode, + "read", + &decision, + Some(¶ms.worker_id), + Some(&domain.id()), + json!({ "source": params.source }), + )) + } + "worker/call" => { + let params: CallParams = parse_params(params)?; + let record = active_worker(&self.workers, ¶ms.worker_id)?; + if record.poisoned { + return Ok(poisoned_decision(self.mode, ¶ms.worker_id, record)); + } + let domain = record.domain.as_ref().ok_or_else(|| { + BrokerError::state("worker has not entered an execution domain") + })?; + let decision = RuleEvaluator::call(domain, ¶ms.operation); + Ok(decision_value( + self.mode, + "call", + &decision, + Some(¶ms.worker_id), + Some(&domain.id()), + json!({ "operation": params.operation }), + )) + } + "worker/publish" => { + let params: PublishParams = parse_params(params)?; + let record = active_worker(&self.workers, ¶ms.worker_id)?; + if record.poisoned { + return Ok(poisoned_decision(self.mode, ¶ms.worker_id, record)); + } + let domain = record.domain.as_ref().ok_or_else(|| { + BrokerError::state("worker has not entered an execution domain") + })?; + let (audience, context) = params.destination.resolve(domain)?; + let digest = parse_digest(¶ms.content_sha256)?; + let decision = record + .state + .publish(domain, &audience, &context, &digest, None); + Ok(decision_value( + self.mode, + "publish", + &decision, + Some(¶ms.worker_id), + Some(&domain.id()), + json!({ "destination": context.kind() }), + )) + } + "worker/retire" => { + let params: WorkerParams = parse_params(params)?; + validate_worker_id(¶ms.worker_id)?; + Ok(json!({ + "worker_id": params.worker_id, + "retired": self.workers.remove(¶ms.worker_id).is_some(), + })) + } + _ => Err(BrokerError::method_not_found()), + } + } +} + +#[derive(Default)] +struct WorkerRecord { + state: ProcessState, + domain: Option, + poisoned: bool, +} + +#[derive(Deserialize)] +struct RpcRequest { + jsonrpc: String, + id: Option, + method: String, + #[serde(default)] + params: Value, +} + +#[derive(Serialize)] +struct RpcResponse { + jsonrpc: &'static str, + id: Value, + #[serde(skip_serializing_if = "Option::is_none")] + result: Option, + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, +} + +impl RpcResponse { + fn success(id: Value, result: Value) -> Self { + Self { + jsonrpc: "2.0", + id, + result: Some(result), + error: None, + } + } + + fn error(id: Value, code: i64, message: &str, data: Option) -> Self { + Self { + jsonrpc: "2.0", + id, + result: None, + error: Some(RpcErrorBody { + code, + message: message.to_string(), + data, + }), + } + } +} + +#[derive(Serialize)] +struct RpcErrorBody { + code: i64, + message: String, + #[serde(skip_serializing_if = "Option::is_none")] + data: Option, +} + +struct BrokerError { + code: i64, + message: &'static str, + data: Option, +} + +impl BrokerError { + fn invalid_params(detail: impl Into) -> Self { + Self { + code: -32602, + message: "invalid params", + data: Some(json!({ "detail": detail.into() })), + } + } + + fn state(detail: impl Into) -> Self { + Self { + code: -32001, + message: "invalid broker state", + data: Some(json!({ "detail": detail.into() })), + } + } + + fn method_not_found() -> Self { + Self { + code: -32601, + message: "method not found", + data: None, + } + } +} + +fn parse_params Deserialize<'de>>(params: Value) -> Result { + serde_json::from_value(params).map_err(|error| BrokerError::invalid_params(error.to_string())) +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct DeriveParams { + invocation: InvocationSpec, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct EnterParams { + worker_id: String, + invocation: InvocationSpec, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct ObserveParams { + worker_id: String, + source: String, + resource: ResourceSpec, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct CallParams { + worker_id: String, + operation: String, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct PublishParams { + worker_id: String, + content_sha256: String, + destination: DestinationSpec, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct WorkerParams { + worker_id: String, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct InvocationSpec { + realm_url: String, + channel_id: Uuid, + conversation_kind: ConversationKindSpec, + epoch: String, + #[serde(default)] + members: Vec, + executing_agent: String, + requesters: Vec, + system_principal: Option, + owner: Option, + #[serde(default)] + bot_capabilities: Vec, + #[serde(default)] + conversation_capabilities: Vec, +} + +impl InvocationSpec { + fn resolve(self) -> Result { + if self.realm_url.trim().is_empty() { + return Err(BrokerError::invalid_params("realm_url must not be empty")); + } + if self.epoch.trim().is_empty() { + return Err(BrokerError::invalid_params("epoch must not be empty")); + } + let policy = CapabilityPolicy::new( + CapabilitySet::from_names(self.bot_capabilities), + CapabilitySet::from_names(self.conversation_capabilities), + ); + derive_execution_domain( + DomainFacts { + realm: RealmId::from_relay_url(&self.realm_url), + channel_id: self.channel_id, + kind: self.conversation_kind.into(), + epoch: MembershipEpoch::new(self.epoch), + members: resolve_principals(self.members)?, + executing_agent: parse_principal(&self.executing_agent)?, + requesters: resolve_principals(self.requesters)?, + system_principal: self + .system_principal + .as_deref() + .map(parse_principal) + .transpose()?, + owner: self.owner.as_deref().map(parse_principal).transpose()?, + }, + &policy, + ) + .map_err(|error| BrokerError::invalid_params(error.to_string())) + } +} + +#[derive(Clone, Copy, Deserialize)] +#[serde(rename_all = "snake_case")] +enum ConversationKindSpec { + Public, + Restricted, + DirectMessage, +} + +impl From for ConversationKind { + fn from(value: ConversationKindSpec) -> Self { + match value { + ConversationKindSpec::Public => Self::Public, + ConversationKindSpec::Restricted => Self::Restricted, + ConversationKindSpec::DirectMessage => Self::DirectMessage, + } + } +} + +#[derive(Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +enum ResourceSpec { + Domain, + TrustedConfiguration, + Public, + Conversation { + channel_id: Uuid, + readers: Vec, + }, + OwnerPrivate { + owner: String, + }, +} + +impl ResourceSpec { + fn resolve(self, domain: &ExecutionDomain) -> Result { + let realm = domain.audience().realm().clone(); + match self { + Self::Domain => Ok(ResourceLabel::domain(domain)), + Self::TrustedConfiguration => Ok(ResourceLabel::trusted_configuration(realm)), + Self::Public => Ok(ResourceLabel::realm_public(realm)), + Self::Conversation { + channel_id, + readers, + } => ResourceLabel::conversation(realm, channel_id, resolve_principals(readers)?) + .map_err(|error| BrokerError::invalid_params(error.to_string())), + Self::OwnerPrivate { owner } => Ok(ResourceLabel::owner_private( + realm, + parse_principal(&owner)?, + )), + } + } +} + +#[derive(Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +enum DestinationSpec { + Current, + Public, + Conversation { + channel_id: Uuid, + readers: Vec, + }, + OwnerPrivate { + owner: String, + }, +} + +impl DestinationSpec { + fn resolve( + self, + domain: &ExecutionDomain, + ) -> Result<(ConfidentialityLabel, DomainContext), BrokerError> { + let realm = domain.audience().realm().clone(); + match self { + Self::Current => Ok((domain.audience().clone(), domain.context().clone())), + Self::Public => Ok(( + ConfidentialityLabel::public(realm.clone()), + DomainContext::RealmPublic(realm), + )), + Self::Conversation { + channel_id, + readers, + } => Ok(( + ConfidentialityLabel::restricted(realm.clone(), resolve_principals(readers)?) + .map_err(|error| BrokerError::invalid_params(error.to_string()))?, + DomainContext::Conversation { realm, channel_id }, + )), + Self::OwnerPrivate { owner } => { + let owner = parse_principal(&owner)?; + Ok(( + ConfidentialityLabel::restricted( + realm.clone(), + BTreeSet::from([owner.clone()]), + ) + .map_err(|error| BrokerError::invalid_params(error.to_string()))?, + DomainContext::OwnerPrivate { realm, owner }, + )) + } + } + } +} + +fn resolve_principals(values: Vec) -> Result, BrokerError> { + values + .into_iter() + .map(|value| parse_principal(&value)) + .collect() +} + +fn parse_principal(value: &str) -> Result { + Principal::from_hex(value).map_err(|error| BrokerError::invalid_params(error.to_string())) +} + +fn parse_digest(value: &str) -> Result<[u8; 32], BrokerError> { + let bytes = hex::decode(value) + .map_err(|error| BrokerError::invalid_params(format!("invalid content_sha256: {error}")))?; + bytes + .try_into() + .map_err(|_| BrokerError::invalid_params("content_sha256 must contain 32 bytes")) +} + +fn validate_worker_id(value: &str) -> Result<(), BrokerError> { + let length = value.len(); + if length == 0 || length > 256 { + return Err(BrokerError::invalid_params( + "worker_id must contain between 1 and 256 bytes", + )); + } + Ok(()) +} + +fn active_worker<'a>( + workers: &'a HashMap, + worker_id: &str, +) -> Result<&'a WorkerRecord, BrokerError> { + validate_worker_id(worker_id)?; + workers + .get(worker_id) + .ok_or_else(|| BrokerError::state("worker has not entered an execution domain")) +} + +fn active_worker_mut<'a>( + workers: &'a mut HashMap, + worker_id: &str, +) -> Result<&'a mut WorkerRecord, BrokerError> { + validate_worker_id(worker_id)?; + workers + .get_mut(worker_id) + .ok_or_else(|| BrokerError::state("worker has not entered an execution domain")) +} + +fn domain_description(domain: &ExecutionDomain) -> Value { + json!({ + "domain_id": domain.id(), + "domain_fingerprint": domain.key().fingerprint(), + "context": domain.context().kind(), + "compartment": domain.compartment_profile().as_str(), + "audience": if domain.audience().is_public() { "public" } else { "restricted" }, + "reader_count": domain.audience().reader_count(), + "epoch_fingerprint": domain.epoch_fingerprint(), + }) +} + +fn decision_value( + mode: EnforcementMode, + rule: &'static str, + decision: &RuleDecision, + worker_id: Option<&str>, + domain_id: Option<&str>, + details: Value, +) -> Value { + tracing::info!( + target: "buzz_ifc_broker::decision", + rule, + decision = decision.result(), + reason = decision.reason(), + enforced = mode.enforced(), + worker_id, + domain_id, + "IFC broker security decision" + ); + json!({ + "decision": decision.result(), + "allowed": decision.allowed(), + "reason": decision.reason(), + "enforced": mode.enforced(), + "worker_id": worker_id, + "domain_id": domain_id, + "details": details, + }) +} + +fn poisoned_decision(mode: EnforcementMode, worker_id: &str, record: &WorkerRecord) -> Value { + let reason = "worker crossed an execution-domain boundary and must be retired"; + let domain_id = record.domain.as_ref().map(ExecutionDomain::id); + tracing::info!( + target: "buzz_ifc_broker::decision", + rule = "reuse", + decision = "deny", + reason, + enforced = mode.enforced(), + worker_id, + domain_id, + "IFC broker security decision" + ); + json!({ + "decision": "deny", + "allowed": false, + "reason": reason, + "enforced": mode.enforced(), + "worker_id": worker_id, + "domain_id": domain_id, + "details": { "replace_worker": true }, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + const ALICE: &str = "0101010101010101010101010101010101010101010101010101010101010101"; + const BOB: &str = "0202020202020202020202020202020202020202020202020202020202020202"; + const AGENT: &str = "79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798"; + + fn request(id: u64, method: &str, params: Value) -> String { + json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params }).to_string() + } + + fn restricted_invocation(epoch: &str) -> Value { + json!({ + "realm_url": "wss://buzz.example", + "channel_id": "00000000-0000-0000-0000-000000000001", + "conversation_kind": "restricted", + "epoch": epoch, + "members": [AGENT, ALICE, BOB], + "executing_agent": AGENT, + "requesters": [ALICE], + "owner": ALICE, + "bot_capabilities": ["buzz.read.current", "email.read"], + "conversation_capabilities": ["buzz.read.current"] + }) + } + + fn response(line: &str) -> Value { + serde_json::from_str(line).expect("broker response") + } + + #[test] + fn cross_domain_reuse_requires_worker_replacement() { + let mut broker = Broker::new(EnforcementMode::Enforce); + let first = response(&broker.handle_line(&request( + 1, + "worker/enter", + json!({ "worker_id": "worker-1", "invocation": restricted_invocation("v1") }), + ))); + assert_eq!(first["result"]["decision"], "allow"); + assert_eq!(first["result"]["details"]["compartment"], "domain_confined"); + + let second = response(&broker.handle_line(&request( + 2, + "worker/enter", + json!({ "worker_id": "worker-1", "invocation": restricted_invocation("v2") }), + ))); + assert_eq!(second["result"]["decision"], "deny"); + assert_eq!(second["result"]["details"]["replace_worker"], true); + + let call = response(&broker.handle_line(&request( + 3, + "worker/call", + json!({ "worker_id": "worker-1", "operation": "buzz.read.current" }), + ))); + assert_eq!(call["result"]["decision"], "deny"); + } + + #[test] + fn owner_private_read_is_denied_in_a_conversation_domain() { + let mut broker = Broker::new(EnforcementMode::Enforce); + let _ = broker.handle_line(&request( + 1, + "worker/enter", + json!({ "worker_id": "worker-1", "invocation": restricted_invocation("v1") }), + )); + let read = response(&broker.handle_line(&request( + 2, + "worker/observe", + json!({ + "worker_id": "worker-1", + "source": "email", + "resource": { "kind": "owner_private", "owner": ALICE } + }), + ))); + assert_eq!(read["result"]["decision"], "deny"); + assert_eq!(read["result"]["enforced"], true); + } + + #[test] + fn conversation_worker_is_bound_to_shared_tools_and_its_source_context() { + let mut broker = Broker::new(EnforcementMode::Enforce); + let _ = broker.handle_line(&request( + 1, + "worker/enter", + json!({ "worker_id": "worker-1", "invocation": restricted_invocation("v1") }), + )); + + let call = response(&broker.handle_line(&request( + 2, + "worker/call", + json!({ "worker_id": "worker-1", "operation": "email.read" }), + ))); + assert_eq!(call["result"]["decision"], "deny"); + + let observe = response(&broker.handle_line(&request( + 3, + "worker/observe", + json!({ + "worker_id": "worker-1", + "source": "conversation_history", + "resource": { "kind": "domain" } + }), + ))); + assert_eq!(observe["result"]["decision"], "allow"); + + let current = response(&broker.handle_line(&request( + 4, + "worker/publish", + json!({ + "worker_id": "worker-1", + "content_sha256": "00".repeat(32), + "destination": { "kind": "current" } + }), + ))); + assert_eq!(current["result"]["decision"], "allow"); + + let other = response(&broker.handle_line(&request( + 5, + "worker/publish", + json!({ + "worker_id": "worker-1", + "content_sha256": "00".repeat(32), + "destination": { + "kind": "conversation", + "channel_id": "00000000-0000-0000-0000-000000000002", + "readers": [ALICE, BOB] + } + }), + ))); + assert_eq!(other["result"]["decision"], "deny"); + } + + #[test] + fn malformed_verified_fact_fails_derivation_without_panicking() { + let mut broker = Broker::new(EnforcementMode::Audit); + let invalid = response(&broker.handle_line(&request( + 1, + "domain/derive", + json!({ + "invocation": { + "realm_url": "wss://buzz.example", + "channel_id": "00000000-0000-0000-0000-000000000001", + "conversation_kind": "direct_message", + "epoch": "v1", + "members": [AGENT, ALICE], + "executing_agent": AGENT, + "requesters": [ALICE], + "owner": "not-a-key" + } + }), + ))); + assert_eq!(invalid["error"]["code"], -32602); + } + + #[test] + fn public_domain_selects_the_shared_public_compartment() { + let mut broker = Broker::new(EnforcementMode::Audit); + let derived = response(&broker.handle_line(&request( + 1, + "domain/derive", + json!({ + "invocation": { + "realm_url": "wss://buzz.example", + "channel_id": "00000000-0000-0000-0000-000000000001", + "conversation_kind": "public", + "epoch": "community:v1", + "executing_agent": AGENT, + "requesters": [ALICE], + "owner": ALICE + } + }), + ))); + + assert_eq!(derived["result"]["compartment"], "shared_public"); + } +} diff --git a/crates/buzz-ifc-broker/src/main.rs b/crates/buzz-ifc-broker/src/main.rs new file mode 100644 index 0000000000..2841256acd --- /dev/null +++ b/crates/buzz-ifc-broker/src/main.rs @@ -0,0 +1,69 @@ +use std::io; + +use anyhow::Context; +use buzz_ifc_broker::{Broker, EnforcementMode}; +use clap::{Parser, ValueEnum}; +use futures_util::StreamExt; +use tokio::io::{AsyncWriteExt, BufWriter}; +use tokio_util::codec::{FramedRead, LinesCodec}; + +const MAX_REQUEST_BYTES: usize = 1024 * 1024; + +#[derive(Parser)] +#[command(name = "buzz-ifc-broker", version, about)] +struct Args { + /// Whether decisions are observational or must be enforced by the adapter. + #[arg(long, env = "BUZZ_IFC_BROKER_MODE", default_value = "audit")] + mode: Mode, +} + +#[derive(Clone, Copy, ValueEnum)] +enum Mode { + Audit, + Enforce, +} + +impl From for EnforcementMode { + fn from(value: Mode) -> Self { + match value { + Mode::Audit => Self::Audit, + Mode::Enforce => Self::Enforce, + } + } +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let args = Args::parse(); + let _ = tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .with_writer(io::stderr) + .try_init(); + + let mode = EnforcementMode::from(args.mode); + tracing::info!(%mode, "Buzz IFC broker started"); + let mut broker = Broker::new(mode); + let stdin = tokio::io::stdin(); + let mut frames = FramedRead::new(stdin, LinesCodec::new_with_max_length(MAX_REQUEST_BYTES)); + let mut stdout = BufWriter::new(tokio::io::stdout()); + + while let Some(frame) = frames.next().await { + let response = match frame { + Ok(line) => broker.handle_line(&line), + Err(error) => { + tracing::warn!(%error, "Buzz IFC broker rejected an input frame"); + Broker::frame_error("request frame exceeds the broker limit") + } + }; + stdout + .write_all(response.as_bytes()) + .await + .context("write broker response")?; + stdout.write_all(b"\n").await.context("write line ending")?; + stdout.flush().await.context("flush broker response")?; + } + Ok(()) +} diff --git a/crates/buzz-ifc-broker/tests/conformance.rs b/crates/buzz-ifc-broker/tests/conformance.rs new file mode 100644 index 0000000000..92ef2e3f14 --- /dev/null +++ b/crates/buzz-ifc-broker/tests/conformance.rs @@ -0,0 +1,22 @@ +use buzz_ifc_broker::{Broker, EnforcementMode, PROTOCOL_VERSION}; +use serde::Deserialize; +use serde_json::Value; + +#[derive(Deserialize)] +struct Fixture { + protocol_version: String, + request: Value, + expected_result: Value, +} + +#[test] +fn domain_golden_fixture_matches_the_wire_contract() { + let fixture: Fixture = serde_json::from_str(include_str!("fixtures/domain-golden.json")) + .expect("valid conformance fixture"); + assert_eq!(fixture.protocol_version, PROTOCOL_VERSION); + + let mut broker = Broker::new(EnforcementMode::Enforce); + let response: Value = serde_json::from_str(&broker.handle_line(&fixture.request.to_string())) + .expect("valid broker response"); + assert_eq!(response["result"], fixture.expected_result); +} diff --git a/crates/buzz-ifc-broker/tests/fixtures/domain-golden.json b/crates/buzz-ifc-broker/tests/fixtures/domain-golden.json new file mode 100644 index 0000000000..c71b96f4ac --- /dev/null +++ b/crates/buzz-ifc-broker/tests/fixtures/domain-golden.json @@ -0,0 +1,37 @@ +{ + "protocol_version": "buzz-ifc-broker/1", + "request": { + "jsonrpc": "2.0", + "id": 1, + "method": "domain/derive", + "params": { + "invocation": { + "realm_url": "wss://buzz.example", + "channel_id": "00000000-0000-0000-0000-000000000001", + "conversation_kind": "restricted", + "epoch": "membership:event-1", + "members": [ + "79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + "0101010101010101010101010101010101010101010101010101010101010101", + "0202020202020202020202020202020202020202020202020202020202020202" + ], + "executing_agent": "79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + "requesters": [ + "0101010101010101010101010101010101010101010101010101010101010101" + ], + "owner": "0101010101010101010101010101010101010101010101010101010101010101", + "bot_capabilities": ["buzz.read.current"], + "conversation_capabilities": ["buzz.read.current"] + } + } + }, + "expected_result": { + "audience": "restricted", + "compartment": "domain_confined", + "context": "conversation", + "domain_fingerprint": "77ed1593198b", + "domain_id": "fbad5ec7aed0f274f9c517fed619cb41262c2295fb62451477a3df7caf1f3ef1", + "epoch_fingerprint": "434d4af44419", + "reader_count": 2 + } +} diff --git a/crates/buzz-ifc-broker/tests/stdio.rs b/crates/buzz-ifc-broker/tests/stdio.rs new file mode 100644 index 0000000000..cdb228acb8 --- /dev/null +++ b/crates/buzz-ifc-broker/tests/stdio.rs @@ -0,0 +1,35 @@ +use std::io::Write; +use std::process::{Command, Stdio}; + +use serde_json::{json, Value}; + +#[test] +fn stdio_keeps_protocol_output_to_one_json_line() { + let mut child = Command::new(env!("CARGO_BIN_EXE_buzz-ifc-broker")) + .args(["--mode", "enforce"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn broker"); + let request = json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "broker/info", + "params": {} + }); + child + .stdin + .take() + .expect("broker stdin") + .write_all(format!("{request}\n").as_bytes()) + .expect("write request"); + + let output = child.wait_with_output().expect("broker output"); + assert!(output.status.success()); + let stdout = String::from_utf8(output.stdout).expect("UTF-8 stdout"); + assert_eq!(stdout.lines().count(), 1); + let response: Value = serde_json::from_str(stdout.trim()).expect("JSON-RPC response"); + assert_eq!(response["result"]["mode"], "enforce"); + assert_eq!(response["result"]["worker_count"], 0); +}