diff --git a/src/host/mod.rs b/src/host/mod.rs index bde0594..ea87de5 100644 --- a/src/host/mod.rs +++ b/src/host/mod.rs @@ -3,6 +3,7 @@ mod capture; mod pipeline; mod quality; mod serve; +pub mod taint; mod wayland; mod x11; diff --git a/src/host/taint/fixture.rs b/src/host/taint/fixture.rs new file mode 100644 index 0000000..5a5273d --- /dev/null +++ b/src/host/taint/fixture.rs @@ -0,0 +1,282 @@ +//! 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, + ports: Vec, + links: Vec, + clients: Vec, + /// 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, + client_by_module: BTreeMap, + session_client: Option, +} + +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) -> 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)) + } + + /// 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 = 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 + } + }; + 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, and + /// WirePlumber's client and PID — shared with every other device. + 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)) + } + + 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 = dropped.iter().map(|n| n.serial).collect(); + let nodes: Vec = 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 = 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. + pub fn drop_clients(&mut self, ids: &[GlobalId]) { + self.clients.retain(|c| !ids.contains(&c.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. +pub fn device(session_client: GlobalId, session_pid: u32) -> NodeProps { + NodeProps { + client_id: Some(session_client), + process_id: Some(session_pid), + ..NodeProps::default() + } +} + +pub fn peerspeak_owned(client: GlobalId, pid: u32) -> NodeProps { + NodeProps { + peerspeak_owned: true, + ..app(client, pid) + } +} diff --git a/src/host/taint/mod.rs b/src/host/taint/mod.rs new file mode 100644 index 0000000..3719bdd --- /dev/null +++ b/src/host/taint/mod.rs @@ -0,0 +1,745 @@ +//! 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. + +// 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 }, + /// 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, + /// pipewire-pulse's own PID, derived by the observer (phase 3) from a + /// consistent `pipewire.sec.pid` across Pulse clients validated against + /// `/proc//comm`. `None` is safe but coarse — see [`owner`]. + pub pipewire_pulse_pid: Option, + /// Serials of objects pixelpass itself created this run. + pub pixelpass_owned: BTreeSet, + /// 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, + /// 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, +} + +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, +} + +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, + pub eligibility: Eligibility, +} + +impl NodeDecision { + pub fn is_eligible(&self) -> bool { + matches!(self.eligibility, Eligibility::Eligible) + } + + pub fn reason(&self) -> Option { + 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, + /// Taint over *all* node roles, for diagnostics and for the phase 5 + /// audit output. + pub taint: BTreeMap, +} + +impl Decisions { + /// Serials of eligible candidates, ascending. + pub fn eligible(&self) -> Vec { + 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 (O(V+E) over a ~40-node +/// desktop graph); there is deliberately no incremental dirty-set. +pub fn evaluate( + snapshot: &GraphSnapshot, + ctx: &ExclusionCtx, + prior: &StickyState, +) -> (Decisions, StickyState) { + let components = OwnerComponents::build(snapshot, ctx.pipewire_pulse_pid); + + let mut taint: BTreeMap = BTreeMap::new(); + let mut sticky_serials: BTreeSet = BTreeSet::new(); + + seed_local_roots(snapshot, ctx, &mut taint); + seed_sticky( + snapshot, + 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 downstream = downstream_edges(snapshot, &mut taint); + loop { + let mut changed = false; + changed |= propagate_links(&downstream, &mut taint); + changed |= propagate_owner_bridge(snapshot, ctx, &components, &mut taint); + changed |= propagate_unresolved_owner(snapshot, ctx, &mut taint); + if !changed { + break; + } + } + + let decisions = build_decisions(snapshot, ctx, &taint, &sticky_serials); + let next_sticky = build_sticky(snapshot, &components, &taint, prior); + (decisions, next_sticky) +} + +/// Roots that are visible on the node itself. +fn seed_local_roots( + snapshot: &GraphSnapshot, + ctx: &ExclusionCtx, + taint: &mut BTreeMap, +) { + 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 { + 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). +fn seed_sticky( + snapshot: &GraphSnapshot, + prior: &StickyState, + components: &OwnerComponents, + taint: &mut BTreeMap, + sticky_serials: &mut BTreeSet, +) { + for entry in &prior.owners { + let live_nodes: Vec = entry + .members + .iter() + .filter_map(|member| match member { + ObjectRef::Node(serial) => snapshot.node(*serial).map(|_| *serial), + ObjectRef::Client(_) => None, + }) + .collect(); + // 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); + } + } + } + } +} + +/// `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, +) -> BTreeMap> { + let mut edges: BTreeMap> = BTreeMap::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); + } + (_, Some(IdLookup::Unique(to))) => { + // Something feeds this node and we cannot say what. + raise(taint, to, Reason::UnresolvedAncestry); + } + _ => {} + } + } + for targets in edges.values_mut() { + targets.sort_unstable(); + targets.dedup(); + } + edges +} + +fn propagate_links( + downstream: &BTreeMap>, + taint: &mut BTreeMap, +) -> bool { + let mut changed = false; + let mut queue: VecDeque = 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( + snapshot: &GraphSnapshot, + ctx: &ExclusionCtx, + components: &OwnerComponents, + taint: &mut BTreeMap, +) -> bool { + let mut changed = false; + for members in components.components() { + let sources: Vec = members + .iter() + .copied() + .filter(|serial| { + taint.get(serial).is_some_and(|r| r.propagates()) + && snapshot + .node(*serial) + .is_some_and(|n| n.role.receives_audio()) + }) + .collect(); + if sources.is_empty() { + continue; + } + for target in members { + if sources.contains(target) { + continue; + } + let Some(target_node) = snapshot.node(*target) else { + 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| snapshot.node(*source)) + .filter_map(|source| { + owner::strongest_shared_key(source, target_node, ctx.pipewire_pulse_pid) + }) + .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): a capture stream is reading tainted audio and nothing about +/// it lets us enumerate its sibling output legs, so we cannot know which +/// output leg is re-emitting what it read. Exclude the output legs that are +/// equally unbounded. +/// +/// Two deliberate restrictions keep the blast radius at "small", as the +/// design promises: +/// +/// - **Source must be a `Stream/Input/Audio`.** A tainted *device* sink is +/// the normal case, not an anomaly — peerspeak's own playback taints the +/// default sink on literally every recompute — and letting that trip this +/// rule would exclude the entire desktop. +/// - **Targets must themselves be unbounded.** Ordinary application streams +/// carry a real `application.process.id` and are bounded, so they are +/// never swept up; in practice only other daemon-owned keyless module +/// streams are. +fn propagate_unresolved_owner( + snapshot: &GraphSnapshot, + ctx: &ExclusionCtx, + taint: &mut BTreeMap, +) -> bool { + let unbounded_reader = snapshot.nodes().any(|node| { + node.role == MediaRole::StreamInput + && taint.get(&node.serial).is_some_and(|r| r.propagates()) + && !owner::owner_is_bounded(node, ctx.pipewire_pulse_pid) + }); + if !unbounded_reader { + return false; + } + let mut changed = false; + for node in snapshot.nodes() { + if node.role == MediaRole::StreamOutput + && !owner::owner_is_bounded(node, ctx.pipewire_pulse_pid) + { + changed |= raise(taint, node.serial, Reason::UnresolvedOwner); + } + } + changed +} + +fn build_decisions( + snapshot: &GraphSnapshot, + ctx: &ExclusionCtx, + taint: &BTreeMap, + sticky_serials: &BTreeSet, +) -> 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 { + 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, + components: &OwnerComponents, + taint: &BTreeMap, + prior: &StickyState, +) -> StickyState { + let mut entries: Vec = 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. + for entry in &prior.owners { + if entry + .members + .iter() + .any(|member| is_live(snapshot, *member)) + { + entries.push(entry.clone()); + } + } + + for members in components.components() { + let node_reasons: BTreeMap = 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 = members.iter().map(|s| ObjectRef::Node(*s)).collect(); + refs.extend( + owner::client_serials_of(snapshot, members) + .into_iter() + .map(ObjectRef::Client), + ); + entries.push(StickyOwner { + members: refs, + 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) -> Vec { + let mut merged: Vec = 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); + 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: Serial, reason: Reason) -> bool { + match taint.get(&serial) { + Some(existing) if existing.priority() <= reason.priority() => false, + _ => { + taint.insert(serial, reason); + true + } + } +} diff --git a/src/host/taint/owner.rs b/src/host/taint/owner.rs new file mode 100644 index 0000000..0105432 --- /dev/null +++ b/src/host/taint/owner.rs @@ -0,0 +1,272 @@ +//! 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 device-role nodes.** ⚠️ 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: +//! every ALSA `Audio/Sink` and `Audio/Source` on the box is created by +//! WirePlumber and therefore shares one `client.id` and one +//! `application.process.id`. Without this rule, the hardware sink +//! carrying peerspeak's playback (tainted by design, every single time) +//! 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. Device nodes that genuinely +//! belong to a module still bridge, via keys 1 and 2. +//! +//! 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) -> 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 device-role nodes. + if node.role.is_device_role() { + 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) -> bool { + keys_of(node, pipewire_pulse_pid) + .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, +) -> Option { + 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) + }) +} + +/// Nodes partitioned into owner components. +#[derive(Clone, Debug, Default)] +pub struct OwnerComponents { + /// node serial → component index. + of_node: BTreeMap, + /// component index → member node serials, ascending. + members: Vec>, +} + +impl OwnerComponents { + pub fn build(snapshot: &GraphSnapshot, pipewire_pulse_pid: Option) -> Self { + let serials: Vec = snapshot.nodes().map(|n| n.serial).collect(); + let index: BTreeMap = + 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> = 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 = BTreeMap::new(); + let mut members: Vec> = 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 { + 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 { + self.members.iter().map(Vec::as_slice) + } +} + +struct UnionFind { + parent: Vec, +} + +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, nodes: &[Serial]) -> Vec { + let mut out: Vec = nodes + .iter() + .filter_map(|serial| snapshot.node(*serial)) + .filter_map(|node| node.props.client_id) + .filter_map(|id: GlobalId| match snapshot.client_by_id(id) { + Some(super::snapshot::IdLookup::Unique(serial)) => Some(serial), + _ => None, + }) + .collect(); + out.sort_unstable(); + out.dedup(); + out +} diff --git a/src/host/taint/snapshot.rs b/src/host/taint/snapshot.rs new file mode 100644 index 0000000..ba4a3f4 --- /dev/null +++ b/src/host/taint/snapshot.rs @@ -0,0 +1,266 @@ +//! 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, + /// `node.link-group` — owner key 1, and the `echo-cancel-` hazard + /// prefix (v3.4 §5.4 / D3). + pub link_group: Option, + /// `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, + /// `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, + /// The stream negotiated an encoded/passthrough format; a second link + /// would refuse or corrupt it (v3.4 §6.2). + pub passthrough: 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, + 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, + pub input_port: Option, +} + +#[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, +} + +/// 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, + ports: BTreeMap, + links: BTreeMap, + clients: BTreeMap, + node_ids: BTreeMap, + client_ids: BTreeMap, +} + +impl GraphSnapshot { + pub fn new( + nodes: Vec, + ports: Vec, + links: Vec, + clients: Vec, + ) -> 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 { + self.nodes.values() + } + + pub fn node(&self, serial: Serial) -> Option<&NodeSnapshot> { + self.nodes.get(&serial) + } + + pub fn links(&self) -> impl Iterator { + self.links.values() + } + + pub fn ports(&self) -> impl Iterator { + self.ports.values() + } + + pub fn clients(&self) -> impl Iterator { + 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 { + self.node_ids.get(&id).copied() + } + + pub fn client_by_id(&self, id: GlobalId) -> Option { + self.client_ids.get(&id).copied() + } + + /// Ports belonging to a node, by the node's snapshot-local id. + pub fn ports_of(&self, node: GlobalId) -> impl Iterator { + self.ports.values().filter(move |p| p.node == node) + } +} + +fn index_ids(entries: impl Iterator) -> BTreeMap { + let mut out: BTreeMap = 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 +} diff --git a/src/host/taint/tests.rs b/src/host/taint/tests.rs new file mode 100644 index 0000000..1e71d61 --- /dev/null +++ b/src/host/taint/tests.rs @@ -0,0 +1,926 @@ +//! The taint-engine fixture matrix — design v3.4 §12 plus the impl plan's +//! degenerate-snapshot addition. +//! +//! **Every test asserts an exact partition of the whole candidate universe**, +//! never a spot check on one named node. Checking only the node a test cares +//! about constrains nothing about the others, and an exclude-everything +//! implementation would pass. So each test names *every* `Stream/Output/Audio` +//! node in its graph as either eligible or excluded-with-a-reason-code. +//! +//! Fixture realism that is load-bearing: an ordinary app holds **one client +//! connection**, each pactl module holds its own, and every device node +//! shares the session manager's. Getting that wrong makes the engine look +//! broken (or, worse, makes a leak look fine). + +use std::collections::BTreeSet; + +use super::fixture::{Graph, NodeRef, PULSE_PID, app}; +use super::owner::{OwnerKey, strongest_shared_key}; +use super::snapshot::{MediaRole, NodeProps, PortDirection, Serial}; +use super::{Decisions, Eligibility, ExclusionCtx, ObjectRef, Reason, StickyState, evaluate}; + +fn ctx() -> ExclusionCtx { + ExclusionCtx { + aec_module_id: None, + pipewire_pulse_pid: Some(PULSE_PID), + pixelpass_owned: BTreeSet::new(), + graph_ready: true, + } +} + +fn run(graph: &Graph, ctx: &ExclusionCtx) -> Decisions { + evaluate(&graph.build(), ctx, &StickyState::default()).0 +} + +/// Assert the complete candidate partition. `excluded` names reason codes. +#[track_caller] +fn assert_partition( + decisions: &Decisions, + eligible: &[(&str, NodeRef)], + excluded: &[(&str, NodeRef, &str)], +) { + let mut expected_eligible: Vec = eligible.iter().map(|(_, n)| n.serial).collect(); + expected_eligible.sort_unstable(); + let mut expected_excluded: Vec<(Serial, &str)> = excluded + .iter() + .map(|(_, n, code)| (n.serial, *code)) + .collect(); + expected_excluded.sort_unstable(); + + let label = |serial: Serial| -> String { + eligible + .iter() + .map(|(name, n)| (*name, *n)) + .chain(excluded.iter().map(|(name, n, _)| (*name, *n))) + .find(|(_, n)| n.serial == serial) + .map_or_else(|| format!(""), |(name, _)| name.into()) + }; + let actual: Vec = decisions + .candidates + .values() + .map(|d| match d.reason() { + None => format!("{} = eligible", label(d.serial)), + Some(reason) => format!("{} = {}", label(d.serial), reason.code()), + }) + .collect(); + + assert_eq!( + (decisions.eligible(), decisions.excluded()), + (expected_eligible, expected_excluded), + "actual decisions: {actual:#?}" + ); +} + +#[track_caller] +fn assert_tainted(decisions: &Decisions, node: NodeRef, reason: &str) { + let entry = decisions + .taint + .get(&node.serial) + .unwrap_or_else(|| panic!("expected {:?} to be tainted", node.serial)); + assert_eq!(entry.reason.code(), reason); +} + +#[track_caller] +fn assert_untainted(decisions: &Decisions, node: NodeRef) { + assert_eq!( + decisions.taint.get(&node.serial).map(|e| e.reason.code()), + None, + "expected {:?} to be untainted", + node.serial + ); +} + +// ────────────────────────────────────────────────────────────────────── +// Boundary / degenerate cases — the fail-closed default +// ────────────────────────────────────────────────────────────────────── + +#[test] +fn empty_snapshot_yields_nothing_eligible() { + let decisions = run(&Graph::new(), &ctx()); + assert!(decisions.candidates.is_empty()); + assert!(decisions.eligible().is_empty()); +} + +#[test] +fn a_graph_with_only_untainted_streams_still_partitions() { + // The other half of the degenerate case: "nothing eligible" must not be + // reached by excluding everything unconditionally. + let mut graph = Graph::new(); + let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114); + assert_partition(&run(&graph, &ctx()), &[("firefox", firefox)], &[]); +} + +#[test] +fn graph_not_ready_excludes_every_candidate() { + let mut graph = Graph::new(); + let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114); + let decisions = run( + &graph, + &ExclusionCtx { + graph_ready: false, + ..ctx() + }, + ); + assert_partition(&decisions, &[], &[("firefox", firefox, "graph-not-ready")]); +} + +#[test] +fn decisions_do_not_depend_on_insertion_order() { + // The reported reason must come from an explicit priority, not from + // traversal order, or the audit output is unstable between recomputes. + let build = |reverse: bool| { + let mut graph = Graph::new(); + let sink = graph.device_node("sink", MediaRole::Sink); + let call = graph.peerspeak_node("peerspeak", 7); + let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114); + if reverse { + graph.link(firefox, sink); + graph.link(call, sink); + } else { + graph.link(call, sink); + graph.link(firefox, sink); + } + run(&graph, &ctx()).excluded() + }; + assert_eq!(build(false), build(true)); +} + +// ────────────────────────────────────────────────────────────────────── +// Node-local roots (v3.4 §5.1, §5.2, §6.2, §6.6) +// ────────────────────────────────────────────────────────────────────── + +#[test] +fn peerspeak_tagged_nodes_are_excluded_and_plain_apps_are_not() { + let mut graph = Graph::new(); + let sink = graph.device_node("hw-sink", MediaRole::Sink); + let call = graph.peerspeak_node("peerspeak-call", 7); + let mpv = graph.peerspeak_node("peerspeak-mpv", 8); + let notify = graph.peerspeak_node("peerspeak-notify", 9); + let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114); + for node in [call, mpv, notify, firefox] { + graph.link(node, sink); + } + + let decisions = run(&graph, &ctx()); + assert_partition( + &decisions, + &[("firefox", firefox)], + &[ + ("call", call, "peerspeak-owned"), + ("mpv", mpv, "peerspeak-owned"), + ("notify", notify, "peerspeak-owned"), + ], + ); + // The sink carried peerspeak audio, so it is tainted — that is what + // makes every monitor reader downstream of it tainted too. + assert_tainted(&decisions, sink, "tainted-upstream"); +} + +#[test] +fn aec_identity_is_exact_equality_and_other_modules_stay_eligible() { + let mut graph = Graph::new(); + let ours = graph.module_node("echo-cancel-playback", MediaRole::StreamOutput, 536_870_919); + // A tunnel/RTP module: it has a `pulse.module.id`, just not ours. + // "Has any pulse.module.id" is explicitly rejected as an exclusion rule + // — such a module may be the only carrier of audio the user wants + // shared (v3.4 §5.2 correction 2). + let tunnel = graph.module_node("tunnel-out", MediaRole::StreamOutput, 536_870_921); + let plain = graph.app_node("firefox", MediaRole::StreamOutput, 11114); + + let decisions = run( + &graph, + &ExclusionCtx { + aec_module_id: Some(536_870_919), + ..ctx() + }, + ); + assert_partition( + &decisions, + &[("tunnel", tunnel), ("plain", plain)], + &[("ours", ours, "aec-identity")], + ); +} + +#[test] +fn aec_module_id_is_compared_beyond_u32() { + // Guards the phase 0a widening at the engine boundary too: these two + // differ only above bit 32. + let big = u64::from(u32::MAX) + 7; + let mut graph = Graph::new(); + let ours = graph.module_node("aec", MediaRole::StreamOutput, big); + let other = graph.module_node("other", MediaRole::StreamOutput, big ^ 0x1_0000_0000); + let decisions = run( + &graph, + &ExclusionCtx { + aec_module_id: Some(big), + ..ctx() + }, + ); + assert_partition( + &decisions, + &[("other", other)], + &[("ours", ours, "aec-identity")], + ); +} + +#[test] +fn foreign_echo_cancel_group_is_excluded_not_shared() { + // Decision D3: warn and exclude. The group prefix is hazard detection, + // never ownership — it cannot tell peerspeak's AEC from anyone else's. + let mut graph = Graph::new(); + let foreign = graph.group_node( + "echo-cancel-playback", + MediaRole::StreamOutput, + "echo-cancel-9999-13", + 4321, + ); + let plain = graph.app_node("firefox", MediaRole::StreamOutput, 11114); + assert_partition( + &run(&graph, &ctx()), + &[("plain", plain)], + &[("foreign", foreign, "foreign-echo-cancel")], + ); +} + +#[test] +fn pixelpass_capture_sink_and_its_downstream_are_excluded() { + // v3.4 §6.2 cycle prevention, and impl-plan §5.1 row 7: observing the + // capture sink is not enough — a *downstream* candidate must be named, + // or recognising `pixelpass_capture_*` as a mere sink name would pass + // without any transitive propagation. + let mut graph = Graph::new(); + let other_host_sink = graph.module_node("pixelpass_capture_31337", MediaRole::Sink, 42); + let fwd_in = graph.module_node("fwd-in", MediaRole::StreamInput, 77); + let fwd_out = graph.module_node("fwd-out", MediaRole::StreamOutput, 77); + graph.link(other_host_sink, fwd_in); + let plain = graph.app_node("firefox", MediaRole::StreamOutput, 11114); + + let decisions = run(&graph, &ctx()); + assert_partition( + &decisions, + &[("plain", plain)], + &[("fwd-out", fwd_out, "tainted-owner-bridge")], + ); + assert_tainted(&decisions, other_host_sink, "pixelpass-owned"); + assert_tainted(&decisions, fwd_in, "tainted-upstream"); +} + +#[test] +fn pixelpass_owned_serials_are_roots_even_without_a_capture_sink_name() { + let mut graph = Graph::new(); + let ours = graph.device_node("some-sink", MediaRole::Sink); + let reader = graph.app_node("reader-in", MediaRole::StreamInput, 5150); + let leg = graph.app_node("reader-out", MediaRole::StreamOutput, 5150); + graph.link(ours, reader); + let decisions = run( + &graph, + &ExclusionCtx { + pixelpass_owned: BTreeSet::from([ours.serial]), + ..ctx() + }, + ); + assert_partition(&decisions, &[], &[("leg", leg, "tainted-owner-bridge")]); + assert_tainted(&decisions, ours, "pixelpass-owned"); +} + +#[test] +fn port_exclusive_and_passthrough_are_local_exclusions() { + let mut graph = Graph::new(); + let exclusive = graph.app_node("exclusive", MediaRole::StreamOutput, 100); + graph.port(exclusive, PortDirection::Out, true); + let client = graph.client_of_app(101); + let passthrough = graph.node( + "passthrough", + MediaRole::StreamOutput, + NodeProps { + passthrough: true, + ..app(client, 101) + }, + ); + let ok = graph.app_node("ok", MediaRole::StreamOutput, 102); + graph.port(ok, PortDirection::Out, false); + + let decisions = run(&graph, &ctx()); + assert_partition( + &decisions, + &[("ok", ok)], + &[ + ("exclusive", exclusive, "port-exclusive"), + ("passthrough", passthrough, "passthrough"), + ], + ); + // Neither is hazardous — an unlinkable stream must not taint anything. + assert_untainted(&decisions, exclusive); + assert_untainted(&decisions, passthrough); +} + +#[test] +fn a_node_with_no_props_at_all_is_eligible_and_non_streams_are_not_candidates() { + let mut graph = Graph::new(); + let bare = graph.node("bare", MediaRole::StreamOutput, NodeProps::default()); + graph.node("video", MediaRole::Other, NodeProps::default()); + graph.node("mic", MediaRole::Source, NodeProps::default()); + assert_partition(&run(&graph, &ctx()), &[("bare", bare)], &[]); +} + +// ────────────────────────────────────────────────────────────────────── +// Edge type 2 — sink → monitor, free at node granularity +// ────────────────────────────────────────────────────────────────────── + +#[test] +fn taint_crosses_app_then_sink_then_monitor_reader() { + let mut graph = Graph::new(); + let sink = graph.device_node("hw-sink", MediaRole::Sink); + let call = graph.peerspeak_node("peerspeak", 7); + graph.link(call, sink); + // A recorder reading the sink's monitor. The monitor connection IS a + // real Link whose output node is the sink itself, so no synthetic edge + // is needed at node granularity. + let rec_in = graph.app_node("recorder-in", MediaRole::StreamInput, 555); + graph.link(sink, rec_in); + let rec_out = graph.app_node("recorder-out", MediaRole::StreamOutput, 555); + + let decisions = run(&graph, &ctx()); + assert_partition( + &decisions, + &[], + &[ + ("call", call, "peerspeak-owned"), + ("rec-out", rec_out, "tainted-owner-bridge"), + ], + ); + assert_tainted(&decisions, rec_in, "tainted-upstream"); +} + +#[test] +fn playing_into_a_tainted_sink_does_not_taint_the_player() { + // Taint flows downstream only. Every app on the box plays into the same + // sink peerspeak does; if that tainted them, nothing would ever be + // shareable and the feature would ship as silence. + let mut graph = Graph::new(); + let sink = graph.device_node("hw-sink", MediaRole::Sink); + let call = graph.peerspeak_node("peerspeak", 7); + let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114); + graph.link(call, sink); + graph.link(firefox, sink); + + assert_partition( + &run(&graph, &ctx()), + &[("firefox", firefox)], + &[("call", call, "peerspeak-owned")], + ); +} + +// ────────────────────────────────────────────────────────────────────── +// Edge type 3 — the owner bridge (v3.4 §6.1.1, §6.1.2) +// ────────────────────────────────────────────────────────────────────── + +/// The measured `module-null-sink` + `module-loopback` shape. Walking Links +/// alone from the leaking output leg finds *no inbound links at all* — a +/// dead end that reads as "clean". +#[test] +fn loopback_forwarder_output_leg_is_excluded_via_the_owner_bridge() { + let mut graph = Graph::new(); + let hw = graph.device_node("hw-sink", MediaRole::Sink); + let null_sink = graph.module_node("fabletest_sink", MediaRole::Sink, 536_870_917); + let call = graph.peerspeak_node("peerspeak", 7); + graph.link(call, null_sink); + + // The tainted forwarder: legs share `pulse.module.id`, no Link between. + let lb_in = graph.module_node("input.loopback", MediaRole::StreamInput, 536_870_918); + let lb_out = graph.module_node("output.loopback", MediaRole::StreamOutput, 536_870_918); + graph.link(null_sink, lb_in); + graph.link(lb_out, hw); + + // The same shape with a clean input — impl-plan §5.1 row 1's eligible + // half. Without it the test would also pass on "exclude all forwarders". + let clean_sink = graph.device_node("clean-sink", MediaRole::Sink); + let clean_in = graph.module_node("clean-in", MediaRole::StreamInput, 536_870_920); + let clean_out = graph.module_node("clean-out", MediaRole::StreamOutput, 536_870_920); + graph.link(clean_sink, clean_in); + graph.link(clean_out, hw); + + let decisions = run(&graph, &ctx()); + assert_partition( + &decisions, + &[("clean-out", clean_out)], + &[ + ("call", call, "peerspeak-owned"), + ("lb-out", lb_out, "tainted-owner-bridge"), + ], + ); + // ...and it was the bridge, naming the key — not a Link walk. + assert_eq!( + decisions.taint[&lb_out.serial].reason, + Reason::TaintedOwnerBridge { + key: Some(OwnerKey::PulseModuleId) + } + ); +} + +#[test] +fn easyeffects_shape_bridges_from_an_audio_sink_to_its_re_emitting_leg() { + // The canonical installed case. The tainted member here is an + // `Audio/Sink`, not a `Stream/Input/Audio` — if the bridge only fired + // from stream inputs, EasyEffects would forward the whole call into the + // share at full level. + let mut graph = Graph::new(); + let hw = graph.device_node("hw-sink", MediaRole::Sink); + let ee_sink = graph.group_node("ee_sink", MediaRole::Sink, "ee", 4321); + let ee_out = graph.group_node("easyeffects-out", MediaRole::StreamOutput, "ee", 4321); + graph.link(ee_out, hw); + let call = graph.peerspeak_node("peerspeak", 7); + let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114); + graph.link(call, ee_sink); + graph.link(firefox, ee_sink); + + let decisions = run(&graph, &ctx()); + assert_partition( + &decisions, + &[("firefox", firefox)], + &[ + ("call", call, "peerspeak-owned"), + ("ee-out", ee_out, "tainted-owner-bridge"), + ], + ); + assert_eq!( + decisions.taint[&ee_out.serial].reason, + Reason::TaintedOwnerBridge { + key: Some(OwnerKey::LinkGroup) + } + ); +} + +#[test] +fn gstreamer_split_clients_bridge_on_process_id() { + // The C2 refutation, measured: one `gst-launch pulsesrc ! pulsesink` + // process produced TWO client objects (209 and 210) for its two legs. + let mut graph = Graph::new(); + let tainted_sink = graph.device_node("null-sink", MediaRole::Sink); + let call = graph.peerspeak_node("peerspeak", 7); + graph.link(call, tainted_sink); + + let client_in = graph.client(Some(PULSE_PID)); + let client_out = graph.client(Some(PULSE_PID)); + let gst_in = graph.node("gst-in", MediaRole::StreamInput, app(client_in, 20172)); + let gst_out = graph.node("gst-out", MediaRole::StreamOutput, app(client_out, 20172)); + graph.link(tainted_sink, gst_in); + + // Control: the same split-client shape reading an *untainted* source. + let mic = graph.device_node("mic", MediaRole::Source); + let ok_client_in = graph.client(Some(PULSE_PID)); + let ok_client_out = graph.client(Some(PULSE_PID)); + let ok_in = graph.node("ok-in", MediaRole::StreamInput, app(ok_client_in, 30000)); + let ok_out = graph.node("ok-out", MediaRole::StreamOutput, app(ok_client_out, 30000)); + graph.link(mic, ok_in); + + let decisions = run(&graph, &ctx()); + assert_partition( + &decisions, + &[("ok-out", ok_out)], + &[ + ("call", call, "peerspeak-owned"), + ("gst-out", gst_out, "tainted-owner-bridge"), + ], + ); + assert_eq!( + decisions.taint[&gst_out.serial].reason, + Reason::TaintedOwnerBridge { + key: Some(OwnerKey::ProcessId) + } + ); +} + +#[test] +fn owner_key_union_falls_through_a_present_but_unequal_key() { + // The §6.1.2 wording trap, asserted on the key resolver itself: a + // first-present implementation stops at `client.id`, sees 209 != 210, + // concludes "different owners", and the leak survives. + let mut graph = Graph::new(); + let a_client = graph.client(Some(PULSE_PID)); + let b_client = graph.client(Some(PULSE_PID)); + let a = graph.node("a", MediaRole::StreamInput, app(a_client, 20172)); + let b = graph.node("b", MediaRole::StreamOutput, app(b_client, 20172)); + let snapshot = graph.build(); + let (a, b) = ( + snapshot.node(a.serial).unwrap(), + snapshot.node(b.serial).unwrap(), + ); + assert_ne!(a.props.client_id, b.props.client_id); + assert_eq!( + strongest_shared_key(a, b, Some(PULSE_PID)), + Some(OwnerKey::ProcessId) + ); +} + +#[test] +fn the_strongest_shared_key_wins_when_several_match() { + let mut graph = Graph::new(); + let a = graph.group_node("a", MediaRole::StreamInput, "g", 500); + let b = graph.group_node("b", MediaRole::StreamOutput, "g", 500); + let snapshot = graph.build(); + assert_eq!( + strongest_shared_key( + snapshot.node(a.serial).unwrap(), + snapshot.node(b.serial).unwrap(), + Some(PULSE_PID) + ), + Some(OwnerKey::LinkGroup) + ); +} + +#[test] +fn the_pipewire_pulse_pid_does_not_fuse_unrelated_modules() { + // impl-plan §5.1 row 3: two Pulse modules, one tainted input. The other + // module's output MUST stay eligible — that is the only thing that + // makes wrong pipewire-pulse-PID fusion observable at all. + let mut graph = Graph::new(); + let hw = graph.device_node("hw-sink", MediaRole::Sink); + let call = graph.peerspeak_node("peerspeak", 7); + graph.link(call, hw); + + let a_in = graph.module_node("a-in", MediaRole::StreamInput, 1); + let a_out = graph.module_node("a-out", MediaRole::StreamOutput, 1); + graph.link(hw, a_in); + + let mic = graph.device_node("mic", MediaRole::Source); + let b_in = graph.module_node("b-in", MediaRole::StreamInput, 2); + let b_out = graph.module_node("b-out", MediaRole::StreamOutput, 2); + graph.link(mic, b_in); + + let decisions = run(&graph, &ctx()); + assert_partition( + &decisions, + &[("b-out", b_out)], + &[ + ("call", call, "peerspeak-owned"), + ("a-out", a_out, "tainted-owner-bridge"), + ], + ); + assert_untainted(&decisions, b_in); +} + +#[test] +fn an_unknown_pipewire_pulse_pid_over_excludes_rather_than_leaks() { + // v3.4 §6.1.2's failure-mode paragraph: if pixelpass cannot identify + // the daemon PID, key 4 stops being suppressed, every module-created + // stream fuses into one owner, and the result is broad over-exclusion — + // annoying and safe. This test pins that direction; flipping it to + // "leave key 4 out when the PID is unknown" would be a leak. + let mut graph = Graph::new(); + let hw = graph.device_node("hw-sink", MediaRole::Sink); + let call = graph.peerspeak_node("peerspeak", 7); + graph.link(call, hw); + let a_in = graph.module_node("a-in", MediaRole::StreamInput, 1); + let a_out = graph.module_node("a-out", MediaRole::StreamOutput, 1); + graph.link(hw, a_in); + let mic = graph.device_node("mic", MediaRole::Source); + let b_in = graph.module_node("b-in", MediaRole::StreamInput, 2); + let b_out = graph.module_node("b-out", MediaRole::StreamOutput, 2); + graph.link(mic, b_in); + + let decisions = run( + &graph, + &ExclusionCtx { + pipewire_pulse_pid: None, + ..ctx() + }, + ); + assert_partition( + &decisions, + &[], + &[ + ("call", call, "peerspeak-owned"), + ("a-out", a_out, "tainted-owner-bridge"), + ("b-out", b_out, "tainted-owner-bridge"), + ], + ); +} + +#[test] +fn coarse_keys_do_not_bridge_device_nodes() { + // ⚠️ A rule added during implementation, NOT present in design v3.4. + // Every ALSA device node is created by one WirePlumber process, so all + // of them share one `client.id` and one `application.process.id`. + // peerspeak's playback taints the default sink on every recompute; if + // coarse keys bridged devices, that taint would jump to the microphone + // source, and then every app holding a mic would lose its playback — + // the exact §6.1.1 catastrophe by another route. + let mut graph = Graph::new(); + let hw = graph.device_node("hw-sink", MediaRole::Sink); + let mic = graph.device_node("mic", MediaRole::Source); + let call = graph.peerspeak_node("peerspeak", 7); + graph.link(call, hw); + let ff_in = graph.app_node("firefox-mic", MediaRole::StreamInput, 11114); + let ff_out = graph.app_node("firefox-out", MediaRole::StreamOutput, 11114); + graph.link(mic, ff_in); + + let decisions = run(&graph, &ctx()); + assert_partition( + &decisions, + &[("ff-out", ff_out)], + &[("call", call, "peerspeak-owned")], + ); + assert_tainted(&decisions, hw, "tainted-upstream"); + assert_untainted(&decisions, mic); + assert_untainted(&decisions, ff_in); +} + +#[test] +fn a_module_owned_device_still_bridges_on_its_strong_key() { + // The other side of that rule: exception 2 must not disarm the bridge + // for virtual sinks that genuinely belong to a module. + let mut graph = Graph::new(); + let hw = graph.device_node("hw-sink", MediaRole::Sink); + let call = graph.peerspeak_node("peerspeak", 7); + let virt = graph.module_node("virtual-sink", MediaRole::Sink, 900_001); + let leg = graph.module_node("virtual-out", MediaRole::StreamOutput, 900_001); + graph.link(call, virt); + graph.link(leg, hw); + + let decisions = run(&graph, &ctx()); + assert_partition( + &decisions, + &[], + &[ + ("call", call, "peerspeak-owned"), + ("leg", leg, "tainted-owner-bridge"), + ], + ); +} + +#[test] +fn firefox_three_cases() { + // v3.4 §6.1.1's table, as one graph with an exact partition. + let mut graph = Graph::new(); + let hw = graph.device_node("hw-sink", MediaRole::Sink); + let mic = graph.device_node("mic", MediaRole::Source); + let call = graph.peerspeak_node("peerspeak", 7); + graph.link(call, hw); + + // 1. music only + let music = graph.app_node("ff-music", MediaRole::StreamOutput, 1001); + graph.link(music, hw); + // 2. a Meet call: a mic input leg on an untainted source + let meet_in = graph.app_node("ff-meet-in", MediaRole::StreamInput, 1002); + let meet_out = graph.app_node("ff-meet-out", MediaRole::StreamOutput, 1002); + graph.link(mic, meet_in); + graph.link(meet_out, hw); + // 3. screen-sharing with desktop audio: an input leg on a tainted monitor + let share_in = graph.app_node("ff-share-in", MediaRole::StreamInput, 1003); + let share_out = graph.app_node("ff-share-out", MediaRole::StreamOutput, 1003); + graph.link(hw, share_in); + graph.link(share_out, hw); + + assert_partition( + &run(&graph, &ctx()), + &[("music", music), ("meet-out", meet_out)], + &[ + ("call", call, "peerspeak-owned"), + ("share-out", share_out, "tainted-owner-bridge"), + ], + ); +} + +#[test] +fn an_unbounded_module_forwarder_fails_closed() { + // v3.4 §12: "a module forwarder with neither link-group nor + // pulse.module.id ⇒ unresolved ⇒ excluded". Its process id is the + // daemon's (so key 4 is suppressed) and its legs carry different + // `client.id`s, so nothing can enumerate its siblings. + let mut graph = Graph::new(); + let hw = graph.device_node("hw-sink", MediaRole::Sink); + let call = graph.peerspeak_node("peerspeak", 7); + graph.link(call, hw); + + let in_client = graph.client(Some(PULSE_PID)); + let out_client = graph.client(Some(PULSE_PID)); + let fwd_in = graph.node("fwd-in", MediaRole::StreamInput, app(in_client, PULSE_PID)); + let fwd_out = graph.node( + "fwd-out", + MediaRole::StreamOutput, + app(out_client, PULSE_PID), + ); + graph.link(hw, fwd_in); + + // The blast radius must stay small: an ordinary app is bounded by its + // own PID and must not be swept up. + let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114); + graph.link(firefox, hw); + + assert_partition( + &run(&graph, &ctx()), + &[("firefox", firefox)], + &[ + ("call", call, "peerspeak-owned"), + ("fwd-out", fwd_out, "unresolved-owner"), + ], + ); +} + +#[test] +fn audio_duplex_over_taints_known_accepted() { + // v3.4 §6.1 edge-type-2 caveat, asserted so that fixing it later is a + // deliberate change and not a silent regression. A duplex device whose + // playback side is tainted has its capture side treated as tainted too, + // so an app recording from it loses its playback leg. + let mut graph = Graph::new(); + let duplex = graph.device_node("duplex-card", MediaRole::Duplex); + let call = graph.peerspeak_node("peerspeak", 7); + graph.link(call, duplex); + let rec_in = graph.app_node("rec-in", MediaRole::StreamInput, 4444); + let rec_out = graph.app_node("rec-out", MediaRole::StreamOutput, 4444); + graph.link(duplex, rec_in); + + assert_partition( + &run(&graph, &ctx()), + &[], + &[ + ("call", call, "peerspeak-owned"), + ("rec-out", rec_out, "tainted-owner-bridge"), + ], + ); +} + +// ────────────────────────────────────────────────────────────────────── +// Unresolved ancestry — fail closed (v3.4 §6.1.4) +// ────────────────────────────────────────────────────────────────────── + +#[test] +fn a_link_from_an_unknown_node_fails_closed() { + let mut graph = Graph::new(); + let hw = graph.device_node("hw-sink", MediaRole::Sink); + let mystery_in = graph.app_node("mystery-in", MediaRole::StreamInput, 6000); + let mystery_out = graph.app_node("mystery-out", MediaRole::StreamOutput, 6000); + let ghost = graph.dangling_id(); + graph.link_ids(ghost, mystery_in.id); + graph.link(mystery_out, hw); + let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114); + + let decisions = run(&graph, &ctx()); + assert_partition( + &decisions, + &[("firefox", firefox)], + &[("mystery-out", mystery_out, "tainted-owner-bridge")], + ); + assert_tainted(&decisions, mystery_in, "unresolved-ancestry"); +} + +#[test] +fn an_ambiguous_recycled_global_id_fails_closed() { + // Two live nodes claiming one id means the observer missed a removal, + // so every edge touching that id is untrustworthy. + let mut graph = Graph::new(); + let shared = graph.dangling_id(); + let client = graph.client_of_app(7001); + let first = graph.node_with_id("first", MediaRole::StreamOutput, shared, app(client, 7001)); + let second = graph.node_with_id("second", MediaRole::StreamOutput, shared, app(client, 7002)); + let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114); + + assert_partition( + &run(&graph, &ctx()), + &[("firefox", firefox)], + &[ + ("first", first, "unresolved-ancestry"), + ("second", second, "unresolved-ancestry"), + ], + ); +} + +// ────────────────────────────────────────────────────────────────────── +// Stickiness and lifetime-awareness (v3.4 §6.1.3) +// ────────────────────────────────────────────────────────────────────── + +/// The scene the sticky tests share: one owner with a tainted input leg and +/// an output leg, plus an untouched bystander. +fn sticky_scene() -> (Graph, NodeRef, NodeRef, NodeRef, NodeRef) { + let mut graph = Graph::new(); + let hw = graph.device_node("hw-sink", MediaRole::Sink); + let call = graph.peerspeak_node("peerspeak", 7); + graph.link(call, hw); + let rec_in = graph.app_node("rec-in", MediaRole::StreamInput, 8080); + let rec_out = graph.app_node("rec-out", MediaRole::StreamOutput, 8080); + graph.link(hw, rec_in); + let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114); + (graph, call, rec_in, rec_out, firefox) +} + +#[test] +fn sticky_taint_survives_the_tainted_input_leg_disappearing() { + // The C3 buffered-audio defect: the recorder stops capturing, but its + // 5-second ring buffer is still full of peerspeak's audio and no graph + // event marks the moment it drains. + let (graph, call, rec_in, rec_out, firefox) = sticky_scene(); + let c = ctx(); + let (first, sticky) = evaluate(&graph.build(), &c, &StickyState::default()); + assert_partition( + &first, + &[("firefox", firefox)], + &[ + ("call", call, "peerspeak-owned"), + ("rec-out", rec_out, "tainted-owner-bridge"), + ], + ); + + let (second, _) = evaluate(&graph.build_without(&[rec_in]), &c, &sticky); + assert_partition( + &second, + &[("firefox", firefox)], + &[ + ("call", call, "peerspeak-owned"), + ("rec-out", rec_out, "tainted-owner-bridge"), + ], + ); + assert!( + matches!( + second.candidates[&rec_out.serial].eligibility, + Eligibility::NotEligible { sticky: true, .. } + ), + "the second exclusion must be recorded as sticky, not re-derived" + ); +} + +#[test] +fn a_new_leg_of_a_still_tainted_owner_inherits_the_taint() { + // Stickiness is per *owner*, not per node: the recorder opening a + // second output stream after its input leg closed must not escape. + let (mut graph, _, rec_in, _, _) = sticky_scene(); + let c = ctx(); + let (_, sticky) = evaluate(&graph.build(), &c, &StickyState::default()); + let late = graph.app_node("rec-out-2", MediaRole::StreamOutput, 8080); + let (second, _) = evaluate(&graph.build_without(&[rec_in]), &c, &sticky); + assert_eq!( + second.candidates[&late.serial].reason().map(Reason::code), + Some("tainted-owner-bridge") + ); +} + +#[test] +fn sticky_taint_clears_once_every_owner_member_is_gone() { + // Otherwise an app is unshareable forever, which is the over-exclusion + // half of the trade and just as much a bug. + let (mut graph, call, rec_in, rec_out, firefox) = sticky_scene(); + let c = ctx(); + let (_, sticky) = evaluate(&graph.build(), &c, &StickyState::default()); + + // Full teardown: both legs and the client object. + let recorder_client = graph.client_of_app(8080); + graph.drop_clients(&[recorder_client]); + let torn_down = graph.build_without(&[rec_in, rec_out]); + let (_, sticky) = evaluate(&torn_down, &c, &sticky); + assert!( + sticky + .owners + .iter() + .all(|owner| !owner.members.contains(&ObjectRef::Node(rec_out.serial))), + "the recorder's sticky entry should have been dropped: {sticky:#?}" + ); + + // A brand-new recorder with the same PID starts clean. + let fresh = graph.app_node("rec-out-fresh", MediaRole::StreamOutput, 8080); + let (third, _) = evaluate(&graph.build_without(&[rec_in, rec_out]), &c, &sticky); + assert_partition( + &third, + &[("firefox", firefox), ("fresh", fresh)], + &[("call", call, "peerspeak-owned")], + ); +} + +#[test] +fn recycled_ids_module_indices_and_link_groups_do_not_inherit_taint() { + // Every one of these recycles on this stack — measured for module + // indices, node ids and link-group strings. Only `object.serial` does + // not, which is exactly why stickiness is keyed on it. + let mut graph = Graph::new(); + let hw = graph.device_node("hw-sink", MediaRole::Sink); + let call = graph.peerspeak_node("peerspeak", 7); + graph.link(call, hw); + let old_in = graph.group_node("old-in", MediaRole::StreamInput, "loopback-2541-13", 6100); + let old_out = graph.group_node("old-out", MediaRole::StreamOutput, "loopback-2541-13", 6100); + graph.link(hw, old_in); + + let c = ctx(); + let (first, sticky) = evaluate(&graph.build(), &c, &StickyState::default()); + assert_eq!( + first.candidates[&old_out.serial].reason().map(Reason::code), + Some("tainted-owner-bridge") + ); + + // Teardown, then a brand-new module reusing the global id, the PID and + // the link-group string verbatim — everything except the serial. + let old_client = graph.client_of_app(6100); + graph.drop_clients(&[old_client]); + let reborn_client = graph.client(Some(PULSE_PID)); + let reborn = graph.node_with_id( + "reborn-out", + MediaRole::StreamOutput, + old_out.id, + super::fixture::link_group("loopback-2541-13", reborn_client, 6100), + ); + let (second, _) = evaluate(&graph.build_without(&[old_in, old_out]), &c, &sticky); + assert_partition( + &second, + &[("reborn", reborn)], + &[("call", call, "peerspeak-owned")], + ); +}