diff --git a/src/host/mod.rs b/src/host/mod.rs index ea87de5..ec59cd3 100644 --- a/src/host/mod.rs +++ b/src/host/mod.rs @@ -1,5 +1,6 @@ pub mod audio; mod capture; +mod observer; mod pipeline; mod quality; mod serve; diff --git a/src/host/observer/classify.rs b/src/host/observer/classify.rs new file mode 100644 index 0000000..3639c27 --- /dev/null +++ b/src/host/observer/classify.rs @@ -0,0 +1,119 @@ +//! The `session_device` classifier — pure, no PipeWire. +//! +//! `NodeProps::session_device` (see [`super::super::taint::snapshot`]) is a +//! **positive high-confidence** claim that a node is a passive hardware +//! terminal: a real sound card's sink or source that terminates audio rather +//! than forwarding it. Setting it *removes* two protections at once — the +//! node's coarse owner keys and its ability to trip the fail-closed backstop +//! — so a false positive is a **leak**, and the whole classifier is shaped so +//! that anything less than a positive identification resolves to `false`. +//! +//! The observer (phase 3) owes this classification; the adapter must never +//! stuff a raw property through. Two facts from the design (v3.4 §6.1.1, +//! Codex rounds 2–4) drive the shape here: +//! +//! - `device.id` / `device.api` describe *which* Device a node belongs to and +//! *how* that Device is reached — **neither promises the node passively +//! terminates audio.** A filter chain associated with a card satisfies +//! both. So the discriminator is `factory.name` on an **allowlist** of +//! real hardware-PCM factories, never a substring or a denylist: an unknown +//! factory is not a device. +//! - The backing Device must actually have been observed. A node that claims +//! a `device.id` we have not yet resolved is **withheld**, not admitted with +//! a provisional `false` — a provisional `false` during the not-ready +//! window fuses sink and mic on the shared session client and that fusion +//! can persist as sticky over-exclusion (round-3 finding 3). + +use crate::host::taint::snapshot::GlobalId; + +/// Factory names that positively identify a passive hardware-PCM terminal. +/// +/// **An allowlist, deliberately.** Membership *removes* protections, so the +/// safe error direction is to leave a genuine-but-unlisted device off the +/// list (it merely keeps its owner keys — over-exclusion, no echo). Adding a +/// backend here is a security-relevant change and wants the same measurement +/// the ALSA entries got (snapshot.rs `session_device` contract: the target +/// box's five ALSA nodes carry `factory.name=api.alsa.pcm.{sink,source}`; the +/// three `support.null-audio-sink` nodes carry neither). +/// +/// `support.null-audio-sink`, `*.loopback`, and any filter factory are +/// intentionally **absent**: those forward audio, which is exactly the shape +/// this feature must be able to exclude. +const HARDWARE_PCM_FACTORIES: &[&str] = &[ + // ALSA — measured on the target box. + "api.alsa.pcm.sink", + "api.alsa.pcm.source", + // BlueZ — the equivalent real backend for Bluetooth audio terminals. + "api.bluez5.pcm.sink", + "api.bluez5.pcm.source", +]; + +/// The three node properties the classifier reads, exactly as the adapter +/// parsed them off the Node global. Kept separate from +/// [`super::super::taint::snapshot::NodeProps`] because these feed the +/// *decision* whose output is the `session_device` field — they are inputs, +/// not part of the graph the engine reasons over. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct DeviceClaim { + /// `device.id` — the Device this node belongs to, if any. Absent on + /// `Stream/*` nodes, which is exactly why their absence means "not a + /// device", not "unknown". + pub device_id: Option, + /// `device.api` — the access API of that Device (e.g. `alsa`, `bluez5`). + /// Its mere presence is **not** sufficient (a card-associated filter has + /// it too); required only as a corroborating signal alongside the factory + /// allowlist. + pub device_api: Option, + /// `factory.name` — the discriminator. Only an allowlisted hardware-PCM + /// factory earns `session_device`. + pub factory_name: Option, +} + +/// The outcome of classifying one node's device claim. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub enum Classification { + /// No `device.id` — a `Stream/*` node. Admit with `session_device=false`. + NotADevice, + /// A `device.id` is claimed but the backing Device has not been resolved + /// yet. **Withhold the node and keep the readiness epoch not-ready**; + /// re-classify when the Device is observed. + Withhold { device_id: GlobalId }, + /// Positively a passive hardware terminal. Admit with + /// `session_device=true`. + SessionDevice, + /// Backed by a *resolved* Device but not a hardware-PCM terminal — a + /// filter or virtual node on a card, an unknown factory, or a Device with + /// no `device.api`. Admit with `session_device=false` (fail closed). + NotSessionDevice, +} + +/// Classify a node's device claim. +/// +/// `device_resolved` is whether [`DeviceClaim::device_id`] has been observed +/// as a Device global; it is only consulted when a `device_id` is present. +/// Pure: the model supplies `device_resolved` from its resolved-Device set, +/// and the I/O of *binding* the Device lives in the adapter. +pub fn classify(claim: &DeviceClaim, device_resolved: bool) -> Classification { + let Some(device_id) = claim.device_id else { + // No backing Device: a stream. Not withheld, not a device. + return Classification::NotADevice; + }; + if !device_resolved { + // Backed by a Device we have not seen — the one case that blocks + // readiness. A provisional answer here is the leak the contract + // forbids. + return Classification::Withhold { device_id }; + } + let is_hardware_pcm = claim.device_api.is_some() + && claim + .factory_name + .as_deref() + .is_some_and(|f| HARDWARE_PCM_FACTORIES.contains(&f)); + if is_hardware_pcm { + Classification::SessionDevice + } else { + // Resolved, but not positively a terminal: fail closed to false so + // the node keeps its owner keys and its backstop. + Classification::NotSessionDevice + } +} diff --git a/src/host/observer/mod.rs b/src/host/observer/mod.rs new file mode 100644 index 0000000..cb0abbe --- /dev/null +++ b/src/host/observer/mod.rs @@ -0,0 +1,464 @@ +//! The registry observer's **pure core** (impl plan §4, phase 3). +//! +//! This is my half of the phase-3 split: a reducer that folds a stream of +//! typed [`RegEvent`]s into a live model of the PipeWire graph and projects +//! the [`GraphSnapshot`] + context the taint engine (phase 2) consumes. **No +//! PipeWire types appear here** — the I/O adapter (Codex's half) translates +//! live registry callbacks, Link/Device binds, `/proc` reads, and the +//! `core.sync`/`done` round-trip into these events and feeds them in. Every +//! test in this module builds the event stream by hand. +//! +//! Three things this core is shaped to get right, each an exit-gate row: +//! +//! - **Removal by recycled id.** `global_remove` names only a 32-bit global +//! id, and those recycle. The model keeps an insertion-ordered index per id +//! so a removal accounts for the *oldest* generation first, and the +//! snapshot projection treats any id still claimed by two live objects as +//! [`IdLookup::Ambiguous`] — fail closed (v3.4 §6.1.3). +//! - **The readiness epoch.** `graph_ready` is false until the initial graph +//! is fully observed: the server has synced **and** no binds/withheld nodes +//! remain outstanding. A bounded timeout makes it fail closed. It gates +//! sticky *retirement* only; withholding after completion is per-object. +//! - **Withholding on unresolved devices.** A node claiming a `device.id` +//! whose Device we have not observed is held out of the snapshot entirely +//! rather than admitted with a provisional `session_device` (see +//! [`classify`]). + +#![allow(dead_code)] // Wired by the phase-3 adapter (Codex's half) and consumed by later phases. + +pub mod classify; +pub mod pulse_pid; + +#[cfg(test)] +mod tests; + +use crate::host::taint::snapshot::{ + ClientSnapshot, GlobalId, GraphSnapshot, LinkSnapshot, MediaRole, NodeProps, NodeSnapshot, + PortSnapshot, Serial, +}; +use classify::{Classification, DeviceClaim}; +use std::collections::{BTreeMap, VecDeque}; + +/// A monotonic millisecond clock value, supplied by the adapter via +/// [`RegEvent::Tick`]. Kept as a bare integer rather than +/// [`std::time::Instant`] so the readiness timeout is deterministic in tests. +pub type Millis = u64; + +/// A Node as observed off the registry, before `session_device` has been +/// decided. The adapter fills [`NodeProps`] with everything it can parse and +/// leaves `session_device` at its `false` default; the model overwrites it +/// from the [`classify`] result once the backing Device (if any) is resolved. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct NodeObservation { + pub serial: Serial, + pub id: GlobalId, + pub name: Option, + pub role: MediaRole, + pub props: NodeProps, + pub device_claim: DeviceClaim, +} + +/// The four endpoint references a Link carries. Node endpoints are required — +/// a Link with unknown nodes is useless — so this whole struct is what the +/// adapter must resolve (from the global's props if present, else by binding +/// `LinkInfoRef`, the correctness path) before a Link enters the snapshot. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct LinkEndpoints { + pub output_node: GlobalId, + pub input_node: GlobalId, + pub output_port: Option, + pub input_port: Option, +} + +/// A typed observation of the live graph. The adapter produces these; the +/// model consumes them in [`RegistryModel::apply`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum RegEvent { + /// A Node global appeared. Admitted immediately unless it claims an + /// unresolved Device (then withheld — see [`classify`]). + NodeAdded(NodeObservation), + /// A Port global appeared. + PortAdded(PortSnapshot), + /// A Client global appeared. Feeds pulse-PID derivation via `sec_pid`. + ClientAdded(ClientSnapshot), + /// A Device global appeared. Resolves any nodes withheld on its id. + DeviceAdded { id: GlobalId }, + /// A Link global appeared. `endpoints` is `Some` when the global carried + /// them (the optimisation) and `None` when the adapter must bind to learn + /// them (the correctness path) — the latter is an outstanding obligation + /// until a matching [`RegEvent::LinkEndpointsResolved`] arrives. + LinkAdded { + serial: Serial, + id: GlobalId, + endpoints: Option, + }, + /// The bind-`LinkInfoRef` fallback resolved a Link's endpoints. + LinkEndpointsResolved { + serial: Serial, + endpoints: LinkEndpoints, + }, + /// The adapter read `/proc//comm` (`None` = the read failed / the + /// process is gone). Validates the pulse-PID candidate. + ProcCommProbed { pid: u32, comm: Option }, + /// Any global was removed. Only its 32-bit id is known. + Removed { id: GlobalId }, + /// A `core.sync()` issued after the initial enumeration completed its + /// round-trip (`done`). One half of readiness; the other is that no + /// binds/withheld nodes are still outstanding. + ServerSynced, + /// A monotonic clock sample. Drives the readiness timeout only. + Tick { now: Millis }, +} + +/// Which slot in the id index a live object occupies. `global_remove` gives +/// only the id, so the index remembers what each id currently holds. A Node +/// slot's serial may live in either the admitted or the withheld map. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Slot { + Node(Serial), + Port(Serial), + Link(Serial), + Client(Serial), + Device, +} + +/// The readiness epoch. A one-time transition out of [`Readiness::Waiting`]; +/// both terminal states are sticky (a completed graph is not un-completed by +/// later per-object withholding, and a timed-out observer stays fail-closed +/// for its lifetime). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Readiness { + /// The initial enumeration is still in flight. + Waiting, + /// Server synced and every obligation resolved. `graph_ready` is true. + Complete, + /// The bounded deadline passed with obligations outstanding. + /// `graph_ready` stays false — fail closed. + TimedOut, +} + +/// The pure handoff to the taint engine: a coherent [`GraphSnapshot`] plus the +/// two context fields phase 3 owns. The caller merges these into +/// [`crate::host::taint::ExclusionCtx`] alongside `aec_module_id` (phase 4) +/// and `pixelpass_owned` (pixelpass's own tracking). +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Projection { + pub snapshot: GraphSnapshot, + pub pipewire_pulse_pid: Option, + pub graph_ready: bool, +} + +/// The live model. Folds [`RegEvent`]s; project with [`RegistryModel::project`]. +#[derive(Clone, Debug)] +pub struct RegistryModel { + // Admitted objects, keyed by their never-recycled serial. + nodes: BTreeMap, + ports: BTreeMap, + links: BTreeMap, + clients: BTreeMap, + + /// Nodes held out of the snapshot pending their Device's resolution. + withheld: BTreeMap, + /// Links whose endpoints the adapter is still binding; the id is kept so + /// removal and resolution can find them. + pending_links: BTreeMap, + + /// Live Device global ids, ref-counted so a recycled id is only + /// considered resolved while a Device actually holds it. + resolved_devices: BTreeMap, + + /// Insertion-ordered holders of each live global id. `global_remove` + /// accounts for the oldest generation first (v3.4 §6.1.3). + live_ids: BTreeMap>, + + /// `/proc//comm` reads keyed by pid, for pulse-PID validation. + probed_comm: BTreeMap>, + + server_synced: bool, + readiness: Readiness, + deadline: Millis, + last_now: Millis, +} + +impl RegistryModel { + /// `now` seeds the clock; `timeout` is the readiness budget. The deadline + /// is `now + timeout`; a [`RegEvent::Tick`] at or past it while still + /// [`Readiness::Waiting`] fails the epoch closed. + pub fn new(now: Millis, timeout: Millis) -> Self { + Self { + nodes: BTreeMap::new(), + ports: BTreeMap::new(), + links: BTreeMap::new(), + clients: BTreeMap::new(), + withheld: BTreeMap::new(), + pending_links: BTreeMap::new(), + resolved_devices: BTreeMap::new(), + live_ids: BTreeMap::new(), + probed_comm: BTreeMap::new(), + server_synced: false, + readiness: Readiness::Waiting, + deadline: now.saturating_add(timeout), + last_now: now, + } + } + + pub fn readiness(&self) -> Readiness { + self.readiness + } + + pub fn graph_ready(&self) -> bool { + matches!(self.readiness, Readiness::Complete) + } + + /// The pulse-PID candidate the adapter should be probing (`None` = no + /// repeated `sec_pid`, nothing to probe). Exposed so the adapter re-probes + /// only when the candidate changes. + pub fn pulse_pid_candidate(&self) -> Option { + let clients: Vec = self.clients.values().cloned().collect(); + pulse_pid::candidate(&clients) + } + + /// Fold one observation into the model. + pub fn apply(&mut self, event: RegEvent) { + match event { + RegEvent::NodeAdded(obs) => self.on_node_added(obs), + RegEvent::PortAdded(port) => { + self.push_id(port.id, Slot::Port(port.serial)); + self.ports.insert(port.serial, port); + } + RegEvent::ClientAdded(client) => { + self.push_id(client.id, Slot::Client(client.serial)); + self.clients.insert(client.serial, client); + // A new client can change the pulse candidate; the adapter + // learns that via `pulse_pid_candidate`. No readiness effect. + } + RegEvent::DeviceAdded { id } => self.on_device_added(id), + RegEvent::LinkAdded { + serial, + id, + endpoints, + } => self.on_link_added(serial, id, endpoints), + RegEvent::LinkEndpointsResolved { serial, endpoints } => { + self.on_link_resolved(serial, endpoints) + } + RegEvent::ProcCommProbed { pid, comm } => { + self.probed_comm.insert(pid, comm); + } + RegEvent::Removed { id } => self.on_removed(id), + RegEvent::ServerSynced => { + self.server_synced = true; + self.maybe_complete(); + } + RegEvent::Tick { now } => { + self.last_now = now; + self.maybe_timeout(now); + } + } + } + + fn on_node_added(&mut self, obs: NodeObservation) { + self.push_id(obs.id, Slot::Node(obs.serial)); + let resolved = obs + .device_claim + .device_id + .is_some_and(|id| self.device_resolved(id)); + match classify::classify(&obs.device_claim, resolved) { + Classification::Withhold { .. } => { + self.withheld.insert(obs.serial, obs); + } + Classification::SessionDevice => self.admit_node(obs, true), + Classification::NotADevice | Classification::NotSessionDevice => { + self.admit_node(obs, false) + } + } + // Withholding a node adds an obligation; admitting one can never + // complete readiness on its own, but re-check is cheap and keeps the + // invariant local. + self.maybe_complete(); + } + + fn admit_node(&mut self, obs: NodeObservation, session_device: bool) { + let mut props = obs.props; + props.session_device = session_device; + self.nodes.insert( + obs.serial, + NodeSnapshot { + serial: obs.serial, + id: obs.id, + name: obs.name, + role: obs.role, + props, + }, + ); + } + + fn on_device_added(&mut self, id: GlobalId) { + self.push_id(id, Slot::Device); + *self.resolved_devices.entry(id).or_insert(0) += 1; + // Admit every node that was withheld waiting on exactly this Device. + let ready: Vec = self + .withheld + .iter() + .filter(|(_, obs)| obs.device_claim.device_id == Some(id)) + .map(|(&serial, _)| serial) + .collect(); + for serial in ready { + if let Some(obs) = self.withheld.remove(&serial) { + // Resolved now, so classify yields a terminal answer, never + // Withhold again. + let session_device = matches!( + classify::classify(&obs.device_claim, true), + Classification::SessionDevice + ); + self.admit_node(obs, session_device); + } + } + self.maybe_complete(); + } + + fn on_link_added(&mut self, serial: Serial, id: GlobalId, endpoints: Option) { + self.push_id(id, Slot::Link(serial)); + match endpoints { + Some(e) => { + self.links.insert(serial, link_snapshot(serial, id, e)); + } + None => { + // Correctness path: withhold the Link until the bind fallback + // resolves it. Counts as an outstanding obligation. + self.pending_links.insert(serial, id); + } + } + self.maybe_complete(); + } + + fn on_link_resolved(&mut self, serial: Serial, endpoints: LinkEndpoints) { + // `remove` also guards against a stale resolution for a Link already + // gone: unknown serial ⇒ ignore. + if let Some(id) = self.pending_links.remove(&serial) { + self.links + .insert(serial, link_snapshot(serial, id, endpoints)); + self.maybe_complete(); + } + } + + fn on_removed(&mut self, id: GlobalId) { + let Some(queue) = self.live_ids.get_mut(&id) else { + tracing::warn!(global_id = id.0, "observer: remove for an id we never saw"); + return; + }; + // Oldest generation first — the id may be shared during a + // missed-removal window. + let slot = queue.pop_front(); + if queue.is_empty() { + self.live_ids.remove(&id); + } + match slot { + Some(Slot::Node(serial)) => { + if self.nodes.remove(&serial).is_none() { + // Was still withheld — drop the obligation. + self.withheld.remove(&serial); + } + } + Some(Slot::Port(serial)) => { + self.ports.remove(&serial); + } + Some(Slot::Link(serial)) => { + self.links.remove(&serial); + self.pending_links.remove(&serial); + } + Some(Slot::Client(serial)) => { + self.clients.remove(&serial); + } + Some(Slot::Device) => { + if let Some(count) = self.resolved_devices.get_mut(&id) { + *count -= 1; + if *count == 0 { + self.resolved_devices.remove(&id); + } + } + } + None => { + tracing::warn!(global_id = id.0, "observer: empty id slot on remove"); + } + } + // A removal can drain the last obligation (a withheld node or pending + // link vanished before it resolved). + self.maybe_complete(); + } + + fn push_id(&mut self, id: GlobalId, slot: Slot) { + self.live_ids.entry(id).or_default().push_back(slot); + } + + fn device_resolved(&self, id: GlobalId) -> bool { + self.resolved_devices.get(&id).is_some_and(|&n| n > 0) + } + + /// Every obligation that must clear before the initial graph is trusted: + /// no node withheld on an unresolved Device, no Link awaiting its bind. + fn obligations_outstanding(&self) -> bool { + !self.withheld.is_empty() || !self.pending_links.is_empty() + } + + /// Completion needs no clock — only the sync flag and an empty obligation + /// set — so it may fire on any mutating event. Sticky once reached. + fn maybe_complete(&mut self) { + if self.readiness != Readiness::Waiting { + return; + } + if self.server_synced && !self.obligations_outstanding() { + self.readiness = Readiness::Complete; + tracing::info!("observer: readiness epoch reached (synced + no obligations)"); + } + } + + /// Only the timeout consults the clock. + fn maybe_timeout(&mut self, now: Millis) { + if self.readiness != Readiness::Waiting { + return; + } + if now >= self.deadline { + self.readiness = Readiness::TimedOut; + tracing::warn!( + withheld = self.withheld.len(), + pending_links = self.pending_links.len(), + "observer: readiness epoch timed out with obligations outstanding — fail closed" + ); + } + } + + /// pipewire-pulse's PID from the current clients, validated against the + /// probed `comm`. `None` whenever anything is ambiguous or unconfirmed — + /// the safe answer (key 4 unusable). + fn pulse_pid(&self) -> Option { + let candidate = self.pulse_pid_candidate()?; + let comm = self.probed_comm.get(&candidate).and_then(|c| c.as_deref()); + pulse_pid::validate(candidate, comm) + } + + /// Project the current state into the taint engine's inputs. + pub fn project(&self) -> Projection { + let snapshot = GraphSnapshot::new( + self.nodes.values().cloned().collect(), + self.ports.values().cloned().collect(), + self.links.values().cloned().collect(), + self.clients.values().cloned().collect(), + ); + Projection { + snapshot, + pipewire_pulse_pid: self.pulse_pid(), + graph_ready: self.graph_ready(), + } + } +} + +fn link_snapshot(serial: Serial, id: GlobalId, e: LinkEndpoints) -> LinkSnapshot { + LinkSnapshot { + serial, + id, + output_node: e.output_node, + input_node: e.input_node, + output_port: e.output_port, + input_port: e.input_port, + } +} diff --git a/src/host/observer/pulse_pid.rs b/src/host/observer/pulse_pid.rs new file mode 100644 index 0000000..ccdf663 --- /dev/null +++ b/src/host/observer/pulse_pid.rs @@ -0,0 +1,89 @@ +//! Deriving pipewire-pulse's own PID — pure, no PipeWire and no `/proc` I/O. +//! +//! The owner bridge's key 4 is `application.process.id`. For a stream created +//! by a **Pulse-emulated** client that PID is *pipewire-pulse's own*, shared +//! verbatim across every unrelated Pulse app, so bridging on it would fuse +//! every Pulse module into one tainted owner (design v3.4 §5.2 correction 5, +//! §6.1.2). The engine therefore needs to know that one PID so it can refuse +//! to bridge on it — and **every** way of deriving it can fail, in which case +//! the safe answer is `None`: key 4 becomes unusable (coarser, never wrong). +//! +//! The derivation is split into two pure stages so the I/O — reading +//! `/proc//comm` — stays in the adapter: +//! +//! 1. [`candidate`] finds the PID that *looks* like pulse from the graph +//! alone: the `pipewire.sec.pid` value shared across multiple Clients. +//! Native PipeWire clients carry their own distinct PID; only the +//! Pulse shim repeats one value, so a repeated value is the signal. +//! 2. [`validate`] confirms that candidate against the `comm` the adapter +//! read from `/proc`. This is what closes **PID reuse**: a recycled PID +//! that coincidentally repeats in the graph is rejected because +//! `/proc//comm` now names a different process. +//! +//! Any failure at either stage — no repeated value, two repeated values, +//! the property missing, `/proc` gone, a `comm` mismatch — yields `None`. + +use crate::host::taint::snapshot::ClientSnapshot; +use std::collections::BTreeMap; + +/// The kernel `comm` of the pipewire-pulse process. `comm` is truncated to +/// 15 bytes by the kernel; `pipewire-pulse` is 14 bytes, so it is exact — +/// and exact is the only safe match, since a prefix match would accept a +/// recycled PID belonging to e.g. `pipewire-pulseX`. +const PULSE_COMM: &str = "pipewire-pulse"; + +/// Stage 1: the PID that looks like pipewire-pulse from the client graph. +/// +/// Returns `Some(pid)` only when **exactly one** `pipewire.sec.pid` value is +/// shared by two or more clients. Rationale, matched to the failure matrix: +/// +/// - **consistent** — one value repeats, the rest (native clients) are +/// distinct ⇒ that value. +/// - **inconsistent** — two or more values each repeat ⇒ we cannot tell which +/// is pulse ⇒ `None`. +/// - **missing property** — the Pulse clients carry no `sec_pid` ⇒ nothing +/// repeats ⇒ `None`. +/// +/// A count threshold of two is deliberate: a single client carrying a PID is +/// indistinguishable from a lone native app, and pulse always mints many. +pub fn candidate(clients: &[ClientSnapshot]) -> Option { + let mut counts: BTreeMap = BTreeMap::new(); + for client in clients { + if let Some(pid) = client.sec_pid { + *counts.entry(pid).or_insert(0) += 1; + } + } + + // Every PID seen on 2+ clients is a pulse candidate. If there is exactly + // one such PID we trust it; zero or several ⇒ fail closed. + let mut repeated = counts.iter().filter(|&(_, &n)| n >= 2).map(|(&pid, _)| pid); + let first = repeated.next()?; + if repeated.next().is_some() { + // Ambiguous: more than one value repeats. + return None; + } + Some(first) +} + +/// Stage 2: confirm the candidate against the `comm` read from +/// `/proc//comm`. +/// +/// `comm` is `None` when the adapter's read failed — the `/proc` entry is +/// gone (the process exited between derivation and probe) — which is itself a +/// reason to fail closed. A present-but-different `comm` is the **PID reuse** +/// guard: the number is live but now belongs to someone else. +pub fn validate(candidate: u32, comm: Option<&str>) -> Option { + match comm { + Some(PULSE_COMM) => Some(candidate), + _ => None, + } +} + +/// The two stages composed, for callers that already hold the probed `comm`. +/// The model keeps them separate (it recomputes the candidate as clients +/// churn, and only re-probes when the candidate *changes*), so this is a +/// convenience for tests and for the fully-resolved path. +pub fn derive(clients: &[ClientSnapshot], comm_of: impl Fn(u32) -> Option) -> Option { + let candidate = candidate(clients)?; + validate(candidate, comm_of(candidate).as_deref()) +} diff --git a/src/host/observer/tests.rs b/src/host/observer/tests.rs new file mode 100644 index 0000000..a5eae36 --- /dev/null +++ b/src/host/observer/tests.rs @@ -0,0 +1,620 @@ +//! Pure exit-gate coverage for the phase-3 observer core. +//! +//! Five of the six exit-gate rows live here (the sixth — a live create/destroy +//! topology diff — needs the daemon and belongs to the adapter). Each test +//! builds the [`RegEvent`] stream by hand; nothing links PipeWire. +//! +//! Carrying the phase-0a lesson: the id/pid/serial tests use **interior** +//! values, not just 1 and a huge number, so a middle-of-range mistake cannot +//! hide. + +use super::classify::{Classification, DeviceClaim, classify}; +use super::pulse_pid; +use super::*; +use crate::host::taint::snapshot::{ + ClientSnapshot, GlobalId, IdLookup, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial, +}; + +// ---- builders ------------------------------------------------------------- + +fn ser(n: u64) -> Serial { + Serial(n) +} +fn gid(n: u32) -> GlobalId { + GlobalId(n) +} +fn model() -> RegistryModel { + // now=0, a 5 s readiness budget. + RegistryModel::new(0, 5000) +} + +fn no_device() -> DeviceClaim { + DeviceClaim::default() +} + +fn hw_claim(device_id: u32, api: &str, factory: &str) -> DeviceClaim { + DeviceClaim { + device_id: Some(gid(device_id)), + device_api: Some(api.to_string()), + factory_name: Some(factory.to_string()), + } +} + +/// A `Stream/Output/Audio` node with no backing Device — admitted at once. +fn stream_out(serial: u64, id: u32) -> RegEvent { + RegEvent::NodeAdded(NodeObservation { + serial: ser(serial), + id: gid(id), + name: Some(format!("stream-{id}")), + role: MediaRole::StreamOutput, + props: NodeProps::default(), + device_claim: no_device(), + }) +} + +/// A node backed by a Device (withheld until that Device resolves). +fn device_node(serial: u64, id: u32, role: MediaRole, claim: DeviceClaim) -> RegEvent { + RegEvent::NodeAdded(NodeObservation { + serial: ser(serial), + id: gid(id), + name: Some(format!("dev-node-{id}")), + role, + props: NodeProps::default(), + device_claim: claim, + }) +} + +fn client(serial: u64, id: u32, sec_pid: Option) -> RegEvent { + RegEvent::ClientAdded(ClientSnapshot { + serial: ser(serial), + id: gid(id), + sec_pid, + }) +} + +fn port(serial: u64, id: u32, node_id: u32, dir: PortDirection) -> RegEvent { + RegEvent::PortAdded(PortSnapshot { + serial: ser(serial), + id: gid(id), + node: gid(node_id), + direction: dir, + exclusive: false, + monitor: false, + }) +} + +fn endpoints(out_node: u32, in_node: u32) -> LinkEndpoints { + LinkEndpoints { + output_node: gid(out_node), + input_node: gid(in_node), + output_port: None, + input_port: None, + } +} + +// ========================================================================== +// classify() — session_device +// ========================================================================== + +#[test] +fn classify_no_device_is_not_a_device() { + assert_eq!(classify(&no_device(), false), Classification::NotADevice); + // `device_resolved` is irrelevant with no device_id. + assert_eq!(classify(&no_device(), true), Classification::NotADevice); +} + +#[test] +fn classify_unresolved_device_withholds() { + let claim = hw_claim(42, "alsa", "api.alsa.pcm.sink"); + assert_eq!( + classify(&claim, false), + Classification::Withhold { device_id: gid(42) } + ); +} + +#[test] +fn classify_resolved_hardware_pcm_is_session_device() { + for factory in [ + "api.alsa.pcm.sink", + "api.alsa.pcm.source", + "api.bluez5.pcm.sink", + "api.bluez5.pcm.source", + ] { + let api = if factory.contains("bluez5") { + "bluez5" + } else { + "alsa" + }; + assert_eq!( + classify(&hw_claim(7, api, factory), true), + Classification::SessionDevice, + "factory {factory} should be a session device" + ); + } +} + +#[test] +fn classify_resolved_but_not_hardware_pcm_fails_closed() { + // A null sink, a loopback, and an unknown factory are all forwarders, not + // terminals: resolved, but session_device stays false. + for factory in ["support.null-audio-sink", "api.alsa.pcm.loopback", "wat"] { + assert_eq!( + classify(&hw_claim(7, "alsa", factory), true), + Classification::NotSessionDevice, + "factory {factory} must not be a session device" + ); + } +} + +#[test] +fn classify_missing_device_api_fails_closed() { + // Even with an allowlisted factory, no device.api ⇒ not positively a + // real-backend terminal. + let claim = DeviceClaim { + device_id: Some(gid(7)), + device_api: None, + factory_name: Some("api.alsa.pcm.sink".to_string()), + }; + assert_eq!(classify(&claim, true), Classification::NotSessionDevice); +} + +#[test] +fn classify_allowlist_is_exact_not_substring() { + // A factory that merely *contains* an allowlisted name must not pass. + let claim = hw_claim(7, "alsa", "api.alsa.pcm.sink.evil"); + assert_eq!(classify(&claim, true), Classification::NotSessionDevice); +} + +// ========================================================================== +// pulse_pid — the six-case derivation matrix +// ========================================================================== + +fn clients_with(pids: &[Option]) -> Vec { + pids.iter() + .enumerate() + .map(|(i, &sec_pid)| ClientSnapshot { + serial: ser(1000 + i as u64), + id: gid(200 + i as u32), + sec_pid, + }) + .collect() +} + +#[test] +fn pid_candidate_consistent_repeated_value() { + // interior pid values, not 1 / u32::MAX. + let cs = clients_with(&[Some(4137), Some(4137), Some(9001), Some(12034)]); + assert_eq!(pulse_pid::candidate(&cs), Some(4137)); +} + +#[test] +fn pid_candidate_inconsistent_two_repeats_is_none() { + let cs = clients_with(&[Some(4137), Some(4137), Some(9001), Some(9001)]); + assert_eq!(pulse_pid::candidate(&cs), None); +} + +#[test] +fn pid_candidate_missing_property_is_none() { + let cs = clients_with(&[None, None, None]); + assert_eq!(pulse_pid::candidate(&cs), None); +} + +#[test] +fn pid_candidate_single_occurrence_is_none() { + // A lone native client carrying its own pid is indistinguishable from a + // one-client pulse; the >=2 threshold rejects it. + let cs = clients_with(&[Some(4137), Some(9001), Some(12034)]); + assert_eq!(pulse_pid::candidate(&cs), None); +} + +#[test] +fn pid_validate_matches_pulse_comm() { + assert_eq!( + pulse_pid::validate(4137, Some("pipewire-pulse")), + Some(4137) + ); +} + +#[test] +fn pid_validate_proc_missing_is_none() { + // case 4: /proc entry gone. + assert_eq!(pulse_pid::validate(4137, None), None); +} + +#[test] +fn pid_validate_comm_mismatch_is_none() { + // case 5: a different process holds the number. + assert_eq!(pulse_pid::validate(4137, Some("firefox")), None); +} + +#[test] +fn pid_validate_reuse_named_other_process_is_none() { + // case 6: PID reuse — the number is live but /proc names someone else. + assert_eq!(pulse_pid::validate(4137, Some("Xwayland")), None); + // and a truncation-adjacent near-miss must not pass an exact match. + assert_eq!(pulse_pid::validate(4137, Some("pipewire-pulseX")), None); +} + +#[test] +fn pid_derive_end_to_end_valid() { + let cs = clients_with(&[Some(4137), Some(4137), Some(9001)]); + let got = pulse_pid::derive(&cs, |pid| { + (pid == 4137).then(|| "pipewire-pulse".to_string()) + }); + assert_eq!(got, Some(4137)); +} + +// ========================================================================== +// model — pulse pid through project() +// ========================================================================== + +/// Drive the model to Complete so `project` reflects a trusted graph, without +/// caring about the specific objects. +fn drive_ready(m: &mut RegistryModel) { + m.apply(RegEvent::ServerSynced); +} + +#[test] +fn model_pulse_pid_valid_through_projection() { + let mut m = model(); + m.apply(client(1, 200, Some(4137))); + m.apply(client(2, 201, Some(4137))); + m.apply(client(3, 202, Some(9001))); + assert_eq!(m.pulse_pid_candidate(), Some(4137)); + m.apply(RegEvent::ProcCommProbed { + pid: 4137, + comm: Some("pipewire-pulse".to_string()), + }); + assert_eq!(m.project().pipewire_pulse_pid, Some(4137)); +} + +#[test] +fn model_pulse_pid_none_until_probed() { + let mut m = model(); + m.apply(client(1, 200, Some(4137))); + m.apply(client(2, 201, Some(4137))); + // candidate exists, but no /proc confirmation yet ⇒ fail closed. + assert_eq!(m.project().pipewire_pulse_pid, None); +} + +#[test] +fn model_pulse_pid_none_on_comm_mismatch() { + let mut m = model(); + m.apply(client(1, 200, Some(4137))); + m.apply(client(2, 201, Some(4137))); + m.apply(RegEvent::ProcCommProbed { + pid: 4137, + comm: Some("firefox".to_string()), + }); + assert_eq!(m.project().pipewire_pulse_pid, None); +} + +// ========================================================================== +// model — add / remove of all four object types +// ========================================================================== + +#[test] +fn model_adds_all_four_object_types() { + let mut m = model(); + m.apply(stream_out(100, 50)); + m.apply(port(101, 60, 50, PortDirection::Out)); + m.apply(client(102, 70, Some(4137))); + m.apply(RegEvent::LinkAdded { + serial: ser(103), + id: gid(80), + endpoints: Some(endpoints(50, 55)), + }); + let snap = m.project().snapshot; + assert_eq!(snap.nodes().count(), 1); + assert_eq!(snap.ports().count(), 1); + assert_eq!(snap.clients().count(), 1); + assert_eq!(snap.links().count(), 1); +} + +#[test] +fn model_removes_all_four_object_types() { + let mut m = model(); + m.apply(stream_out(100, 50)); + m.apply(port(101, 60, 50, PortDirection::Out)); + m.apply(client(102, 70, Some(4137))); + m.apply(RegEvent::LinkAdded { + serial: ser(103), + id: gid(80), + endpoints: Some(endpoints(50, 55)), + }); + + m.apply(RegEvent::Removed { id: gid(50) }); + m.apply(RegEvent::Removed { id: gid(60) }); + m.apply(RegEvent::Removed { id: gid(70) }); + m.apply(RegEvent::Removed { id: gid(80) }); + + let snap = m.project().snapshot; + assert_eq!(snap.nodes().count(), 0); + assert_eq!(snap.ports().count(), 0); + assert_eq!(snap.clients().count(), 0); + assert_eq!(snap.links().count(), 0); +} + +#[test] +fn model_remove_of_unknown_id_is_harmless() { + let mut m = model(); + m.apply(stream_out(100, 50)); + m.apply(RegEvent::Removed { id: gid(999) }); + assert_eq!(m.project().snapshot.nodes().count(), 1); +} + +// ========================================================================== +// model — recycled global id, oldest generation first (fail closed) +// ========================================================================== + +#[test] +fn model_recycled_id_is_ambiguous_until_removal_accounted() { + let mut m = model(); + // A missed removal: two live nodes claim id 50 (serials 100 then 200). + m.apply(stream_out(100, 50)); + m.apply(stream_out(200, 50)); + + // The snapshot fails closed: id 50 is ambiguous. + let snap = m.project().snapshot; + assert_eq!(snap.node_by_id(gid(50)), Some(IdLookup::Ambiguous)); + assert_eq!(snap.nodes().count(), 2); + + // One removal accounts for the OLDEST generation (serial 100); the newer + // node survives and the id is unambiguous again. + m.apply(RegEvent::Removed { id: gid(50) }); + let snap = m.project().snapshot; + assert_eq!(snap.node_by_id(gid(50)), Some(IdLookup::Unique(ser(200)))); + assert!(snap.node(ser(200)).is_some()); + assert!(snap.node(ser(100)).is_none()); +} + +// ========================================================================== +// model — Link endpoint resolution (bind fallback path) +// ========================================================================== + +#[test] +fn model_link_with_endpoints_appears_immediately() { + let mut m = model(); + m.apply(RegEvent::LinkAdded { + serial: ser(103), + id: gid(80), + endpoints: Some(endpoints(50, 55)), + }); + assert_eq!(m.project().snapshot.links().count(), 1); +} + +#[test] +fn model_link_without_endpoints_is_withheld_until_resolved() { + let mut m = model(); + // The correctness path: the global carried no endpoint props. + m.apply(RegEvent::LinkAdded { + serial: ser(103), + id: gid(80), + endpoints: None, + }); + // Not in the snapshot yet, and it blocks readiness. + assert_eq!(m.project().snapshot.links().count(), 0); + m.apply(RegEvent::ServerSynced); + assert!(!m.graph_ready(), "pending link must hold readiness"); + + // The bind fallback resolves it. + m.apply(RegEvent::LinkEndpointsResolved { + serial: ser(103), + endpoints: endpoints(50, 55), + }); + let snap = m.project().snapshot; + assert_eq!(snap.links().count(), 1); + let link = snap.links().next().unwrap(); + assert_eq!(link.output_node, gid(50)); + assert_eq!(link.input_node, gid(55)); + assert!( + m.graph_ready(), + "resolving the last obligation completes readiness" + ); +} + +#[test] +fn model_stale_link_resolution_is_ignored() { + let mut m = model(); + m.apply(RegEvent::LinkAdded { + serial: ser(103), + id: gid(80), + endpoints: None, + }); + // Link removed before the bind returned. + m.apply(RegEvent::Removed { id: gid(80) }); + // A late resolution for the gone link must not resurrect it. + m.apply(RegEvent::LinkEndpointsResolved { + serial: ser(103), + endpoints: endpoints(50, 55), + }); + assert_eq!(m.project().snapshot.links().count(), 0); + m.apply(RegEvent::ServerSynced); + assert!( + m.graph_ready(), + "the obligation cleared when the link was removed" + ); +} + +// ========================================================================== +// model — readiness epoch +// ========================================================================== + +#[test] +fn model_readiness_waits_for_sync() { + let mut m = model(); + m.apply(stream_out(100, 50)); + assert_eq!(m.readiness(), Readiness::Waiting); + assert!(!m.project().graph_ready); + m.apply(RegEvent::ServerSynced); + assert_eq!(m.readiness(), Readiness::Complete); + assert!(m.project().graph_ready); +} + +#[test] +fn model_readiness_does_not_release_with_obligation_outstanding() { + let mut m = model(); + // A node withheld on an unresolved device is an outstanding obligation. + m.apply(device_node( + 100, + 50, + MediaRole::Sink, + hw_claim(42, "alsa", "api.alsa.pcm.sink"), + )); + m.apply(RegEvent::ServerSynced); + // Synced, but the withheld node keeps the epoch shut. + assert_eq!(m.readiness(), Readiness::Waiting); + assert!(!m.graph_ready()); + + // Resolving the device admits the node and completes readiness. + m.apply(RegEvent::DeviceAdded { id: gid(42) }); + assert_eq!(m.readiness(), Readiness::Complete); + assert!(m.graph_ready()); +} + +#[test] +fn model_readiness_times_out_fail_closed() { + let mut m = model(); + m.apply(device_node( + 100, + 50, + MediaRole::Sink, + hw_claim(42, "alsa", "api.alsa.pcm.sink"), + )); + m.apply(RegEvent::ServerSynced); + assert_eq!(m.readiness(), Readiness::Waiting); + + // The device never resolves; the deadline passes. + m.apply(RegEvent::Tick { now: 5000 }); + assert_eq!(m.readiness(), Readiness::TimedOut); + assert!(!m.graph_ready(), "timeout fails closed"); +} + +#[test] +fn model_tick_before_deadline_does_not_time_out() { + let mut m = model(); + m.apply(device_node( + 100, + 50, + MediaRole::Sink, + hw_claim(42, "alsa", "api.alsa.pcm.sink"), + )); + m.apply(RegEvent::Tick { now: 4999 }); + assert_eq!(m.readiness(), Readiness::Waiting); +} + +#[test] +fn model_complete_is_sticky_across_later_churn() { + let mut m = model(); + m.apply(RegEvent::ServerSynced); + assert_eq!(m.readiness(), Readiness::Complete); + // A node withheld AFTER completion must not un-complete the epoch — post + // enumeration, withholding is per-object (the node is simply absent). + m.apply(device_node( + 100, + 50, + MediaRole::Sink, + hw_claim(42, "alsa", "api.alsa.pcm.sink"), + )); + assert_eq!(m.readiness(), Readiness::Complete); + assert!(m.graph_ready()); + // ...and a late timeout Tick is inert once Complete. + m.apply(RegEvent::Tick { now: 100_000 }); + assert_eq!(m.readiness(), Readiness::Complete); +} + +#[test] +fn model_withheld_node_removed_clears_obligation() { + let mut m = model(); + m.apply(device_node( + 100, + 50, + MediaRole::Sink, + hw_claim(42, "alsa", "api.alsa.pcm.sink"), + )); + m.apply(RegEvent::ServerSynced); + assert_eq!(m.readiness(), Readiness::Waiting); + // The withheld node disappears before its device ever showed up. + m.apply(RegEvent::Removed { id: gid(50) }); + assert_eq!(m.readiness(), Readiness::Complete); +} + +// ========================================================================== +// model — device withholding & session_device flag +// ========================================================================== + +#[test] +fn model_device_first_admits_node_immediately() { + let mut m = model(); + // Device enumerated before the node that references it. + m.apply(RegEvent::DeviceAdded { id: gid(42) }); + m.apply(device_node( + 100, + 50, + MediaRole::Sink, + hw_claim(42, "alsa", "api.alsa.pcm.sink"), + )); + let snap = m.project().snapshot; + let node = snap.node(ser(100)).expect("node admitted immediately"); + assert!( + node.props.session_device, + "hardware sink is a session device" + ); + // No obligation ⇒ a sync completes readiness. + m.apply(RegEvent::ServerSynced); + assert!(m.graph_ready()); +} + +#[test] +fn model_withheld_node_admitted_with_correct_session_device() { + let mut m = model(); + // A real hardware sink and a card-associated filter share client/device + // ancestry but classify differently once the device resolves. + m.apply(device_node( + 100, + 50, + MediaRole::Sink, + hw_claim(42, "alsa", "api.alsa.pcm.sink"), + )); + m.apply(device_node( + 200, + 51, + MediaRole::Sink, + hw_claim(42, "alsa", "support.null-audio-sink"), + )); + // Both withheld until the device resolves. + assert_eq!(m.project().snapshot.nodes().count(), 0); + m.apply(RegEvent::DeviceAdded { id: gid(42) }); + + let snap = m.project().snapshot; + assert_eq!( + snap.nodes().count(), + 2, + "both admitted once the device resolved" + ); + assert!( + snap.node(ser(100)).unwrap().props.session_device, + "the real hardware sink is a session device" + ); + assert!( + !snap.node(ser(200)).unwrap().props.session_device, + "the null sink sharing the same device is not" + ); +} + +#[test] +fn model_withheld_filter_admitted_as_not_session_device() { + let mut m = model(); + m.apply(device_node( + 200, + 51, + MediaRole::Sink, + hw_claim(42, "alsa", "support.null-audio-sink"), + )); + m.apply(RegEvent::DeviceAdded { id: gid(42) }); + let snap = m.project().snapshot; + assert!( + !snap.node(ser(200)).unwrap().props.session_device, + "a null sink on a card is not a session device" + ); +}