Merge phase 2: pure graph model + taint engine

Impl-plan §4/Phase 2, design v3.4 §6.1-§6.1.3. Pure engine
(evaluate(&GraphSnapshot,&ExclusionCtx,&StickyState)->(Decisions,StickyState)),
never linked against libpipewire; fed by the phase-3 observer to come.

Six adversarial review rounds with Codex (gpt-5.6-sol xhigh). Real echo
leaks found and closed in rounds 1-3 (owner-bridge, sticky-client
contamination, asymmetric forwarders, device mis-classification); my F1
narrowing refuted and conceded in round 4; contract strengthenings in 5-6.
Every fix mutation-verified (reverting it is killed by its intended test).
57 tests, each asserting an exact eligible/excluded partition.

KNOWN v1 LIMITATION (user-accepted 2026-07-22, owed to design doc round 8):
an app that buffers the call, fully tears down its PipeWire objects, and
replays after reconnecting can leak. In-threat-model but contrived; the
fix (process-generation sticky lifetime, revising v3.4 §6.1.3) is deferred
to phase 3's process-liveness work. Documented in src/host/taint/mod.rs.

Phase-3 obligations recorded in the taint module docs and Codex's round-6
report. Unblocks phase 3 (registry observer).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-22 04:42:52 -04:00
co-authored by Claude Opus 4.8
6 changed files with 3825 additions and 0 deletions
+1
View File
@@ -3,6 +3,7 @@ mod capture;
mod pipeline;
mod quality;
mod serve;
pub mod taint;
mod wayland;
mod x11;
+338
View File
@@ -0,0 +1,338 @@
//! Synthetic graph builders for the taint-engine tests.
//!
//! Serials are handed out monotonically and never reused, exactly as
//! PipeWire does; global ids are handed out separately and **may be reused
//! on purpose**, which is what the recycling tests need.
use std::collections::BTreeMap;
use super::snapshot::{
ClientSnapshot, GlobalId, GraphSnapshot, LinkSnapshot, MediaRole, NodeProps, NodeSnapshot,
PortDirection, PortSnapshot, Serial,
};
/// pipewire-pulse's PID, as measured on the target machine.
pub const PULSE_PID: u32 = 2541;
/// WirePlumber's PID — one process owning every device node on the box.
pub const SESSION_PID: u32 = 900;
/// A node's identity in a fixture: what tests pass around.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct NodeRef {
pub serial: Serial,
pub id: GlobalId,
}
#[derive(Default)]
pub struct Graph {
next_serial: u64,
next_id: u32,
nodes: Vec<NodeSnapshot>,
ports: Vec<PortSnapshot>,
links: Vec<LinkSnapshot>,
clients: Vec<ClientSnapshot>,
/// One client connection per process / per module, which is what the
/// live graph looks like. Tests that need the *split*-client shape
/// (GStreamer opens one per stream) pass clients explicitly instead.
client_by_app: BTreeMap<u32, GlobalId>,
client_by_module: BTreeMap<u64, GlobalId>,
session_client: Option<GlobalId>,
}
impl Graph {
pub fn new() -> Self {
Self {
// Start past u32::MAX so every fixture also exercises the phase
// 0a widening: a serial that a u32 model would have truncated.
next_serial: u64::from(u32::MAX) + 1,
next_id: 1,
..Self::default()
}
}
fn serial(&mut self) -> Serial {
self.next_serial += 1;
Serial(self.next_serial)
}
fn id(&mut self) -> GlobalId {
self.next_id += 1;
GlobalId(self.next_id)
}
/// A client object. `sec_pid` is `pipewire.sec.pid` — pipewire-pulse's
/// PID for Pulse-emulated clients.
pub fn client(&mut self, sec_pid: Option<u32>) -> GlobalId {
let serial = self.serial();
let id = self.id();
self.clients.push(ClientSnapshot {
serial,
id,
sec_pid,
});
id
}
/// The client connection an ordinary process holds — one per PID,
/// created on demand.
pub fn client_of_app(&mut self, pid: u32) -> GlobalId {
if let Some(id) = self.client_by_app.get(&pid) {
return *id;
}
let id = self.client(Some(PULSE_PID));
self.client_by_app.insert(pid, id);
id
}
/// An ordinary application stream: its own client, its own PID.
pub fn app_node(&mut self, name: &str, role: MediaRole, pid: u32) -> NodeRef {
let client = self.client_of_app(pid);
self.node(name, role, app(client, pid))
}
/// The client a pactl module holds. Measured: each module gets its own
/// (`sink-sunshine-*` were clients 83/86/92), which is why one tainted
/// module does not fuse with the next.
pub fn client_of_module(&mut self, module: u64) -> GlobalId {
match self.client_by_module.get(&module) {
Some(id) => *id,
None => {
let id = self.client(Some(PULSE_PID));
self.client_by_module.insert(module, id);
id
}
}
}
/// A leg of a pactl-loaded module: one client per module, and the
/// node's `application.process.id` is **pipewire-pulse's own**, because
/// pipewire-pulse genuinely is the client.
pub fn module_node(&mut self, name: &str, role: MediaRole, module: u64) -> NodeRef {
let client = self.client_of_module(module);
self.node(name, role, pulse_module(client, module, PULSE_PID))
}
/// A leg joined to its siblings by `node.link-group` — loopback,
/// filter-chain, echo-cancel.
pub fn group_node(&mut self, name: &str, role: MediaRole, group: &str, pid: u32) -> NodeRef {
let client = self.client_of_app(pid);
self.node(name, role, link_group(group, client, pid))
}
/// A device node as the session manager creates it: no strong key,
/// WirePlumber's client and PID — shared with every other device — and
/// a `device.id`, which is what marks it as session-manager-exported.
pub fn device_node(&mut self, name: &str, role: MediaRole) -> NodeRef {
let session = match self.session_client {
Some(id) => id,
None => {
let id = self.client(None);
self.session_client = Some(id);
id
}
};
self.node(name, role, device(session, SESSION_PID))
}
/// A node that *belongs to* a Device but is not a passive device node —
/// a filter associated with a card. Phase 3 must not classify this as a
/// session device, or it loses both its coarse owner keys and its
/// ability to trip the fail-closed backstop.
pub fn device_associated_filter(&mut self, name: &str, role: MediaRole, pid: u32) -> NodeRef {
let client = self.client_of_app(pid);
self.node(name, role, app(client, pid))
}
/// A **virtual** sink an application created natively: an `Audio/Sink`
/// with no `device.id` and no strong key, sharing one client with the
/// stream that re-emits what it receives. Coarse keys must still bridge
/// these two, or the whole call leaks through the re-emitting leg.
pub fn native_virtual_node(&mut self, name: &str, role: MediaRole, pid: u32) -> NodeRef {
let client = self.client_of_app(pid);
self.node(name, role, app(client, pid))
}
pub fn peerspeak_node(&mut self, name: &str, pid: u32) -> NodeRef {
let client = self.client_of_app(pid);
self.node(name, MediaRole::StreamOutput, peerspeak_owned(client, pid))
}
pub fn node(&mut self, name: &str, role: MediaRole, props: NodeProps) -> NodeRef {
let id = self.id();
self.node_with_id(name, role, id, props)
}
/// Force a global id — for reproducing id recycling after teardown.
pub fn node_with_id(
&mut self,
name: &str,
role: MediaRole,
id: GlobalId,
props: NodeProps,
) -> NodeRef {
let serial = self.serial();
self.nodes.push(NodeSnapshot {
serial,
id,
name: Some(name.to_string()),
role,
props,
});
NodeRef { serial, id }
}
pub fn port(&mut self, node: NodeRef, direction: PortDirection, exclusive: bool) {
let serial = self.serial();
let id = self.id();
self.ports.push(PortSnapshot {
serial,
id,
node: node.id,
direction,
exclusive,
monitor: false,
});
}
/// A signal edge: audio flows `from → to`.
pub fn link(&mut self, from: NodeRef, to: NodeRef) {
self.link_ids(from.id, to.id);
}
/// A link naming raw ids, so a test can dangle an endpoint.
pub fn link_ids(&mut self, from: GlobalId, to: GlobalId) {
let serial = self.serial();
let id = self.id();
self.links.push(LinkSnapshot {
serial,
id,
output_node: from,
input_node: to,
output_port: None,
input_port: None,
});
}
/// An id that belongs to nothing — for unresolved-endpoint tests.
pub fn dangling_id(&mut self) -> GlobalId {
self.id()
}
pub fn build(&self) -> GraphSnapshot {
self.build_without(&[])
}
/// A later snapshot in which some nodes have gone away, along with
/// their ports and every link touching them. Surviving objects keep
/// their serials, which is what makes sticky-taint sequences testable.
pub fn build_without(&self, dropped: &[NodeRef]) -> GraphSnapshot {
let gone_serials: Vec<Serial> = dropped.iter().map(|n| n.serial).collect();
let nodes: Vec<NodeSnapshot> = self
.nodes
.iter()
.filter(|n| !gone_serials.contains(&n.serial))
.cloned()
.collect();
// Filter by what was *dropped*, not by what is live: a link to an id
// that never had a node is a dangling endpoint, and dropping those
// here would quietly disarm every unresolved-ancestry test.
let gone_ids: Vec<GlobalId> = dropped.iter().map(|n| n.id).collect();
GraphSnapshot::new(
nodes,
self.ports
.iter()
.filter(|p| !gone_ids.contains(&p.node))
.cloned()
.collect(),
self.links
.iter()
.filter(|l| !gone_ids.contains(&l.output_node) && !gone_ids.contains(&l.input_node))
.cloned()
.collect(),
self.clients.clone(),
)
}
/// Drop clients too — full owner teardown.
///
/// Invalidates the per-app/per-module caches as well: leaving them
/// stale made a later `client_of_app` hand back the *removed* client's
/// id, so a test that meant "a brand-new client after teardown" was
/// really building a node pointing at a client object that no longer
/// existed (Codex round 1, finding 8).
pub fn drop_clients(&mut self, ids: &[GlobalId]) {
self.clients.retain(|c| !ids.contains(&c.id));
self.client_by_app.retain(|_, id| !ids.contains(id));
self.client_by_module.retain(|_, id| !ids.contains(id));
if self.session_client.is_some_and(|id| ids.contains(&id)) {
self.session_client = None;
}
}
/// A client that reuses a global id a dead client had — the recycling
/// case, with a fresh serial.
pub fn client_with_id(&mut self, id: GlobalId, sec_pid: Option<u32>) -> GlobalId {
let serial = self.serial();
self.clients.push(ClientSnapshot {
serial,
id,
sec_pid,
});
id
}
}
/// An ordinary application stream: real PID, one client connection.
pub fn app(client: GlobalId, pid: u32) -> NodeProps {
NodeProps {
client_id: Some(client),
process_id: Some(pid),
..NodeProps::default()
}
}
/// A pactl-module-created stream: the daemon is the client, so the node's
/// `application.process.id` is pipewire-pulse's own.
pub fn pulse_module(client: GlobalId, module: u64, pulse_pid: u32) -> NodeProps {
NodeProps {
pulse_module_id: Some(module),
client_id: Some(client),
process_id: Some(pulse_pid),
..NodeProps::default()
}
}
/// A PipeWire-module leg joined to its siblings by `node.link-group`
/// (loopback, filter-chain, echo-cancel).
pub fn link_group(group: &str, client: GlobalId, pid: u32) -> NodeProps {
NodeProps {
link_group: Some(group.to_string()),
client_id: Some(client),
process_id: Some(pid),
..NodeProps::default()
}
}
/// A device node as the session manager creates it: no strong key, and the
/// session manager's own client and PID — shared with every other device.
///
/// Measured 2026-07-21: real ALSA device nodes carry the shared
/// `client.id` but **no** `application.process.id` at all. Giving them one
/// here is deliberately *more* pessimistic than reality — it hands the
/// engine a second coarse key it could fuse devices on, so a test that
/// passes here also passes against the real props.
pub fn device(session_client: GlobalId, session_pid: u32) -> NodeProps {
NodeProps {
client_id: Some(session_client),
process_id: Some(session_pid),
session_device: true,
..NodeProps::default()
}
}
pub fn peerspeak_owned(client: GlobalId, pid: u32) -> NodeProps {
NodeProps {
peerspeak_owned: true,
..app(client, pid)
}
}
+956
View File
@@ -0,0 +1,956 @@
//! The taint engine — decides which `Stream/Output/Audio` nodes may be
//! fanned out into the screen-share capture without echoing peerspeak's own
//! audio back at the viewer.
//!
//! Implements design v3.4 §6.1–§6.1.3 (`peerspeak/docs/
//! screenshare-audio-exclusion-plan.md`), phase 2 of the implementation
//! plan. **Pure**: no PipeWire types appear in any signature, nothing here
//! touches the daemon, and every test builds its own graph.
//!
//! ## The one-sentence predicate
//!
//! > A node is eligible only if **no** signal path reaches it from a
//! > peerspeak-owned node, the live AEC identity, or any pixelpass-owned
//! > object. **Unresolvable ancestry is not eligible.**
//!
//! That last sentence is the invariant the whole design rests on: every
//! other failure mode in here degrades into over-exclusion (one app's audio
//! silently missing from the share) rather than into echo.
//!
//! ## Why a graph walk and not a property check
//!
//! Exclusion does not propagate downstream by itself. Any node that
//! re-emits audio it received is a fresh, *untagged* `Stream/Output/Audio`
//! carrying the mix — including the one peerspeak playback stream that was
//! correctly excluded one hop earlier. EasyEffects, `module-loopback`,
//! combine-sinks, tunnel/RTP sinks and virtual-sink forwarders all have this
//! shape, and at least one such topology has been observed live on the
//! target machine.
//!
//! Taint therefore flows over **three** edge types:
//!
//! 1. **Link edges** — `link.output.node → link.input.node`.
//! 2. **Sink → monitor** — free at node granularity: the monitor connection
//! *is* a real Link whose output node is the sink node itself (measured).
//! A port-granular walk would need a synthetic edge; a node-granular one
//! does not.
//! 3. **Owner bridges** — the intra-process hop the graph cannot see. See
//! [`owner`]; this is the hard one.
//!
//! ## Stickiness
//!
//! Taint is **sticky per owner** for the duration of the share, because a
//! topological recompute forgets *buffered* audio: an app can read a tainted
//! monitor into a 5-second ring buffer, then have its input leg vanish, and
//! a purely topological engine would relink its output while it is still
//! emitting peerspeak's audio out of that buffer. No graph event marks the
//! moment a buffer drains.
//!
//! Stickiness is keyed on [`Serial`] — never on a node id, `client.id`,
//! module index or `link-group` string, **all of which recycle on this
//! stack**. An entry is cleared only once every member object has
//! disappeared; a key that reappears after full teardown is a new owner and
//! starts clean.
//!
//! ## ⚠️ KNOWN OPEN GAP — buffered audio across a full PipeWire teardown of
//! ## a still-live process (Codex phase-2 rounds 56) — DESIGN DECISION OWED
//!
//! **This is an in-threat-model echo gap, not an outside-the-model one — an
//! earlier version of this note wrongly scoped it to keyless streams.**
//!
//! The scenario, entirely with a real PID-bearing app (a recorder, a DAW,
//! a GStreamer pipeline): it reads the call into an application buffer,
//! **fully** tears down its PipeWire Node *and* Client while keeping that
//! buffer, then — still the same live process — opens a fresh Client and a
//! `Stream/Output/Audio` and replays. Every old serial is gone, so
//! [`seed_sticky`] refuses to apply the remembered PID fingerprint (the
//! fingerprint is lifetime-scoped to a live serial member, because bare keys
//! recycle); no reader is live in the new epoch, so the backstop does not
//! fire; the replayed leg is eligible.
//!
//! It is real and reachable by non-adversarial software. It also sits
//! exactly on the design's stated boundary (v3.4 §6.1.3: "a key that
//! reappears after full teardown is a new owner and starts clean"), so
//! closing it is a **design change**, not a local bug fix:
//!
//! - **Option A — accept as a documented v1 limitation.** Contrived in
//! practice (most apps hold their PipeWire connection open for their
//! lifetime; the round-2 fix already covers the common
//! idle-a-client-and-open-another case), never a *silent* correctness
//! regression since it is written down, and phase 5's dry run would show
//! it. But it is a known echo path, which sits badly against the feature's
//! fail-closed ethos.
//! - **Option B — process-generation lifetime.** Key the fingerprint's
//! lifetime on the owning **process** being alive — PID + `/proc` start
//! time (or a pidfd) to defeat PID reuse — instead of on a live PipeWire
//! object. Phase 3 supplies process liveness; §6.1.3's node/client-only
//! lifetime definition is revised. Closes the PID-bearing case; the truly
//! keyless sub-case (no PID at all) genuinely *is* outside the threat
//! model and stays a documented limit.
//!
//! The choice is the designer's (it revises the security surface). Until it
//! is made, `a_fingerprint_does_not_outlive_its_owner` encodes Option A's
//! behaviour — flip it if B is chosen. Owed to the design doc as round 8.
// Phase 2 lands the engine behind its own test surface and nothing else:
// the registry observer that will feed it is phase 3, so in a non-test
// build every item here is legitimately unreachable for now.
#![allow(dead_code)]
pub mod owner;
pub mod snapshot;
#[cfg(test)]
mod fixture;
#[cfg(test)]
mod tests;
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use owner::{OwnerComponents, OwnerKey};
use snapshot::{GraphSnapshot, IdLookup, MediaRole, NodeSnapshot, Serial};
/// The `node.name` prefix of a pixelpass capture sink. Any host's sink
/// counts, not just ours — fanning out a stream that is downstream of
/// *another* pixelpass host's capture sink builds a cycle (v3.4 §6.2).
pub const CAPTURE_SINK_PREFIX: &str = "pixelpass_capture_";
/// `node.link-group` prefix that marks *some* echo canceller. Hazard
/// detection only — it does **not** identify peerspeak's instance, which is
/// what `pulse.module.id` is for (v3.4 §5.2 correction 4).
pub const ECHO_CANCEL_GROUP_PREFIX: &str = "echo-cancel-";
/// Why a node is tainted or excluded. Stable machine-readable codes: this
/// value is the phase 5 audit output, the phase 6 status event, and the
/// eventual answer to "why isn't this app being shared?".
#[derive(Clone, Copy, PartialEq, Eq, Debug, Hash)]
pub enum Reason {
/// Carries the `peerspeak.owned` tag (v3.4 §5.1).
PeerspeakOwned,
/// `pulse.module.id` equals the live AEC module index — exact equality
/// only. "Has any `pulse.module.id`" is explicitly rejected as a rule:
/// tunnel/RTP/loopback modules may be the only carrier of audio the
/// user legitimately wants shared (v3.4 §5.2 correction 2).
AecIdentity,
/// A pixelpass-owned object, ours or another host's capture sink.
PixelpassOwned,
/// An `echo-cancel-*` group that is **not** our validated identity.
/// Decision D3: warn and exclude rather than fan out.
ForeignEchoCancel,
/// Reached by a signal path from a tainted node (link or monitor edge).
TaintedUpstream,
/// Reached across an owner bridge; the key that did it, when the
/// tainted member shares one directly rather than transitively.
TaintedOwnerBridge { key: Option<OwnerKey> },
/// A link endpoint, or a node's own id, could not be resolved in this
/// snapshot. Fail closed (v3.4 §6.1.4).
UnresolvedAncestry,
/// A tainted capture stream whose owner cannot be bounded by any usable
/// key, so its sibling output legs cannot be identified. Fail closed
/// (v3.4 §6.1.1, final paragraph).
UnresolvedOwner,
/// The observer has not reached a complete, coherent view of the graph
/// yet. No decision made from a partial graph is a decision.
GraphNotReady,
/// A `port.exclusive` port — fan-out will be refused (v3.4 §6.2). Local
/// to the node; does not propagate.
PortExclusive,
/// An encoded/passthrough stream — a second link would corrupt it.
/// Local to the node; does not propagate.
Passthrough,
}
impl Reason {
pub fn code(self) -> &'static str {
match self {
Self::PeerspeakOwned => "peerspeak-owned",
Self::AecIdentity => "aec-identity",
Self::PixelpassOwned => "pixelpass-owned",
Self::ForeignEchoCancel => "foreign-echo-cancel",
Self::TaintedUpstream => "tainted-upstream",
Self::TaintedOwnerBridge { .. } => "tainted-owner-bridge",
Self::UnresolvedAncestry => "unresolved-ancestry",
Self::UnresolvedOwner => "unresolved-owner",
Self::GraphNotReady => "graph-not-ready",
Self::PortExclusive => "port-exclusive",
Self::Passthrough => "passthrough",
}
}
/// Lower wins. A node can acquire taint several ways in one recompute
/// and the reported reason must not depend on traversal order, or the
/// audit output is unstable and the fixture tests are flaky. Explicit
/// priority, not BFS arrival order.
fn priority(self) -> u8 {
match self {
Self::PeerspeakOwned => 0,
Self::AecIdentity => 1,
Self::PixelpassOwned => 2,
Self::ForeignEchoCancel => 3,
Self::TaintedUpstream => 4,
Self::TaintedOwnerBridge { .. } => 5,
Self::UnresolvedAncestry => 6,
Self::UnresolvedOwner => 7,
// Non-propagating; never competes with the taint reasons above
// because it is only consulted for untainted candidates.
Self::GraphNotReady => 8,
Self::PortExclusive => 9,
Self::Passthrough => 10,
}
}
/// Does this reason spread to downstream nodes and owner siblings?
fn propagates(self) -> bool {
self.priority() <= Self::UnresolvedOwner.priority()
}
}
/// Everything the engine needs that is not in the graph itself.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ExclusionCtx {
/// The **validated** live AEC module index, or `None` for `--aec=off`.
/// The validation state machine (phase 4) owns the transitions; if it
/// is still `Validating` or has `Failed`, its caller must not fan out at
/// all rather than passing `None` here, which would merely mean "there
/// is no AEC".
pub aec_module_id: Option<u64>,
/// pipewire-pulse's own PID, derived by the observer (phase 3) from a
/// consistent `pipewire.sec.pid` across Pulse clients validated against
/// `/proc/<pid>/comm`. `None` is safe but coarse — see [`owner`].
pub pipewire_pulse_pid: Option<u32>,
/// Serials of objects pixelpass itself created this run.
pub pixelpass_owned: BTreeSet<Serial>,
/// False until the readiness epoch has been reached (phase 3). Every
/// candidate is then ineligible: a decision from a partial graph is not
/// a decision.
pub graph_ready: bool,
}
/// Object identity for sticky bookkeeping. Always a [`Serial`] — never a
/// recyclable id (v3.4 §6.1.3).
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug)]
pub enum ObjectRef {
Node(Serial),
Client(Serial),
}
/// One owner that has been tainted, and every object observed to constitute
/// it. Cleared only when **all** of them are gone.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StickyOwner {
/// Every object seen to be part of this owner, ever. Membership
/// accumulates: that is what makes "clear only once all member objects
/// have disappeared" true across churn.
pub members: BTreeSet<ObjectRef>,
/// Owner keys remembered across connections — strong keys and a usable
/// process id, never `client.id`. Applied only while some serial member
/// above is still live, which is what keeps a recyclable key from
/// resurrecting a dead owner.
///
/// Needed because a live Client is not the same thing as a live owner:
/// a process can leave one connection idle and open a second, and
/// GStreamer opens one connection per stream as a matter of course, so
/// following connections alone lets the next leg escape (Codex round 2,
/// finding 2).
pub fingerprints: BTreeSet<owner::Fingerprint>,
/// The reason recorded for each node that was tainted in its own right.
/// Kept per node rather than collapsed to one owner-wide reason, or a
/// forwarder's output leg inherits its *input* leg's `tainted-upstream`
/// and the audit output stops naming the mechanism that actually
/// excluded it.
pub node_reasons: BTreeMap<Serial, Reason>,
}
impl StickyOwner {
/// The reason to apply to a member: its own recorded one, or — for a
/// leg that appeared later — the fact that it belongs to a tainted
/// owner, which is a bridge by definition.
fn reason_for(&self, serial: Serial) -> Reason {
self.node_reasons
.get(&serial)
.copied()
.unwrap_or(Reason::TaintedOwnerBridge { key: None })
}
}
/// Threaded explicitly through [`evaluate`] so stickiness is testable as a
/// sequence of snapshots rather than as hidden mutable state.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct StickyState {
pub owners: Vec<StickyOwner>,
}
impl StickyState {
pub fn is_empty(&self) -> bool {
self.owners.is_empty()
}
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Eligibility {
Eligible,
NotEligible {
reason: Reason,
/// The taint was carried over from a previous snapshot rather than
/// derived from the current topology.
sticky: bool,
},
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct NodeDecision {
pub serial: Serial,
pub name: Option<String>,
pub eligibility: Eligibility,
}
impl NodeDecision {
pub fn is_eligible(&self) -> bool {
matches!(self.eligibility, Eligibility::Eligible)
}
pub fn reason(&self) -> Option<Reason> {
match self.eligibility {
Eligibility::Eligible => None,
Eligibility::NotEligible { reason, .. } => Some(reason),
}
}
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct TaintEntry {
pub reason: Reason,
pub sticky: bool,
}
/// The result of one recompute.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Decisions {
/// Every `Stream/Output/Audio` node in the snapshot — the complete
/// candidate universe, so callers can assert an exact partition rather
/// than spot-checking named nodes.
pub candidates: BTreeMap<Serial, NodeDecision>,
/// Taint over *all* node roles, for diagnostics and for the phase 5
/// audit output.
pub taint: BTreeMap<Serial, TaintEntry>,
}
impl Decisions {
/// Serials of eligible candidates, ascending.
pub fn eligible(&self) -> Vec<Serial> {
self.candidates
.values()
.filter(|d| d.is_eligible())
.map(|d| d.serial)
.collect()
}
/// `(serial, reason code)` for excluded candidates, ascending.
pub fn excluded(&self) -> Vec<(Serial, &'static str)> {
self.candidates
.values()
.filter_map(|d| d.reason().map(|r| (d.serial, r.code())))
.collect()
}
}
/// Recompute eligibility for the whole graph.
///
/// Full recompute per graph event is the v1 design; there is deliberately
/// no incremental dirty-set.
///
/// ⚠️ **Cost is not O(V+E), despite what v3.4 §6.4 says.** Each fixpoint
/// pass re-runs a full link BFS *and* a full owner scan, and the bridge
/// scans every tainted source in a component for each target, so the bound
/// is `O(D · (V + E + Σ_C |sources_C|·|targets_C|))` — worst case
/// `O(D · (V² + E))` — for an owner-bridge depth D. D is 1 for every
/// topology observed so far and 2 for a forwarder feeding a forwarder, and
/// components on a real desktop are two or three nodes; the quadratic term
/// needs one owner with many legs. A 60-layer chain test guards the depth
/// dimension only. Phase 5 records the real recompute-duration
/// distribution and maximum, which is what "full recompute is fine for v1"
/// should rest on — measured headroom, not a node count.
pub fn evaluate(
snapshot: &GraphSnapshot,
ctx: &ExclusionCtx,
prior: &StickyState,
) -> (Decisions, StickyState) {
let components = OwnerComponents::build(snapshot, ctx.pipewire_pulse_pid);
let keys = owner::OwnerKeyIndex::build(snapshot, ctx.pipewire_pulse_pid);
let mut taint: BTreeMap<Serial, Reason> = BTreeMap::new();
let mut sticky_serials: BTreeSet<Serial> = BTreeSet::new();
seed_local_roots(snapshot, ctx, &mut taint);
seed_sticky(
snapshot,
&keys,
prior,
&components,
&mut taint,
&mut sticky_serials,
);
// Monotone fixpoint: every step only adds taint, or lowers a node's
// reason priority, both of which are bounded. Link propagation and the
// owner bridge feed each other — a bridged output leg has downstream
// links, and a downstream monitor reader bridges to its own siblings —
// so neither can be run once.
let edges = downstream_edges(snapshot, &mut taint);
loop {
let mut changed = false;
changed |= propagate_links(&edges.edges, &mut taint);
changed |= propagate_owner_bridge(&keys, &components, &edges, &mut taint);
changed |= propagate_unresolved_owner(snapshot, &keys, &edges, &mut taint);
if !changed {
break;
}
}
let decisions = build_decisions(snapshot, ctx, &taint, &sticky_serials);
// ⚠️ Readiness gates **retirement only**, never addition (Codex rounds
// 1 and 2, which caught the two halves of this in turn). An object
// missing from an untrustworthy snapshot has not been observed to
// disappear, so retiring on that basis erases history and the next
// ready recompute hands back a clean bill of health. But taint
// *observed* during a not-ready epoch is real — a reader can consume
// and buffer the call and then vanish before readiness — so discarding
// additions was the same defect pointing the other way.
let next_sticky = build_sticky(snapshot, &keys, &components, &taint, prior, ctx.graph_ready);
(decisions, next_sticky)
}
/// Roots that are visible on the node itself.
fn seed_local_roots(
snapshot: &GraphSnapshot,
ctx: &ExclusionCtx,
taint: &mut BTreeMap<Serial, Reason>,
) {
for node in snapshot.nodes() {
if let Some(reason) = local_root_reason(node, ctx) {
raise(taint, node.serial, reason);
}
// A node whose own global id is ambiguous cannot be the reliable
// endpoint of any link, so its ancestry is unresolvable.
if snapshot.node_by_id(node.id) == Some(IdLookup::Ambiguous) {
raise(taint, node.serial, Reason::UnresolvedAncestry);
}
}
}
fn local_root_reason(node: &NodeSnapshot, ctx: &ExclusionCtx) -> Option<Reason> {
if node.props.peerspeak_owned {
return Some(Reason::PeerspeakOwned);
}
if let (Some(module), Some(aec)) = (node.props.pulse_module_id, ctx.aec_module_id)
&& module == aec
{
return Some(Reason::AecIdentity);
}
if ctx.pixelpass_owned.contains(&node.serial)
|| node
.name
.as_deref()
.is_some_and(|name| name.starts_with(CAPTURE_SINK_PREFIX))
{
return Some(Reason::PixelpassOwned);
}
if node
.props
.link_group
.as_deref()
.is_some_and(|group| group.starts_with(ECHO_CANCEL_GROUP_PREFIX))
{
return Some(Reason::ForeignEchoCancel);
}
None
}
/// Carry taint forward from previous snapshots (v3.4 §6.1.3).
///
/// An owner is re-seeded from three kinds of evidence, all lifetime-scoped
/// to a still-live member: its own surviving nodes, nodes on a surviving
/// **Client**, and nodes presenting a remembered owner **fingerprint**.
fn seed_sticky(
snapshot: &GraphSnapshot,
keys: &owner::OwnerKeyIndex,
prior: &StickyState,
components: &OwnerComponents,
taint: &mut BTreeMap<Serial, Reason>,
sticky_serials: &mut BTreeSet<Serial>,
) {
for entry in &prior.owners {
let mut live_nodes: Vec<Serial> = Vec::new();
for member in &entry.members {
match member {
ObjectRef::Node(serial) => {
if snapshot.node(*serial).is_some() {
live_nodes.push(*serial);
}
}
// A surviving **Client** re-seeds too. An app can close
// every stream it had while keeping its PipeWire connection
// open, then open a fresh one — Firefox does exactly this.
ObjectRef::Client(serial) => {
live_nodes.extend(nodes_of_client(snapshot, keys, *serial));
}
}
}
if live_nodes.is_empty() && !entry.members.iter().any(|m| is_live(snapshot, *m)) {
// Nothing of this owner remains; its fingerprints are just
// recyclable strings now and must not be applied to anyone.
continue;
}
// Fingerprints reach a *new connection* of the same still-live
// process, which neither of the two paths above can see.
for fingerprint in &entry.fingerprints {
live_nodes.extend(
snapshot
.nodes()
.filter(|node| keys.has_fingerprint(node.serial, fingerprint))
.map(|node| node.serial),
);
}
// The owner is sticky, not the individual node: a leg that appears
// later in the same still-live owner inherits the taint.
for serial in live_nodes {
for member in components.members_with(serial) {
let reason = entry.reason_for(*member);
if raise(taint, *member, reason) || taint.get(member) == Some(&reason) {
sticky_serials.insert(*member);
}
}
}
}
}
/// Nodes currently attached to a client, by the client's **serial**. The
/// client's snapshot-local id is resolved fresh each time, so a recycled id
/// can never resurrect a dead owner.
///
/// Nodes for which `client.id` is not a usable owner key — session-manager
/// device nodes — are excluded, or the shared `WirePlumber [export]` Client
/// would drag every sound card on the box into one sticky owner.
///
/// The same gate is applied when *recording* clients into a sticky entry
/// (`owner::client_serials_of`). Either one alone closes the leak; both are
/// kept because they answer different questions ("may this client be
/// remembered?" and "may this client speak for that node?"), and the
/// regression test kills the removal of the pair.
fn nodes_of_client(
snapshot: &GraphSnapshot,
keys: &owner::OwnerKeyIndex,
client: Serial,
) -> Vec<Serial> {
let Some(id) = snapshot
.clients()
.find(|c| c.serial == client)
.map(|c| c.id)
else {
return Vec::new();
};
snapshot
.nodes()
.filter(|node| node.props.client_id == Some(id))
.filter(|node| keys.uses_client_key(node.serial))
.map(|node| node.serial)
.collect()
}
/// `output node → input nodes`, resolving snapshot-local ids. An endpoint
/// that does not resolve taints the *other* end as unresolved ancestry when
/// that other end is the input side — we cannot know what is feeding it.
fn downstream_edges(snapshot: &GraphSnapshot, taint: &mut BTreeMap<Serial, Reason>) -> Edges {
let mut edges: BTreeMap<Serial, Vec<Serial>> = BTreeMap::new();
let mut receivers: BTreeSet<Serial> = BTreeSet::new();
for link in snapshot.links() {
let from = snapshot.node_by_id(link.output_node);
let to = snapshot.node_by_id(link.input_node);
match (from, to) {
(Some(IdLookup::Unique(from)), Some(IdLookup::Unique(to))) => {
edges.entry(from).or_default().push(to);
receivers.insert(to);
}
(_, Some(IdLookup::Unique(to))) => {
// Something feeds this node and we cannot say what.
raise(taint, to, Reason::UnresolvedAncestry);
receivers.insert(to);
}
(_, Some(IdLookup::Ambiguous)) => {
// Several nodes claim the input id and we cannot say which
// one this link feeds, so every claimant is a receiver.
// They are already tainted as unresolved by their own
// ambiguous id — but taint without receiver status cannot
// start an owner bridge, so their sibling output legs stayed
// Eligible (Codex round 2, finding 3).
receivers.extend(
snapshot
.nodes_with_id(link.input_node)
.map(|node| node.serial),
);
}
_ => {}
}
}
for targets in edges.values_mut() {
targets.sort_unstable();
targets.dedup();
}
// A node that receives audio by *role* counts even with no inbound link
// yet: a pixelpass capture sink is a taint root the moment it exists,
// and its owner's re-emitting leg must be bridged from it immediately.
receivers.extend(
snapshot
.nodes()
.filter(|node| node.role.receives_audio())
.map(|node| node.serial),
);
Edges { edges, receivers }
}
/// Resolved signal edges plus the set of nodes that can receive audio.
struct Edges {
edges: BTreeMap<Serial, Vec<Serial>>,
/// ⚠️ Membership is "appears as a resolved `link.input.node`" **or**
/// "has a receiving role" — deliberately not role alone. Codex round 1:
/// a node whose `media.class` is absent or unexpected (`Other`), or an
/// `Audio/Source` that is really a filter output, can sit on an inbound
/// link carrying tainted audio; inferring "receives audio" from the role
/// alone left such a node unable to start an owner bridge, and its
/// sibling output leg stayed Eligible while re-emitting the call.
receivers: BTreeSet<Serial>,
}
fn propagate_links(
downstream: &BTreeMap<Serial, Vec<Serial>>,
taint: &mut BTreeMap<Serial, Reason>,
) -> bool {
let mut changed = false;
let mut queue: VecDeque<Serial> = taint
.iter()
.filter(|(_, reason)| reason.propagates())
.map(|(serial, _)| *serial)
.collect();
while let Some(serial) = queue.pop_front() {
let Some(targets) = downstream.get(&serial) else {
continue;
};
for target in targets {
if raise(taint, *target, Reason::TaintedUpstream) {
changed = true;
queue.push_back(*target);
}
}
}
changed
}
/// The conditional owner bridge (v3.4 §6.1.1): taint crosses to an owner's
/// other legs **only** when the tainted member is one that actually
/// receives audio. The naive "this owner has both an input and an output
/// leg ⇒ exclude the output" rule would exclude every app using a
/// microphone, Firefox in a video call included.
fn propagate_owner_bridge(
keys: &owner::OwnerKeyIndex,
components: &OwnerComponents,
edges: &Edges,
taint: &mut BTreeMap<Serial, Reason>,
) -> bool {
let mut changed = false;
for members in components.components() {
let sources: BTreeSet<Serial> = members
.iter()
.copied()
.filter(|serial| {
taint.get(serial).is_some_and(|r| r.propagates())
&& edges.receivers.contains(serial)
})
.collect();
if sources.is_empty() {
continue;
}
for target in members {
if sources.contains(target) {
continue;
}
// Name the strongest key shared directly with any tainted
// member; `None` means the two are only transitively related.
let key = sources
.iter()
.filter_map(|source| keys.strongest_shared(*source, *target))
.min();
changed |= raise(taint, *target, Reason::TaintedOwnerBridge { key });
}
}
changed
}
/// Fail-closed backstop for an owner we cannot bound (v3.4 §6.1.1, final
/// paragraph): something read tainted audio and nothing about the output
/// legs on this box lets us enumerate which of them are its siblings, so we
/// cannot know which one is re-emitting what it read. Exclude the output
/// legs that are equally unbounded.
///
/// The trigger and the sweep, precisely (both edges hard-won across four
/// Codex rounds):
///
/// - **Trigger — any tainted receiver that is not a real device node.** A
/// tainted hardware sink is the normal case, not an anomaly (peerspeak's
/// playback taints the default sink every recompute), so device nodes do
/// not trip it. The source does **not** have to be unbounded: a reader
/// with a `node.link-group` whose re-emitting leg carries none is bounded
/// while its sibling is unfindable (round 1).
/// - **Sweep — depends on whether any tainted reader is itself unbounded.**
/// A *bounded* reader's siblings are exactly the outputs sharing its key,
/// so only the unbounded outputs (which could share its unknowable-only-
/// in-part identity) are swept; a differently-keyed output is provably a
/// different owner. An *unbounded* reader could be **any** owner — a real
/// process may present no PID on its reading leg (round 4) — so every
/// output candidate is swept, real apps included.
///
/// **Two tiers, because a tainted reader we cannot bound is a bigger
/// unknown than one we can** (Codex round 3 — the mirror image of the
/// round-1 case):
///
/// - A *bounded* tainted reader has a strong key or a usable PID, so its
/// siblings are exactly the output legs sharing that key. Any output leg
/// that is *itself* bounded by a **different** key is provably a different
/// owner and stays eligible; only unbounded output legs are its possible
/// siblings. → exclude unbounded outputs.
/// - An *unbounded* tainted reader has nothing that identifies its owner, so
/// its re-emitting leg could be **any** output on the box, and no property
/// on an output leg can prove it is unrelated. → exclude every output
/// candidate.
///
/// ⚠️ I tried to narrow this to "daemon-owned outputs only", on the
/// theory that an unbounded reader must be daemon-owned (a real app has a
/// PID, which would bound it) so a real-PID output is provably a different
/// owner. **Codex refuted it (round 4):** `application.process.id` is
/// optional and client-controlled, so a real process can present *no* PID
/// on its reading leg (unbounded) and a real PID on its output leg — one
/// owner, spared by the narrowing, leaking the call. Only `pipewire.*`
/// properties have protected identity; app properties cannot carry a
/// soundness argument. So: exclude everything. The trigger is genuinely
/// anomalous — a keyless reader actively consuming the call; EasyEffects
/// and loopbacks carry a `node.link-group` and are *bounded*, so they do
/// not trip this tier — and phase 5's dry run surfaces it before it can
/// gate anything real.
fn propagate_unresolved_owner(
snapshot: &GraphSnapshot,
keys: &owner::OwnerKeyIndex,
edges: &Edges,
taint: &mut BTreeMap<Serial, Reason>,
) -> bool {
let mut has_tainted_reader = false;
let mut has_unbounded_tainted_reader = false;
for node in snapshot.nodes() {
let is_tainted_reader = !node.props.session_device
&& edges.receivers.contains(&node.serial)
&& taint.get(&node.serial).is_some_and(|r| r.propagates());
if is_tainted_reader {
has_tainted_reader = true;
has_unbounded_tainted_reader |= !keys.is_bounded(node.serial);
}
}
if !has_tainted_reader {
return false;
}
let mut changed = false;
for node in snapshot.nodes() {
if node.role == MediaRole::StreamOutput
&& (has_unbounded_tainted_reader || !keys.is_bounded(node.serial))
{
changed |= raise(taint, node.serial, Reason::UnresolvedOwner);
}
}
changed
}
fn build_decisions(
snapshot: &GraphSnapshot,
ctx: &ExclusionCtx,
taint: &BTreeMap<Serial, Reason>,
sticky_serials: &BTreeSet<Serial>,
) -> Decisions {
let mut candidates = BTreeMap::new();
for node in snapshot.nodes().filter(|n| n.role.is_candidate()) {
let sticky = sticky_serials.contains(&node.serial);
let eligibility = if !ctx.graph_ready {
Eligibility::NotEligible {
reason: Reason::GraphNotReady,
sticky: false,
}
} else if let Some(reason) = taint.get(&node.serial) {
Eligibility::NotEligible {
reason: *reason,
sticky,
}
} else if let Some(reason) = local_exclusion(snapshot, node) {
Eligibility::NotEligible {
reason,
sticky: false,
}
} else {
Eligibility::Eligible
};
candidates.insert(
node.serial,
NodeDecision {
serial: node.serial,
name: node.name.clone(),
eligibility,
},
);
}
Decisions {
candidates,
taint: taint
.iter()
.map(|(serial, reason)| {
(
*serial,
TaintEntry {
reason: *reason,
sticky: sticky_serials.contains(serial),
},
)
})
.collect(),
}
}
/// Node-local reasons a link cannot be created even though the node is
/// clean. These do not propagate — an exclusive-port stream is unlinkable,
/// not hazardous.
fn local_exclusion(snapshot: &GraphSnapshot, node: &NodeSnapshot) -> Option<Reason> {
if node.props.passthrough {
return Some(Reason::Passthrough);
}
if snapshot.ports_of(node.id).any(|port| port.exclusive) {
return Some(Reason::PortExclusive);
}
None
}
/// Sticky bookkeeping for the next recompute: every tainted owner, with
/// every object observed to constitute it, merged with any prior entry that
/// still overlaps. Members accumulate — that is what makes "clear only once
/// all member objects have disappeared" true across churn.
fn build_sticky(
snapshot: &GraphSnapshot,
keys: &owner::OwnerKeyIndex,
components: &OwnerComponents,
taint: &BTreeMap<Serial, Reason>,
prior: &StickyState,
retire_absent: bool,
) -> StickyState {
let mut entries: Vec<StickyOwner> = Vec::new();
// Carry forward prior entries that still have at least one live member.
// An entry with none is gone for good: serials never recycle, so a
// vanished member can never come back — but only a *trustworthy*
// snapshot is allowed to conclude that a member is absent.
for entry in &prior.owners {
if !retire_absent
|| entry
.members
.iter()
.any(|member| is_live(snapshot, *member))
{
entries.push(entry.clone());
}
}
for members in components.components() {
let node_reasons: BTreeMap<Serial, Reason> = members
.iter()
.filter_map(|serial| {
taint
.get(serial)
.filter(|reason| reason.propagates())
.map(|reason| (*serial, *reason))
})
.collect();
if node_reasons.is_empty() {
continue;
}
let mut refs: BTreeSet<ObjectRef> = members.iter().map(|s| ObjectRef::Node(*s)).collect();
refs.extend(
owner::client_serials_of(snapshot, keys, members)
.into_iter()
.map(ObjectRef::Client),
);
let fingerprints = members
.iter()
.flat_map(|serial| keys.fingerprints(*serial))
.collect();
entries.push(StickyOwner {
members: refs,
fingerprints,
node_reasons,
});
}
StickyState {
owners: merge_overlapping(entries),
}
}
fn is_live(snapshot: &GraphSnapshot, member: ObjectRef) -> bool {
match member {
ObjectRef::Node(serial) => snapshot.node(serial).is_some(),
ObjectRef::Client(serial) => snapshot.clients().any(|c| c.serial == serial),
}
}
/// Merge entries that share any member, keeping the strongest reason.
/// Owners fuse over time (a component that gains a leg belonging to a
/// previously separate sticky owner is one owner now); splitting them back
/// apart would drop taint, which is the unsafe direction.
fn merge_overlapping(mut entries: Vec<StickyOwner>) -> Vec<StickyOwner> {
let mut merged: Vec<StickyOwner> = Vec::new();
while let Some(mut entry) = entries.pop() {
let mut absorbed = true;
while absorbed {
absorbed = false;
let mut rest = Vec::with_capacity(entries.len());
for other in entries.drain(..) {
if entry.members.is_disjoint(&other.members) {
rest.push(other);
} else {
for (serial, reason) in other.node_reasons {
entry
.node_reasons
.entry(serial)
.and_modify(|existing| {
if reason.priority() < existing.priority() {
*existing = reason;
}
})
.or_insert(reason);
}
entry.members.extend(other.members);
entry.fingerprints.extend(other.fingerprints);
absorbed = true;
}
}
entries = rest;
}
merged.push(entry);
}
merged.sort_by(|a, b| a.members.iter().next().cmp(&b.members.iter().next()));
merged
}
/// Record `reason` for `serial` if it is new or strictly stronger than what
/// is already recorded. Returns whether anything changed — the fixpoint's
/// termination argument rests on this being monotone.
fn raise(taint: &mut BTreeMap<Serial, Reason>, serial: Serial, reason: Reason) -> bool {
match taint.get(&serial) {
Some(existing) if existing.priority() <= reason.priority() => false,
_ => {
taint.insert(serial, reason);
true
}
}
}
+390
View File
@@ -0,0 +1,390 @@
//! The owner bridge — grouping nodes that belong to the same *owner* even
//! though the graph shows no Link between them.
//!
//! This is the subtlest part of the design (v3.4 §6.1.2). Measured fact it
//! exists to handle: a `module-loopback` forwarder's input leg and output
//! leg have **no Link between them**, so walking Links alone from the
//! leaking output leg finds no inbound links at all — a dead end that reads
//! as "clean". The legs are related only by shared properties.
//!
//! ## The rule
//!
//! A union of keys, strongest first:
//!
//! | # | key | scope |
//! | --- | --- | --- |
//! | 1 | `node.link-group` | per module/filter instance |
//! | 2 | `pulse.module.id` | per pactl module |
//! | 3 | `client.id` | per **connection** |
//! | 4 | `application.process.id` | per process |
//!
//! ⚠️ **"Resolves" means the two legs carry the key AND the values are
//! EQUAL — not "the first key present".** A first-present implementation
//! reproduces the exact measured leak: for `gst-launch pulsesrc ! pulsesink`
//! both legs carry `client.id` (209 and 210) but the values *differ*, so
//! first-present stops at key 3, sees a mismatch, and concludes "different
//! owners". The legs are in fact one process (`application.process.id`
//! 20172 on both). So: try each key in order, and a key resolves only if
//! both legs carry it and the values are equal; otherwise fall through.
//!
//! ## Two exceptions, both guarding against mass over-exclusion
//!
//! 1. **Never bridge on key 4 when the value is pipewire-pulse's own PID**
//! (v3.4 §6.1.2). Module-created streams all carry the daemon's PID, so
//! bridging on it fuses every Pulse module into one owner and a single
//! tainted module input would exclude every module-created stream on the
//! box. Keys 1 and 2 already cover those cases precisely.
//!
//! 2. **Coarse keys (3 and 4) may not bridge nodes exported from a real
//! `Device`** — i.e. nodes carrying `device.id`. ⚠️ This rule is *not*
//! in design v3.4; it was found while implementing, and it is the exact
//! analogue of exception 1 for the session manager.
//! ✅ **MEASURED on the live graph 2026-07-21:**
//!
//! | node | `client.id` | `device.id` | `factory.name` |
//! | --- | --- | --- | --- |
//! | 5 × `alsa_{output,input}.*` | **42** (`WirePlumber [export]`) | 43/45/46 | `api.alsa.pcm.{sink,source}` |
//! | 3 × `sink-sunshine-*` | 83 / 86 / 92 (each its own) | **absent** | `support.null-audio-sink` |
//!
//! So one shared coarse key genuinely does relate every hardware device
//! on the box, and `device.id` cleanly separates that set from virtual
//! sinks. Without the rule, the hardware sink carrying peerspeak's
//! playback (tainted by design, every single recompute) would bridge to
//! *every other device node including the microphone source*, whose
//! readers would then taint their owners' playback legs — reproducing
//! precisely the §6.1.1 catastrophe ("excludes any app using a
//! microphone") through a different door.
//!
//! ⚠️ **Keyed on `device.id`, NOT on `media.class` being `Audio/Sink`.**
//! The first cut suppressed coarse keys for every device-*role* node,
//! and Codex refuted it: a **native virtual sink** — an app that creates
//! an `Audio/Sink` plus a re-emitting stream on one client, with no
//! `link-group` and no `pulse.module.id` — would then have had its only
//! correlation stripped, and it would have leaked the whole call. Such a
//! sink has no `device.id`, so it now bridges on `client.id` as it
//! should.
//!
//! Grouping is **transitive** (union-find). That is the fail-closed
//! direction: bigger owner components mean more taint, never less.
use std::collections::BTreeMap;
use super::snapshot::{GlobalId, GraphSnapshot, NodeSnapshot, Serial};
/// Which key bridged two legs. Ordered strongest first; the `Ord` derive is
/// load-bearing for "report the strongest shared key".
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug, Hash)]
pub enum OwnerKey {
LinkGroup,
PulseModuleId,
ClientId,
ProcessId,
}
impl OwnerKey {
/// Stable, machine-readable — this ends up in the phase 5 audit output
/// and the phase 6 status event.
pub fn code(self) -> &'static str {
match self {
Self::LinkGroup => "node.link-group",
Self::PulseModuleId => "pulse.module.id",
Self::ClientId => "client.id",
Self::ProcessId => "application.process.id",
}
}
}
/// The value a node presents for a given key, if it presents one at all.
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)]
enum KeyValue {
Text(String),
Num(u64),
}
/// Owner keys usable on this node, strongest first.
///
/// A key that is present but unusable (the pipewire-pulse PID; a coarse key
/// on a device node) is **absent** here — that is the whole mechanism of the
/// two exceptions.
fn keys_of(node: &NodeSnapshot, pipewire_pulse_pid: Option<u32>) -> Vec<(OwnerKey, KeyValue)> {
let mut out = Vec::new();
if let Some(group) = &node.props.link_group {
out.push((OwnerKey::LinkGroup, KeyValue::Text(group.clone())));
}
if let Some(module) = node.props.pulse_module_id {
out.push((OwnerKey::PulseModuleId, KeyValue::Num(module)));
}
// Exception 2: coarse keys never bridge passive session-manager device
// nodes — they all share the session manager's client.
if node.props.session_device {
return out;
}
if let Some(client) = node.props.client_id {
out.push((OwnerKey::ClientId, KeyValue::Num(u64::from(client.0))));
}
if let Some(pid) = node.props.process_id {
// Exception 1. Note the fail-closed asymmetry when the daemon PID is
// unknown (`None`): the exception does *not* fire, key 4 applies to
// everything, and Pulse modules fuse into one owner. That is broad
// over-exclusion — annoying and safe — which is the direction v3.4
// §6.1.2's failure-mode paragraph asks for.
if Some(pid) != pipewire_pulse_pid {
out.push((OwnerKey::ProcessId, KeyValue::Num(u64::from(pid))));
}
}
out
}
/// Can this node's owner be positively bounded — i.e. can we enumerate its
/// sibling legs and be right?
///
/// ⚠️ Not the same as "has any usable key", and the difference is a leak.
/// `client.id` alone does **not** bound an owner: that is the measured
/// GStreamer refutation, where one process presented two different
/// `client.id`s for its two legs. So an owner is bounded only by a strong
/// key (link-group / pulse.module.id) or by a *usable* process id — usable
/// meaning key 4 was not suppressed as pipewire-pulse's own PID.
///
/// The case this exists for is v3.4 §12's "module forwarder with neither
/// `link-group` nor `pulse.module.id`": its process id is the daemon's and
/// therefore suppressed, its two legs may carry different `client.id`s, and
/// nothing else relates them. Its sibling output leg cannot be found, so
/// the engine must fail closed rather than declare it clean
/// (v3.4 §6.1.1, final paragraph).
pub fn owner_is_bounded(node: &NodeSnapshot, pipewire_pulse_pid: Option<u32>) -> bool {
keys_of(node, pipewire_pulse_pid)
.iter()
.any(|(key, _)| *key != OwnerKey::ClientId)
}
/// Owner keys computed once per snapshot.
///
/// `keys_of` allocates a `Vec` and clones the `link-group` string, and the
/// bridge asks for keys once per (tainted member × component member) pair —
/// so recomputing was the hot spot in an otherwise linear pass.
#[derive(Debug, Default)]
pub struct OwnerKeyIndex {
keys: BTreeMap<Serial, Vec<(OwnerKey, KeyValue)>>,
}
impl OwnerKeyIndex {
pub fn build(snapshot: &GraphSnapshot, pipewire_pulse_pid: Option<u32>) -> Self {
Self {
keys: snapshot
.nodes()
.map(|node| (node.serial, keys_of(node, pipewire_pulse_pid)))
.collect(),
}
}
/// The strongest key these two nodes share directly, if any.
pub fn strongest_shared(&self, a: Serial, b: Serial) -> Option<OwnerKey> {
let (Some(a_keys), Some(b_keys)) = (self.keys.get(&a), self.keys.get(&b)) else {
return None;
};
// Stored strongest-first, so the first match is the strongest.
a_keys.iter().find_map(|(key, value)| {
b_keys
.iter()
.any(|(other_key, other_value)| other_key == key && other_value == value)
.then_some(*key)
})
}
/// Is `client.id` a usable owner key for this node?
///
/// ⚠️ Load-bearing for sticky state. A device node's `client.id` is
/// suppressed by exception 2, so recording the session manager's Client
/// as a *member* of a tainted device's sticky owner would smuggle the
/// suppressed key back in: the next recompute would expand that Client
/// to every hardware node on the box — the microphone included — and
/// the §6.1.1 catastrophe would arrive one epoch late instead of never.
/// (Codex round 2, finding 1.)
pub fn uses_client_key(&self, serial: Serial) -> bool {
self.keys
.get(&serial)
.is_some_and(|keys| keys.iter().any(|(key, _)| *key == OwnerKey::ClientId))
}
/// The owner keys that are safe to remember *across* connections, for
/// sticky taint: the strong keys plus a usable process id.
///
/// `client.id` is deliberately excluded — it identifies a *connection*,
/// and the whole point of a fingerprint is to survive one process
/// closing a connection and opening another. A live Client member is
/// what covers the same-connection case, precisely.
///
/// These are recyclable strings and numbers, so they are only ever
/// applied while some **serial** member of the owner is still live
/// (v3.4 §6.1.3): while the process is alive, its PID cannot have been
/// handed to anyone else.
pub fn fingerprints(&self, serial: Serial) -> Vec<Fingerprint> {
self.keys
.get(&serial)
.map(|keys| {
keys.iter()
.filter(|(key, _)| *key != OwnerKey::ClientId)
.map(|(key, value)| Fingerprint(*key, value.clone()))
.collect()
})
.unwrap_or_default()
}
/// Does this node currently present `fingerprint`?
pub fn has_fingerprint(&self, serial: Serial, fingerprint: &Fingerprint) -> bool {
self.keys.get(&serial).is_some_and(|keys| {
keys.iter()
.any(|(key, value)| *key == fingerprint.0 && *value == fingerprint.1)
})
}
/// See [`owner_is_bounded`].
pub fn is_bounded(&self, serial: Serial) -> bool {
self.keys
.get(&serial)
.is_some_and(|keys| keys.iter().any(|(key, _)| *key != OwnerKey::ClientId))
}
}
/// The strongest key two nodes share, or `None` if they share none. Used to
/// *name* the key in a bridge decision; membership itself is transitive and
/// comes from [`OwnerComponents`].
pub fn strongest_shared_key(
a: &NodeSnapshot,
b: &NodeSnapshot,
pipewire_pulse_pid: Option<u32>,
) -> Option<OwnerKey> {
let a_keys = keys_of(a, pipewire_pulse_pid);
let b_keys = keys_of(b, pipewire_pulse_pid);
// `keys_of` yields strongest-first, so the first match is the strongest.
a_keys.iter().find_map(|(key, value)| {
b_keys
.iter()
.any(|(other_key, other_value)| other_key == key && other_value == value)
.then_some(*key)
})
}
/// A remembered owner key — see [`OwnerKeyIndex::fingerprints`].
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)]
pub struct Fingerprint(OwnerKey, KeyValue);
/// Nodes partitioned into owner components.
#[derive(Clone, Debug, Default)]
pub struct OwnerComponents {
/// node serial → component index.
of_node: BTreeMap<Serial, usize>,
/// component index → member node serials, ascending.
members: Vec<Vec<Serial>>,
}
impl OwnerComponents {
pub fn build(snapshot: &GraphSnapshot, pipewire_pulse_pid: Option<u32>) -> Self {
let serials: Vec<Serial> = snapshot.nodes().map(|n| n.serial).collect();
let index: BTreeMap<Serial, usize> =
serials.iter().enumerate().map(|(i, s)| (*s, i)).collect();
let mut uf = UnionFind::new(serials.len());
// Group by (key, value) and union within each group. Equivalent to
// the pairwise "some key resolves" rule, and O(n log n).
let mut buckets: BTreeMap<(OwnerKey, KeyValue), Vec<usize>> = BTreeMap::new();
for node in snapshot.nodes() {
let slot = index[&node.serial];
for (key, value) in keys_of(node, pipewire_pulse_pid) {
buckets.entry((key, value)).or_default().push(slot);
}
}
for group in buckets.values() {
for pair in group.windows(2) {
uf.union(pair[0], pair[1]);
}
}
// Compact roots into dense component indices, deterministically.
let mut root_to_component: BTreeMap<usize, usize> = BTreeMap::new();
let mut members: Vec<Vec<Serial>> = Vec::new();
let mut of_node = BTreeMap::new();
for (slot, serial) in serials.iter().enumerate() {
let root = uf.find(slot);
let component = *root_to_component.entry(root).or_insert_with(|| {
members.push(Vec::new());
members.len() - 1
});
members[component].push(*serial);
of_node.insert(*serial, component);
}
Self { of_node, members }
}
pub fn component_of(&self, serial: Serial) -> Option<usize> {
self.of_node.get(&serial).copied()
}
/// Member serials of the component containing `serial`, including it.
/// Empty if the node is not in this snapshot.
pub fn members_with(&self, serial: Serial) -> &[Serial] {
match self.component_of(serial) {
Some(component) => &self.members[component],
None => &[],
}
}
pub fn components(&self) -> impl Iterator<Item = &[Serial]> {
self.members.iter().map(Vec::as_slice)
}
}
struct UnionFind {
parent: Vec<usize>,
}
impl UnionFind {
fn new(len: usize) -> Self {
Self {
parent: (0..len).collect(),
}
}
fn find(&mut self, mut node: usize) -> usize {
while self.parent[node] != node {
self.parent[node] = self.parent[self.parent[node]];
node = self.parent[node];
}
node
}
fn union(&mut self, a: usize, b: usize) {
let (a, b) = (self.find(a), self.find(b));
if a != b {
// Lowest root wins, so components are deterministic.
let (low, high) = if a < b { (a, b) } else { (b, a) };
self.parent[high] = low;
}
}
}
/// Client objects belonging to an owner component, so sticky taint can be
/// keyed on every object that constitutes the owner (v3.4 §6.1.3: clear the
/// entry only once **all** member objects are gone).
pub fn client_serials_of(
snapshot: &GraphSnapshot,
keys: &OwnerKeyIndex,
nodes: &[Serial],
) -> Vec<Serial> {
let mut out: Vec<Serial> = nodes
.iter()
// Only nodes for which `client.id` is a *usable* owner key. See
// `uses_client_key`: recording a device node's shared session-manager
// Client here would defeat exception 2 on the next recompute.
.filter(|serial| keys.uses_client_key(**serial))
.filter_map(|serial| snapshot.node(*serial))
.filter_map(|node| node.props.client_id)
// An ambiguous client id means two Clients claim it and we cannot
// say which one is ours, so remember both: an entry that recorded
// neither could be retired while its owner was still live.
.flat_map(|id: GlobalId| snapshot.clients_with_id(id).map(|client| client.serial))
.collect();
out.sort_unstable();
out.dedup();
out
}
+332
View File
@@ -0,0 +1,332 @@
//! The plain, owned graph model the taint engine reasons over.
//!
//! **No PipeWire types appear in this file, by design** (impl plan §4,
//! phase 2). The registry observer (phase 3) translates live globals into
//! these structs; every test builds them by hand. Nothing here ever links
//! against libpipewire.
//!
//! Two id-ish things live in this model and confusing them is the bug the
//! whole file is shaped to prevent:
//!
//! - [`Serial`] — `object.serial`, 64-bit, monotonic, **never reused**.
//! This is *identity*. Sticky taint is keyed on it.
//! - [`GlobalId`] — the PipeWire global id, 32-bit and **recycled**. It is
//! a *lookup key within one snapshot* and nothing else: links name their
//! endpoints with it, nodes name their client with it. It must never
//! outlive the snapshot it was read from (design v3.4 §6.1.3).
use std::collections::BTreeMap;
/// `object.serial` — 64-bit, monotonic, never recycled. Identity.
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug, Hash)]
pub struct Serial(pub u64);
/// A PipeWire global id — 32-bit and **recycled**. Snapshot-local lookup
/// key only; see the module docs.
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug, Hash)]
pub struct GlobalId(pub u32);
/// What a node does with audio, parsed from `media.class`.
///
/// Taint is computed at **node** granularity (v3.4 §6.1 edge type 2: the
/// monitor connection is already a real Link whose output node is the sink
/// itself, so a node-level walk crosses `app → sink → monitor-reader` for
/// free). Ports exist in the model for link creation in phase 6 and for the
/// `port.exclusive` predicate, not for taint.
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug, Hash)]
pub enum MediaRole {
/// `Stream/Output/Audio` — an application playing audio. The only
/// fan-out candidate.
StreamOutput,
/// `Stream/Input/Audio` — an application capturing audio.
StreamInput,
/// `Audio/Sink` — a real or virtual sink.
Sink,
/// `Audio/Source` — a real or virtual source.
Source,
/// `Audio/Duplex`. ⚠️ Node granularity smears taint across both roles
/// of these; accepted for v1 as fail-closed over-exclusion
/// (v3.4 §6.1, edge type 2 caveat).
Duplex,
/// Anything else, including video and unparseable/absent `media.class`.
Other,
}
impl MediaRole {
pub fn parse(media_class: Option<&str>) -> Self {
match media_class {
Some("Stream/Output/Audio") => Self::StreamOutput,
Some("Stream/Input/Audio") => Self::StreamInput,
Some("Audio/Sink") => Self::Sink,
Some("Audio/Source") => Self::Source,
Some("Audio/Duplex") => Self::Duplex,
_ => Self::Other,
}
}
/// Can this node *receive* audio? This is the gate on the owner bridge:
/// taint crosses the intra-process hop only when the owner is actually
/// reading tainted audio (v3.4 §6.1.1 — "this client has both an input
/// and an output leg ⇒ exclude the output" is the catastrophic rule
/// that excludes every app with a microphone).
///
/// `Sink` counts: EasyEffects' `ee_sink` is an `Audio/Sink` that
/// receives the tainted mix, and its re-emitting leg is joined to it by
/// `node.link-group` with no Link between them.
pub fn receives_audio(self) -> bool {
matches!(self, Self::StreamInput | Self::Sink | Self::Duplex)
}
/// Device-ish nodes — everything that is not a `Stream/*`. Coarse owner
/// keys are not allowed to bridge these; see [`super::owner`].
pub fn is_device_role(self) -> bool {
matches!(self, Self::Sink | Self::Source | Self::Duplex)
}
/// Only `Stream/Output/Audio` nodes are fan-out candidates (v3.4 §6.2).
pub fn is_candidate(self) -> bool {
matches!(self, Self::StreamOutput)
}
}
/// The subset of node properties the engine actually reasons about.
///
/// Deliberately a struct of parsed fields rather than a property bag: the
/// parsing (and its failure modes) belongs at the observer boundary, and a
/// bag invites `props.get("...")` typos that silently read `None` — which
/// on this feature means "not tainted".
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct NodeProps {
/// `peerspeak.owned` is present and truthy (v3.4 §5.1). A correctness
/// mechanism, explicitly *not* a security boundary.
pub peerspeak_owned: bool,
/// `pulse.module.id`, parsed as `u64` — never `u32`, per v3.4 §5.2's
/// parse-defensively note and the phase 0a truncation bug.
pub pulse_module_id: Option<u64>,
/// `node.link-group` — owner key 1, and the `echo-cancel-` hazard
/// prefix (v3.4 §5.4 / D3).
pub link_group: Option<String>,
/// `client.id` — owner key 3. A **connection**, not an owner: GStreamer
/// opens one per stream (v3.4 §6.1.2, measured refutation).
pub client_id: Option<GlobalId>,
/// `application.process.id` **on the node** — owner key 4. For
/// module-created streams this is pipewire-pulse's own PID, which is
/// why [`super::ExclusionCtx::pipewire_pulse_pid`] exists.
pub process_id: Option<u32>,
/// The stream negotiated an encoded/passthrough format; a second link
/// would refuse or corrupt it (v3.4 §6.2).
pub passthrough: bool,
/// This node is a **passive device node exported by the session
/// manager** — a real sound card's sink or source, not something that
/// forwards audio.
///
/// ⚠️ **A positive high-confidence classification the observer owes, not
/// a raw property** (Codex rounds 23). PipeWire defines `device.id`
/// only as "the Device this node belongs to" and `device.api` as that
/// Device's access API; **neither promises the node passively terminates
/// audio**, so a card-associated filter can satisfy both. Setting this
/// flag *removes* two protections at once — the node's coarse owner keys
/// (`owner` exception 2) and its ability to trip the fail-closed
/// backstop — so a false positive is a leak, not over-exclusion.
///
/// **Phase-3 contract:**
/// - Set `true` only on positively-identified passive hardware
/// terminals: a resolved `device.id` on a real backend
/// (`device.api` present) whose `factory.name` is on an **explicit
/// hardware-PCM allowlist** — `api.alsa.pcm.sink`, `api.alsa.pcm.source`,
/// and the equivalent for other real backends (bluez5, v4l2 for the
/// media case) as phase 3 enumerates them — never a filter, loopback,
/// or `support.null-audio-sink` factory. An allowlist, not a
/// substring or a denylist: an unknown factory is not a device.
/// Measured discriminator on the
/// target box: the five ALSA nodes carry `device.api=alsa` +
/// `factory.name=api.alsa.pcm.*` and share `client.id=42`
/// (`WirePlumber [export]`); the three `support.null-audio-sink` nodes
/// carry neither. (`node.physical` was measured **null** on the ALSA
/// nodes here, so it is *not* a usable discriminator — do not rely on
/// it.)
/// - **Fail closed: unknown ⇒ `false`.** A node that cannot be
/// positively classified keeps its owner keys and can trip the
/// backstop; both are the safe direction.
/// - A node MUST NOT enter a snapshot with this field provisional. If
/// the Device backing a node has not yet been bound, withhold the node
/// and keep the epoch not-ready — otherwise a provisional `false`
/// during not-ready fuses sink and mic on the shared session client
/// and that fusion can persist as sticky over-exclusion (round-3
/// finding 3).
///
/// ⚠️ **A false positive is leak-capable — do not treat it as braced.**
/// I claimed a mis-classified filter could not leak because its legs
/// share a `node.link-group` (strong-key bridge) or trip the unbounded
/// backstop. Codex refuted it (round 4): a filter *without* a shared
/// strong key, marked `session_device=true`, cannot activate the
/// backstop from its reading leg, so a differently-keyed re-emitting leg
/// leaks. Those braces catch *some* shapes, not all. The only real
/// defence is a correct classifier — hence "positive high-confidence"
/// and "fail closed to false" above, without exception.
///
/// What it is for: every real device node shares the session manager's
/// `client.id`, so coarse owner keys must not bridge them — else
/// peerspeak's playback (which taints the default sink every recompute)
/// would reach the microphone. See [`super::owner`] exception 2.
pub session_device: bool,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct NodeSnapshot {
pub serial: Serial,
pub id: GlobalId,
/// `node.name`, for diagnostics and for `pixelpass_capture_*` ancestry
/// detection (v3.4 §6.2, cycle prevention).
pub name: Option<String>,
pub role: MediaRole,
pub props: NodeProps,
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum PortDirection {
In,
Out,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PortSnapshot {
pub serial: Serial,
pub id: GlobalId,
/// Owning node, by snapshot-local id.
pub node: GlobalId,
pub direction: PortDirection,
/// `port.exclusive` — fan-out will be refused (v3.4 §6.2).
pub exclusive: bool,
/// `port.monitor`. Recorded for phase 6 link creation; taint does not
/// need it at node granularity.
pub monitor: bool,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LinkSnapshot {
pub serial: Serial,
pub id: GlobalId,
/// `link.output.node` — the node audio flows **from**.
pub output_node: GlobalId,
/// `link.input.node` — the node audio flows **to**.
pub input_node: GlobalId,
pub output_port: Option<GlobalId>,
pub input_port: Option<GlobalId>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ClientSnapshot {
pub serial: Serial,
pub id: GlobalId,
/// `pipewire.sec.pid` — for Pulse-emulated clients this is
/// **pipewire-pulse's** PID, identical across every unrelated app
/// (v3.4 §5.2 correction 5). Phase 3 derives the daemon PID from the
/// consistency of this value; the engine only consumes the result.
pub sec_pid: Option<u32>,
}
/// How a snapshot-local id resolves.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum IdLookup {
Unique(Serial),
/// Two live objects in one snapshot claim the same global id — the
/// observer missed a removal, so the recycled id is ambiguous. Every
/// edge touching it is treated as unresolved, i.e. fail closed.
Ambiguous,
}
/// One coherent observation of the graph.
///
/// Built through [`GraphSnapshot::new`] so the id indexes and the ambiguity
/// detection cannot be skipped.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct GraphSnapshot {
nodes: BTreeMap<Serial, NodeSnapshot>,
ports: BTreeMap<Serial, PortSnapshot>,
links: BTreeMap<Serial, LinkSnapshot>,
clients: BTreeMap<Serial, ClientSnapshot>,
node_ids: BTreeMap<GlobalId, IdLookup>,
client_ids: BTreeMap<GlobalId, IdLookup>,
}
impl GraphSnapshot {
pub fn new(
nodes: Vec<NodeSnapshot>,
ports: Vec<PortSnapshot>,
links: Vec<LinkSnapshot>,
clients: Vec<ClientSnapshot>,
) -> Self {
let node_ids = index_ids(nodes.iter().map(|n| (n.id, n.serial)));
let client_ids = index_ids(clients.iter().map(|c| (c.id, c.serial)));
Self {
nodes: nodes.into_iter().map(|n| (n.serial, n)).collect(),
ports: ports.into_iter().map(|p| (p.serial, p)).collect(),
links: links.into_iter().map(|l| (l.serial, l)).collect(),
clients: clients.into_iter().map(|c| (c.serial, c)).collect(),
node_ids,
client_ids,
}
}
pub fn nodes(&self) -> impl Iterator<Item = &NodeSnapshot> {
self.nodes.values()
}
pub fn node(&self, serial: Serial) -> Option<&NodeSnapshot> {
self.nodes.get(&serial)
}
pub fn links(&self) -> impl Iterator<Item = &LinkSnapshot> {
self.links.values()
}
pub fn ports(&self) -> impl Iterator<Item = &PortSnapshot> {
self.ports.values()
}
pub fn clients(&self) -> impl Iterator<Item = &ClientSnapshot> {
self.clients.values()
}
/// Resolve a snapshot-local node id. `None` means "no such node in this
/// snapshot", which for a link endpoint means unresolved ancestry.
pub fn node_by_id(&self, id: GlobalId) -> Option<IdLookup> {
self.node_ids.get(&id).copied()
}
pub fn client_by_id(&self, id: GlobalId) -> Option<IdLookup> {
self.client_ids.get(&id).copied()
}
/// Every node claiming a global id. More than one means the id is
/// [`IdLookup::Ambiguous`] and each claimant must be treated as a
/// possible endpoint of any link naming it.
pub fn nodes_with_id(&self, id: GlobalId) -> impl Iterator<Item = &NodeSnapshot> {
self.nodes.values().filter(move |node| node.id == id)
}
/// Every client claiming a global id — same fail-closed reasoning.
pub fn clients_with_id(&self, id: GlobalId) -> impl Iterator<Item = &ClientSnapshot> {
self.clients.values().filter(move |client| client.id == id)
}
/// Ports belonging to a node, by the node's snapshot-local id.
pub fn ports_of(&self, node: GlobalId) -> impl Iterator<Item = &PortSnapshot> {
self.ports.values().filter(move |p| p.node == node)
}
}
fn index_ids(entries: impl Iterator<Item = (GlobalId, Serial)>) -> BTreeMap<GlobalId, IdLookup> {
let mut out: BTreeMap<GlobalId, IdLookup> = BTreeMap::new();
for (id, serial) in entries {
out.entry(id)
.and_modify(|slot| {
if *slot != IdLookup::Unique(serial) {
*slot = IdLookup::Ambiguous;
}
})
.or_insert(IdLookup::Unique(serial));
}
out
}
File diff suppressed because it is too large Load Diff