Add broker-facing IFC sessions

Signed-off-by: Jordan Mecom <jm@squareup.com>

crates/buzz-ifc/Cargo.toml

@@ -5,7 +5,7 @@ edition.workspace = true
 rust-version.workspace = true
 license.workspace = true
 repository.workspace = true
-description = "Buzz execution-domain derivation for information-flow control"
+description = "Buzz execution domains and broker-side information-flow checks"
 
 [dependencies]
 buzz-core = { workspace = true }

crates/buzz-ifc/src/domain.rs

@@ -45,6 +45,26 @@ impl DomainContext {
         }
     }
 
+    /// Whether this retained-state context may admit a resource from `source`.
+    ///
+    /// Public community data may enter any context in that community. A
+    /// conversation admits only its own retained data. Owner-private work may
+    /// also narrow conversation data to the owner when the separate audience
+    /// check permits that flow; it never admits another owner's private state.
+    pub(crate) fn permits(&self, source: &Self) -> bool {
+        if self.community() != source.community() {
+            return false;
+        }
+
+        match source {
+            Self::CommunityPublic(_) => true,
+            Self::Conversation { .. } => {
+                self == source || matches!(self, Self::OwnerPrivate { .. })
+            }
+            Self::OwnerPrivate { .. } => self == source,
+        }
+    }
+
     pub(crate) fn stable_hash(&self, hasher: &mut Sha256) {
         match self {
             Self::CommunityPublic(community) => {
@@ -72,7 +92,8 @@ impl DomainContext {
 ///
 /// The broker configures this; the agent does not get to classify its own calls.
 /// For a publication, permission to call the operation is not enough: the broker
-/// must also check whether the information may flow to the destination's readers.
+/// must use [`crate::IfcSession::publish`] to check whether the information may
+/// flow to the destination's readers.
 #[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
 #[serde(rename_all = "snake_case")]
 pub enum OperationEffect {
@@ -153,6 +174,10 @@ impl CapabilitySet {
         Self(effective)
     }
 
+    pub(crate) fn effect(&self, operation: &str) -> Option<OperationEffect> {
+        self.0.get(operation).copied()
+    }
+
     fn stable_hash(&self, hasher: &mut Sha256) {
         hash_field(hasher, &(self.0.len() as u64).to_be_bytes());
         for (name, effect) in &self.0 {

crates/buzz-ifc/src/lib.rs

@@ -1,4 +1,4 @@
-//! Derives an agent's execution domain from Buzz membership and capability policy.
+//! Derives Buzz execution domains and checks broker reads, calls, and publications.
 //!
 //! An [`ExecutionDomain`] records which agent is running, who may receive its
 //! output, which conversations may share its saved state, and which operations
@@ -6,18 +6,28 @@
 //! and membership version, so a change produces a different key.
 //!
 //! The broker must verify events and membership before supplying [`DomainFacts`].
-//! It must also use the resulting key when selecting a session. This crate
-//! computes the domain; it does not verify signatures, manage sessions, or
-//! enforce tool calls.
+//! It uses the resulting key to select an [`IfcSession`], then checks reads and
+//! calls before executing them. Publications must pass [`IfcSession::publish`]
+//! before reaching a sink. This crate does not verify signatures, load saved
+//! sessions, or execute operations itself.
+
+#![forbid(unsafe_code)]
 
 mod domain;
 mod label;
+mod session;
 
 pub use domain::{
     derive_execution_domain, CapabilityPolicy, CapabilitySet, ConversationKind, DerivationError,
     DomainFacts, DomainKey, ExecutionDomain, MembershipEpoch, OperationEffect,
 };
 pub use label::{CommunityId, ConfidentialityLabel, LabelError, Principal, PrincipalError};
+pub use session::{
+    AuthorizedPublication, IfcError, IfcSession, PublicationRequest, PublicationTarget,
+    ResourceLabel,
+};
 
+#[cfg(test)]
+mod session_tests;
 #[cfg(test)]
 mod tests;

crates/buzz-ifc/src/session.rs

@@ -0,0 +1,254 @@
+use ifc_core::{EgressError, FlowState};
+
+use crate::domain::{DomainContext, DomainKey, ExecutionDomain, MembershipEpoch, OperationEffect};
+use crate::label::{CommunityId, ConfidentialityLabel, Principal};
+
+/// The security metadata the broker checks before exposing a resource to an
+/// agent session.
+///
+/// Constructing this from the domain where the resource originated keeps its
+/// audience, retained-state context, and membership epoch together. Public
+/// community data does not carry an epoch because any context in that
+/// community may read it. Restricted data retains the epoch under which its
+/// source domain was authorized.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ResourceLabel {
+    audience: ConfidentialityLabel,
+    context: DomainContext,
+    epoch: Option<MembershipEpoch>,
+}
+
+impl ResourceLabel {
+    /// Label a resource with the security metadata of its source domain.
+    pub fn from_domain(domain: &ExecutionDomain) -> Self {
+        let epoch = (!matches!(domain.context, DomainContext::CommunityPublic(_)))
+            .then(|| domain.epoch.clone());
+        Self {
+            audience: domain.audience.clone(),
+            context: domain.context.clone(),
+            epoch,
+        }
+    }
+}
+
+/// The audience receiving a publication.
+///
+/// This type describes only the IFC destination. The trusted broker remains
+/// responsible for ordinary product policy, including resolving and checking
+/// the channel, message, URL, or other concrete destination named by the
+/// operation payload.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct PublicationTarget {
+    audience: ConfidentialityLabel,
+}
+
+impl PublicationTarget {
+    /// Use an execution domain's authorized audience as a publication target.
+    pub fn from_domain(domain: &ExecutionDomain) -> Self {
+        Self {
+            audience: domain.audience.clone(),
+        }
+    }
+}
+
+/// A broker request to execute one publication operation with an exact
+/// payload.
+pub struct PublicationRequest<T> {
+    operation: String,
+    target: PublicationTarget,
+    payload: T,
+}
+
+impl<T> PublicationRequest<T> {
+    /// Construct a request after the broker has resolved its IFC target.
+    pub fn new(operation: impl Into<String>, target: PublicationTarget, payload: T) -> Self {
+        Self {
+            operation: operation.into(),
+            target,
+            payload,
+        }
+    }
+}
+
+/// Proof that an exact publication request passed the session's capability
+/// and information-flow checks.
+///
+/// A broker sink should accept this type rather than [`PublicationRequest`].
+/// Its fields are private and it is not cloneable, so only
+/// [`IfcSession::publish`] can create the value that authorizes a sink call.
+///
+/// ```compile_fail
+/// let forged = buzz_ifc::AuthorizedPublication {
+///     operation: "buzz.reply".to_owned(),
+///     target: todo!(),
+///     payload: "unchecked",
+/// };
+/// ```
+#[must_use = "the authorization must be consumed by the publication sink"]
+pub struct AuthorizedPublication<T> {
+    operation: String,
+    target: PublicationTarget,
+    payload: T,
+}
+
+impl<T> AuthorizedPublication<T> {
+    /// Return the checked operation name.
+    pub fn operation(&self) -> &str {
+        &self.operation
+    }
+
+    /// Return the exact payload covered by the information-flow decision.
+    pub fn payload(&self) -> &T {
+        &self.payload
+    }
+
+    /// Consume the authorization and return the checked sink inputs.
+    pub fn into_parts(self) -> (String, PublicationTarget, T) {
+        (self.operation, self.target, self.payload)
+    }
+}
+
+/// Information-flow state for one retained agent execution domain.
+///
+/// The broker calls [`read`](Self::read) before delivering data to the agent,
+/// [`call`](Self::call) before executing operations that cannot publish, and
+/// [`publish`](Self::publish) before handing an exact outbound payload to a
+/// sink. These are the ordinary-flow checks described in Appendices F and G of
+/// the [design paper](../../../docs/practical-information-flow-for-buzz-agents.md#appendix-f-reference-monitor-pseudocode).
+///
+/// ```
+/// # use buzz_ifc::{AuthorizedPublication, ExecutionDomain, IfcError, IfcSession,
+/// #     PublicationRequest, PublicationTarget, ResourceLabel};
+/// # fn broker_sink(_: AuthorizedPublication<Vec<u8>>) {}
+/// # fn run_turn(
+/// #     domain: ExecutionDomain,
+/// #     resource: &ResourceLabel,
+/// #     target: PublicationTarget,
+/// # ) -> Result<(), IfcError> {
+/// let mut session = IfcSession::enter(domain);
+/// session.read(resource)?;
+/// session.call("buzz.read.current")?;
+///
+/// let authorization = session.publish(PublicationRequest::new(
+///     "buzz.reply",
+///     target,
+///     b"hello".to_vec(),
+/// ))?;
+/// broker_sink(authorization);
+/// # Ok(())
+/// # }
+/// ```
+#[derive(Debug)]
+pub struct IfcSession {
+    domain: ExecutionDomain,
+    flow: FlowState<CommunityId, Principal>,
+}
+
+impl IfcSession {
+    /// Enter an execution domain selected by the trusted broker.
+    ///
+    /// The domain's audience is observed immediately because retained agent
+    /// state and broker-provided instructions may already influence the next
+    /// output before the first explicit resource read.
+    pub fn enter(domain: ExecutionDomain) -> Self {
+        let mut flow = FlowState::default();
+        flow.observe(&domain.audience);
+        Self { domain, flow }
+    }
+
+    /// Return the key the broker uses to route later turns to this session.
+    pub fn domain_key(&self) -> DomainKey {
+        self.domain.key()
+    }
+
+    /// Check and record a labeled resource before exposing it to the agent.
+    ///
+    /// A failed read does not taint the session because the broker must not
+    /// deliver the rejected resource. Successful reads are recorded
+    /// monotonically; the domain audience already conservatively bounds output
+    /// in this coarse-grained Buzz model.
+    pub fn read(&mut self, resource: &ResourceLabel) -> Result<(), IfcError> {
+        if !resource.audience.can_flow_to(&self.domain.audience) {
+            return Err(IfcError::ReadAudienceDenied);
+        }
+        if !self.domain.context.permits(&resource.context) {
+            return Err(IfcError::ReadContextDenied);
+        }
+        if resource.context == self.domain.context
+            && resource
+                .epoch
+                .as_ref()
+                .is_some_and(|epoch| epoch != &self.domain.epoch)
+        {
+            return Err(IfcError::StaleResourceEpoch);
+        }
+
+        self.flow.observe(&resource.audience);
+        Ok(())
+    }
+
+    /// Permanently record that unlabeled input reached the agent.
+    ///
+    /// Ordinary publication remains blocked for the rest of the session. A
+    /// future declassification API may authorize a specific exceptional flow,
+    /// but this small wrapper deliberately provides no bypass.
+    pub fn mark_unknown_input(&mut self) {
+        self.flow.mark_unknown();
+    }
+
+    /// Authorize an admitted operation that cannot publish information.
+    pub fn call(&self, operation: &str) -> Result<(), IfcError> {
+        match self.domain.capabilities.effect(operation) {
+            Some(OperationEffect::NonEgressing) => Ok(()),
+            Some(OperationEffect::Publication) => Err(IfcError::PublicationRequiresPublish),
+            None => Err(IfcError::CapabilityDenied),
+        }
+    }
+
+    /// Authorize an exact outbound payload for a checked broker sink.
+    pub fn publish<T>(
+        &self,
+        request: PublicationRequest<T>,
+    ) -> Result<AuthorizedPublication<T>, IfcError> {
+        match self.domain.capabilities.effect(&request.operation) {
+            Some(OperationEffect::Publication) => {}
+            Some(OperationEffect::NonEgressing) => {
+                return Err(IfcError::NonEgressingRequiresCall);
+            }
+            None => return Err(IfcError::CapabilityDenied),
+        }
+
+        self.flow.check_egress(&request.target.audience)?;
+        Ok(AuthorizedPublication {
+            operation: request.operation,
+            target: request.target,
+            payload: request.payload,
+        })
+    }
+}
+
+/// Why an IFC session refused a broker action.
+#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)]
+pub enum IfcError {
+    /// The resource could be read by principals outside the session audience.
+    #[error("resource audience is not safe for this execution domain")]
+    ReadAudienceDenied,
+    /// The resource belongs to retained state this domain may not reuse.
+    #[error("resource belongs to a different retained-state context")]
+    ReadContextDenied,
+    /// Restricted state was created under a different membership epoch.
+    #[error("resource membership epoch does not match the execution domain")]
+    StaleResourceEpoch,
+    /// The execution domain does not admit the requested operation.
+    #[error("operation is not admitted by this execution domain")]
+    CapabilityDenied,
+    /// An egressing operation was presented to the unchecked call path.
+    #[error("publication operation must use IfcSession::publish")]
+    PublicationRequiresPublish,
+    /// A non-egressing operation was presented to the publication path.
+    #[error("non-egressing operation must use IfcSession::call")]
+    NonEgressingRequiresCall,
+    /// Accumulated information cannot flow to the requested audience.
+    #[error("information-flow check failed: {0}")]
+    InformationFlow(#[from] EgressError),
+}

crates/buzz-ifc/src/session_tests.rs

@@ -0,0 +1,360 @@
+use std::collections::{BTreeSet, HashMap};
+
+use nostr::Keys;
+use uuid::Uuid;
+
+use super::*;
+
+const READ: &str = "buzz.read.current";
+const REPLY: &str = "buzz.reply";
+
+fn principal(value: u8) -> Principal {
+    let keys = Keys::parse(&format!("{value:064x}")).expect("test secret key");
+    Principal::from_public_key(&keys.public_key()).expect("valid test principal")
+}
+
+fn community(value: u128) -> CommunityId {
+    CommunityId::from_uuid(Uuid::from_u128(value))
+}
+
+fn policy() -> CapabilityPolicy {
+    let operations = || {
+        CapabilitySet::from_operations([
+            (READ, OperationEffect::NonEgressing),
+            (REPLY, OperationEffect::Publication),
+        ])
+    };
+    CapabilityPolicy::new(operations(), operations())
+}
+
+fn public_domain(community: CommunityId, channel: u128, epoch: &str) -> ExecutionDomain {
+    derive_execution_domain(
+        DomainFacts {
+            community,
+            channel_id: Uuid::from_u128(channel),
+            kind: ConversationKind::Public,
+            epoch: MembershipEpoch::new(epoch),
+            members: BTreeSet::new(),
+            executing_agent: principal(9),
+            requesters: BTreeSet::from([principal(1)]),
+            system_principal: None,
+            owner: Some(principal(1)),
+        },
+        &policy(),
+    )
+    .expect("valid public domain")
+}
+
+fn restricted_domain(
+    community: CommunityId,
+    channel: u128,
+    epoch: &str,
+    readers: &[u8],
+) -> ExecutionDomain {
+    let agent = principal(9);
+    let members = readers
+        .iter()
+        .copied()
+        .map(principal)
+        .chain([agent])
+        .collect();
+    derive_execution_domain(
+        DomainFacts {
+            community,
+            channel_id: Uuid::from_u128(channel),
+            kind: ConversationKind::Restricted,
+            epoch: MembershipEpoch::new(epoch),
+            members,
+            executing_agent: agent,
+            requesters: BTreeSet::from([principal(readers[0])]),
+            system_principal: None,
+            owner: Some(principal(1)),
+        },
+        &policy(),
+    )
+    .expect("valid restricted domain")
+}
+
+fn owner_private_domain(community: CommunityId, channel: u128, epoch: &str) -> ExecutionDomain {
+    let owner = principal(1);
+    let agent = principal(9);
+    derive_execution_domain(
+        DomainFacts {
+            community,
+            channel_id: Uuid::from_u128(channel),
+            kind: ConversationKind::DirectMessage,
+            epoch: MembershipEpoch::new(epoch),
+            members: BTreeSet::from([agent, owner]),
+            executing_agent: agent,
+            requesters: BTreeSet::from([owner]),
+            system_principal: None,
+            owner: Some(owner),
+        },
+        &policy(),
+    )
+    .expect("valid owner-private domain")
+}
+
+fn deliver_to_agent(
+    session: &mut IfcSession,
+    label: &ResourceLabel,
+    value: &str,
+    inbox: &mut Vec<String>,
+) -> Result<(), IfcError> {
+    session.read(label)?;
+    inbox.push(value.to_owned());
+    Ok(())
+}
+
+fn execute_publication(
+    authorization: AuthorizedPublication<String>,
+    sink_log: &mut Vec<(String, String)>,
+) {
+    let (operation, _checked_target, payload) = authorization.into_parts();
+    sink_log.push((operation, payload));
+}
+
+/// This is the intended broker-facing happy path. The broker checks a resource
+/// before delivery and gives its sink only the unforgeable result returned by
+/// `publish`, rather than executing an unchecked `PublicationRequest`.
+#[test]
+fn broker_turn_uses_one_small_checked_surface() {
+    let domain = restricted_domain(community(1), 10, "membership:v1", &[1, 2]);
+    let resource = ResourceLabel::from_domain(&domain);
+    let target = PublicationTarget::from_domain(&domain);
+    let mut session = IfcSession::enter(domain);
+    let mut inbox = Vec::new();
+    let mut sink_log = Vec::new();
+
+    deliver_to_agent(&mut session, &resource, "question", &mut inbox)
+        .expect("broker may deliver the current conversation");
+    session
+        .call(READ)
+        .expect("read operation cannot publish information");
+    let authorization = session
+        .publish(PublicationRequest::new(REPLY, target, "answer".to_owned()))
+        .expect("reply may flow to the current audience");
+    execute_publication(authorization, &mut sink_log);
+
+    assert_eq!(inbox, ["question"]);
+    assert_eq!(sink_log, [(REPLY.to_owned(), "answer".to_owned())]);
+}
+
+/// A rejected read must never reach the agent and must not taint the session.
+/// This binds the test to the broker seam: `read` runs before the value is
+/// appended to the simulated agent inbox.
+#[test]
+fn broker_does_not_deliver_a_resource_with_a_narrower_audience() {
+    let group = restricted_domain(community(1), 10, "membership:v1", &[1, 2]);
+    let alice_only = restricted_domain(community(1), 20, "membership:v1", &[1]);
+    let resource = ResourceLabel::from_domain(&alice_only);
+    let target = PublicationTarget::from_domain(&group);
+    let mut session = IfcSession::enter(group);
+    let mut inbox = Vec::new();
+
+    assert_eq!(
+        deliver_to_agent(&mut session, &resource, "alice secret", &mut inbox),
+        Err(IfcError::ReadAudienceDenied)
+    );
+    assert!(inbox.is_empty());
+    assert!(session
+        .publish(PublicationRequest::new(REPLY, target, "safe"))
+        .is_ok());
+}
+
+/// Publication operations must not enter through `call`, which performs no
+/// destination-flow check. Misclassifying this path would bypass IFC entirely.
+#[test]
+fn egressing_operation_cannot_use_the_call_path() {
+    let session = IfcSession::enter(public_domain(community(1), 10, "community:v1"));
+
+    assert_eq!(
+        session.call(REPLY),
+        Err(IfcError::PublicationRequiresPublish)
+    );
+}
+
+/// The inverse mismatch is also rejected so every operation has one obvious
+/// broker API and policy cannot silently change how a call is executed.
+#[test]
+fn non_egressing_operation_cannot_use_the_publish_path() {
+    let domain = public_domain(community(1), 10, "community:v1");
+    let target = PublicationTarget::from_domain(&domain);
+    let session = IfcSession::enter(domain);
+
+    assert!(matches!(
+        session.publish(PublicationRequest::new(READ, target, "payload")),
+        Err(IfcError::NonEgressingRequiresCall)
+    ));
+}
+
+/// Capability admission fails closed on both paths. An operation name supplied
+/// by an agent cannot become authority merely because the broker recognizes
+/// how to execute it.
+#[test]
+fn operation_absent_from_the_domain_is_denied() {
+    let domain = public_domain(community(1), 10, "community:v1");
+    let target = PublicationTarget::from_domain(&domain);
+    let session = IfcSession::enter(domain);
+
+    assert_eq!(session.call("email.send"), Err(IfcError::CapabilityDenied));
+    assert!(matches!(
+        session.publish(PublicationRequest::new("email.send", target, "payload")),
+        Err(IfcError::CapabilityDenied)
+    ));
+}
+
+/// Accumulated private state must not be widened to a public audience. This is
+/// the central no-write-down confidentiality invariant at the checked sink.
+#[test]
+fn private_session_cannot_publish_to_a_public_audience() {
+    let private = restricted_domain(community(1), 10, "membership:v1", &[1, 2]);
+    let public = public_domain(community(1), 20, "community:v1");
+    let target = PublicationTarget::from_domain(&public);
+    let session = IfcSession::enter(private);
+
+    assert_eq!(
+        session
+            .publish(PublicationRequest::new(REPLY, target, "secret"))
+            .err(),
+        Some(IfcError::InformationFlow(
+            ifc_core::EgressError::DestinationWidensReaders
+        ))
+    );
+}
+
+/// Public information may be sent to fewer readers. Reversing this ordering is
+/// an easy reader-set lattice bug that would reject safe confidentiality
+/// narrowing while potentially allowing the unsafe direction above.
+#[test]
+fn public_session_may_publish_to_a_private_audience() {
+    let public = public_domain(community(1), 10, "community:v1");
+    let private = restricted_domain(community(1), 20, "membership:v1", &[1]);
+    let target = PublicationTarget::from_domain(&private);
+    let session = IfcSession::enter(public);
+
+    assert!(session
+        .publish(PublicationRequest::new(REPLY, target, "public data"))
+        .is_ok());
+}
+
+/// An unknown input permanently poisons ordinary egress. A later labeled read
+/// must not reset the flag and accidentally launder unknown data.
+#[test]
+fn unknown_input_permanently_blocks_publication() {
+    let domain = public_domain(community(1), 10, "community:v1");
+    let resource = ResourceLabel::from_domain(&domain);
+    let target = PublicationTarget::from_domain(&domain);
+    let mut session = IfcSession::enter(domain);
+
+    session.mark_unknown_input();
+    session
+        .read(&resource)
+        .expect("a later labeled read is still admissible");
+    assert_eq!(
+        session
+            .publish(PublicationRequest::new(REPLY, target, "output"))
+            .err(),
+        Some(IfcError::InformationFlow(
+            ifc_core::EgressError::UnresolvedInput
+        ))
+    );
+}
+
+/// Retained restricted state from an older membership snapshot cannot enter a
+/// newly routed session for the same conversation.
+#[test]
+fn same_conversation_rejects_a_stale_membership_epoch() {
+    let old = restricted_domain(community(1), 10, "membership:v1", &[1, 2]);
+    let current = restricted_domain(community(1), 10, "membership:v2", &[1, 2]);
+    let resource = ResourceLabel::from_domain(&old);
+    let mut session = IfcSession::enter(current);
+
+    assert_eq!(session.read(&resource), Err(IfcError::StaleResourceEpoch));
+}
+
+/// Public community data intentionally has no conversation membership epoch,
+/// so a restricted session may read it without comparing unrelated epochs.
+/// This catches the earlier design bug where public data inherited an epoch
+/// and was rejected by every private domain with a different epoch.
+#[test]
+fn private_session_may_read_public_data_from_its_community() {
+    let public = public_domain(community(1), 20, "community:v7");
+    let private = restricted_domain(community(1), 10, "membership:v2", &[1, 2]);
+    let resource = ResourceLabel::from_domain(&public);
+    let mut session = IfcSession::enter(private);
+
+    assert_eq!(session.read(&resource), Ok(()));
+}
+
+/// Owner-private work may explicitly import conversation data when the owner
+/// is an authorized reader. Its output is narrowed to the owner, while another
+/// owner's private state remains protected by the exact-context rule.
+#[test]
+fn owner_private_session_may_read_conversation_data_safe_for_its_owner() {
+    let conversation = restricted_domain(community(1), 20, "membership:v7", &[1, 2]);
+    let owner_private = owner_private_domain(community(1), 10, "membership:v2");
+    let resource = ResourceLabel::from_domain(&conversation);
+    let mut session = IfcSession::enter(owner_private);
+
+    assert_eq!(session.read(&resource), Ok(()));
+}
+
+/// Equal reader sets do not make two conversations the same retained-state
+/// context. Otherwise one private channel could inject history into another.
+#[test]
+fn equal_audiences_do_not_merge_restricted_conversation_contexts() {
+    let source = restricted_domain(community(1), 10, "membership:v1", &[1, 2]);
+    let destination = restricted_domain(community(1), 20, "membership:v1", &[1, 2]);
+    let resource = ResourceLabel::from_domain(&source);
+    let mut session = IfcSession::enter(destination);
+
+    assert_eq!(session.read(&resource), Err(IfcError::ReadContextDenied));
+}
+
+/// Universes are isolated even when two communities happen to contain the
+/// same principals. A target in another community is never a valid IFC sink.
+#[test]
+fn publication_cannot_cross_communities() {
+    let source = public_domain(community(1), 10, "community:v1");
+    let destination = public_domain(community(2), 10, "community:v1");
+    let target = PublicationTarget::from_domain(&destination);
+    let session = IfcSession::enter(source);
+
+    assert!(matches!(
+        session.publish(PublicationRequest::new(REPLY, target, "output")),
+        Err(IfcError::InformationFlow(_))
+    ));
+}
+
+/// This models the broker's retained-session pool. Public channels share one
+/// community domain, while restricted channel identity and membership epoch
+/// each select different state.
+#[test]
+fn broker_routes_retained_sessions_by_complete_domain_key() {
+    let public_a = public_domain(community(1), 10, "community:v1");
+    let public_b = public_domain(community(1), 20, "community:v1");
+    let restricted_a = restricted_domain(community(1), 10, "membership:v1", &[1, 2]);
+    let restricted_b = restricted_domain(community(1), 20, "membership:v1", &[1, 2]);
+    let restricted_new_epoch = restricted_domain(community(1), 10, "membership:v2", &[1, 2]);
+    let public_key = public_a.key();
+    let domains = [
+        public_a,
+        public_b,
+        restricted_a,
+        restricted_b,
+        restricted_new_epoch,
+    ];
+    let mut pool = HashMap::new();
+
+    for domain in domains {
+        pool.entry(domain.key())
+            .or_insert_with(|| IfcSession::enter(domain));
+    }
+
+    assert_eq!(pool.len(), 4);
+    assert_eq!(
+        pool.get(&public_key).map(IfcSession::domain_key),
+        Some(public_key)
+    );
+}