From 8206864a43f8979ae6ddd6c2bf42e805186686aa Mon Sep 17 00:00:00 2001 From: Mollusk Date: Fri, 24 Jul 2026 12:43:59 -0400 Subject: [PATCH 1/4] =?UTF-8?q?host/observer:=20phase=203=20pure=20core=20?= =?UTF-8?q?=E2=80=94=20RegistryModel=20reducer=20+=20classifiers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit My half of the phase-3 split (impl plan §4). Pure, no PipeWire: the adapter (Codex's half) translates live registry callbacks / binds / /proc reads / core.sync into RegEvents and feeds this reducer. - RegistryModel::apply folds RegEvents into serial-keyed maps with an insertion-ordered id index so global_remove accounts for the oldest generation first; recycled ids stay Ambiguous until accounted (v3.4 §6.1.3). - Readiness epoch: graph_ready false until ServerSynced + no outstanding obligations (withheld nodes, pending link binds); bounded timeout fails closed. Gates sticky retirement only; sticky once terminal. - session_device classifier: hardware-PCM factory allowlist, exact match, fail closed to false; a node on an unresolved Device is withheld, never admitted provisional. - pulse-PID derivation split into pure candidate (repeated sec_pid) + validate (/proc comm), so the 6-case failure matrix is unit-testable; any failure => None (key 4 unusable). 34 tests cover 5 of 6 exit-gate rows (the live topology-diff row is the adapter's). cargo test --bins 117 green, fmt + clippy clean. Co-Authored-By: Claude Opus 4.8 --- src/host/mod.rs | 1 + src/host/observer/classify.rs | 119 +++++++ src/host/observer/mod.rs | 464 ++++++++++++++++++++++++ src/host/observer/pulse_pid.rs | 89 +++++ src/host/observer/tests.rs | 620 +++++++++++++++++++++++++++++++++ 5 files changed, 1293 insertions(+) create mode 100644 src/host/observer/classify.rs create mode 100644 src/host/observer/mod.rs create mode 100644 src/host/observer/pulse_pid.rs create mode 100644 src/host/observer/tests.rs 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" + ); +} From 66a0dd54df0148338b8238702ad14b20417618fd Mon Sep 17 00:00:00 2001 From: Mollusk Date: Fri, 24 Jul 2026 13:06:36 -0400 Subject: [PATCH 2/4] host/observer: phase 3 I/O adapter (read-only registry observer) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex authored the adapter half of the phase-3 split; I reviewed it and applied one robustness fix (below). Additive/read-only — does not yet replace the existing audio.rs router (that migration is a later integration phase). adapter.rs: a dedicated libpipewire main-loop thread translating registry globals into RegEvents and publishing the latest Projection via RegistryObserverHandle::latest(). core.sync(0)/done is matched one-shot → ServerSynced; a 250 ms loop timer emits Tick for the fail-closed readiness timeout; pulse-PID candidates are probed from /proc//comm only when the candidate changes; Links missing endpoint props are bound (LinkInfoRef, weak back-ref to avoid the listener cycle) and resolved via LinkEndpointsResolved. mod.rs: `pub mod adapter;`. audio.rs: parse_object_serial → pub(crate) so the adapter reuses the strict 64-bit parser. Review fix: record_global was called unconditionally per global (including unknown object types and dropped globals), which could desync the bound-link FIFO from the model's live_ids and leak a Link proxy on a recycled id. Now folded into `add()` so a slot is recorded only when an Added event is applied — the two id queues are provably lockstep. Exit gate complete: 5 pure rows + the live topology-diff row (row 6) — the #[ignore] adapter test PASSES on this host against the live daemon (module-null-sink + module-loopback observed appearing and disappearing). cargo test --bins 117 + 1 live green, clippy clean, no fmt sweep. Co-Authored-By: Codex (gpt-5.6-sol) Co-Authored-By: Claude Opus 4.8 --- src/host/audio.rs | 2 +- src/host/observer/adapter.rs | 600 +++++++++++++++++++++++++++++++++++ src/host/observer/mod.rs | 1 + 3 files changed, 602 insertions(+), 1 deletion(-) create mode 100644 src/host/observer/adapter.rs diff --git a/src/host/audio.rs b/src/host/audio.rs index b782857..615538e 100644 --- a/src/host/audio.rs +++ b/src/host/audio.rs @@ -615,7 +615,7 @@ fn run_router( /// (empty, signed, whitespace-padded, non-numeric, overflowing) is a /// property we do not understand and must not guess at. Leading zeroes /// are accepted — they are unambiguous and parse to the same value. -fn parse_object_serial(raw: &str) -> Option { +pub(crate) fn parse_object_serial(raw: &str) -> Option { if raw.is_empty() || !raw.bytes().all(|b| b.is_ascii_digit()) { return None; } diff --git a/src/host/observer/adapter.rs b/src/host/observer/adapter.rs new file mode 100644 index 0000000..2fe7c89 --- /dev/null +++ b/src/host/observer/adapter.rs @@ -0,0 +1,600 @@ +//! PipeWire I/O adapter for the pure registry observer. +//! +//! This module owns a read-only PipeWire main-loop thread, translates registry +//! callbacks into [`RegEvent`]s, and publishes the latest [`Projection`] for +//! consumers running outside the PipeWire thread. + +use super::classify::DeviceClaim; +use super::{LinkEndpoints, NodeObservation, Projection, RegEvent, RegistryModel}; +use crate::host::audio::parse_object_serial; +use crate::host::taint::snapshot::{ + ClientSnapshot, GlobalId, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial, +}; +use anyhow::{Context, Result}; +use pipewire::{self as pw, types::ObjectType}; +use std::cell::{Cell, RefCell}; +use std::collections::{BTreeMap, VecDeque}; +use std::rc::Rc; +use std::sync::{Arc, Mutex}; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; + +const READINESS_TIMEOUT_MILLIS: u64 = 2_000; +const TICK_INTERVAL: Duration = Duration::from_millis(250); + +/// Tokio-side access to the observer's most recent coherent projection. +pub struct RegistryObserverHandle { + latest: Arc>>, + shutdown_tx: pw::channel::Sender<()>, + thread: Option>, +} + +impl RegistryObserverHandle { + /// Spawn the read-only PipeWire registry observer. + pub fn spawn() -> Result { + let latest = Arc::new(Mutex::new(None)); + let latest_for_thread = Arc::clone(&latest); + let (shutdown_tx, shutdown_rx) = pw::channel::channel::<()>(); + let thread = std::thread::Builder::new() + .name("pixelpass-pw-observer".to_string()) + .spawn(move || { + if let Err(e) = run_observer(latest_for_thread, shutdown_rx) { + tracing::warn!("registry observer: libpipewire thread exited with error: {e:#}"); + } + }) + .context("failed to spawn libpipewire registry observer thread")?; + + Ok(Self { + latest, + shutdown_tx, + thread: Some(thread), + }) + } + + /// Return a clone of the latest projection, or `None` before the first + /// registry event has been applied. + pub fn latest(&self) -> Option { + self.latest + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone() + } +} + +impl Drop for RegistryObserverHandle { + fn drop(&mut self) { + let _ = self.shutdown_tx.send(()); + if let Some(thread) = self.thread.take() + && let Err(e) = thread.join() + { + tracing::warn!("registry observer: pw thread join failed: {e:?}"); + } + } +} + +struct BoundLink { + _proxy: pw::link::Link, + _listener: pw::link::LinkListener, +} + +#[derive(Default)] +struct LiveGlobal { + bound_link: Option, +} + +struct ObserverState { + model: RegistryModel, + latest: Arc>>, + last_candidate: Option, + live_globals: BTreeMap>, +} + +impl ObserverState { + fn new(latest: Arc>>) -> Self { + Self { + model: RegistryModel::new(0, READINESS_TIMEOUT_MILLIS), + latest, + last_candidate: None, + live_globals: BTreeMap::new(), + } + } + + fn apply(&mut self, event: RegEvent) { + self.model.apply(event); + + let candidate = self.model.pulse_pid_candidate(); + if candidate != self.last_candidate { + self.last_candidate = candidate; + if let Some(pid) = candidate { + let comm = std::fs::read_to_string(format!("/proc/{pid}/comm")) + .ok() + .map(|comm| comm.trim_end_matches(['\r', '\n']).to_string()); + self.model.apply(RegEvent::ProcCommProbed { pid, comm }); + } + } + + self.publish(); + } + + fn publish(&self) { + *self + .latest + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(self.model.project()); + } + + /// Record the global's id and apply its add event as one step, so the + /// bound-link FIFO stays provably lockstep with the model's own `live_ids` + /// index. Recording only on *applied* adds (never on unknown object types + /// or globals dropped for a missing serial) is what keeps the two id + /// queues the same length per id — otherwise a phantom slot ahead of a + /// bound Link would be popped on removal, leaking that Link's proxy. + fn add(&mut self, id: GlobalId, event: RegEvent) { + self.live_globals.entry(id).or_default().push_back(LiveGlobal::default()); + self.apply(event); + } + + fn attach_bound_link(&mut self, id: GlobalId, bound_link: BoundLink) { + let Some(global) = self.live_globals.get_mut(&id).and_then(VecDeque::back_mut) else { + tracing::warn!( + global_id = id.0, + "registry observer: link bind completed without a live global slot" + ); + return; + }; + global.bound_link = Some(bound_link); + } + + fn remove_global(&mut self, id: GlobalId) -> Option { + let (bound_link, empty) = { + let globals = self.live_globals.get_mut(&id)?; + let bound_link = globals.pop_front().and_then(|global| global.bound_link); + (bound_link, globals.is_empty()) + }; + if empty { + self.live_globals.remove(&id); + } + bound_link + } +} + +fn run_observer( + latest: Arc>>, + shutdown_rx: pw::channel::Receiver<()>, +) -> Result<()> { + let started_at = Instant::now(); + let main_loop = + pw::main_loop::MainLoopRc::new(None).context("pw main loop construction failed")?; + let context = + pw::context::ContextRc::new(&main_loop, None).context("pw context construction failed")?; + let core = context + .connect_rc(None) + .context("pw core connect failed (is the daemon running?)")?; + let registry = core.get_registry_rc().context("pw get_registry failed")?; + let state = Rc::new(RefCell::new(ObserverState::new(latest))); + + let main_loop_for_shutdown = main_loop.clone(); + let _shutdown_receiver = shutdown_rx.attach(main_loop.loop_(), move |()| { + main_loop_for_shutdown.quit(); + }); + + let pending_sync = Rc::new(Cell::new(None)); + let pending_sync_for_done = Rc::clone(&pending_sync); + let state_for_done = Rc::clone(&state); + let _core_listener = core + .add_listener_local() + .done(move |id, seq| { + if id == pw::core::PW_ID_CORE && pending_sync_for_done.get() == Some(seq) { + pending_sync_for_done.set(None); + state_for_done.borrow_mut().apply(RegEvent::ServerSynced); + } + }) + .error(|id, seq, res, message| { + tracing::warn!( + id, + seq, + result = res, + %message, + "registry observer: PipeWire core error" + ); + }) + .register(); + + let registry_weak = registry.downgrade(); + let state_for_global = Rc::clone(&state); + let state_for_remove = Rc::clone(&state); + let _registry_listener = registry + .add_listener_local() + .global(move |obj| { + let id = GlobalId(obj.id); + + match obj.type_ { + ObjectType::Node => { + let Some(props) = obj.props.as_ref() else { + tracing::warn!( + node_id = obj.id, + "registry observer: Node has no properties; dropping" + ); + return; + }; + let Some(serial) = + parse_serial(obj.id, "Node", props.get("object.serial")) + else { + return; + }; + let node_props = NodeProps { + peerspeak_owned: truthy(props.get("peerspeak.owned")), + pulse_module_id: props + .get("pulse.module.id") + .and_then(|value| value.parse::().ok()), + link_group: props.get("node.link-group").map(str::to_owned), + client_id: props + .get("client.id") + .and_then(|value| value.parse::().ok()) + .map(GlobalId), + process_id: props + .get("application.process.id") + .and_then(|value| value.parse::().ok()), + passthrough: truthy(props.get("node.passthrough")), + session_device: false, + }; + let observation = NodeObservation { + serial, + id, + name: props.get("node.name").map(str::to_owned), + role: MediaRole::parse(props.get("media.class")), + props: node_props, + device_claim: DeviceClaim { + device_id: props + .get("device.id") + .and_then(|value| value.parse::().ok()) + .map(GlobalId), + device_api: props.get("device.api").map(str::to_owned), + factory_name: props.get("factory.name").map(str::to_owned), + }, + }; + state_for_global + .borrow_mut() + .add(id, RegEvent::NodeAdded(observation)); + } + ObjectType::Port => { + let Some(props) = obj.props.as_ref() else { + tracing::warn!( + port_id = obj.id, + "registry observer: Port has no properties; dropping" + ); + return; + }; + let Some(serial) = + parse_serial(obj.id, "Port", props.get("object.serial")) + else { + return; + }; + let Some(node) = props + .get("node.id") + .and_then(|value| value.parse::().ok()) + .map(GlobalId) + else { + tracing::warn!( + port_id = obj.id, + node_id = props.get("node.id").unwrap_or(""), + "registry observer: Port has no usable node.id; dropping" + ); + return; + }; + let direction = match props.get("port.direction") { + Some("in") => PortDirection::In, + Some("out") => PortDirection::Out, + direction => { + tracing::warn!( + port_id = obj.id, + direction = direction.unwrap_or(""), + "registry observer: Port has no usable direction; dropping" + ); + return; + } + }; + state_for_global + .borrow_mut() + .add(id, RegEvent::PortAdded(PortSnapshot { + serial, + id, + node, + direction, + exclusive: truthy(props.get("port.exclusive")), + monitor: truthy(props.get("port.monitor")), + })); + } + ObjectType::Client => { + let Some(props) = obj.props.as_ref() else { + tracing::warn!( + client_id = obj.id, + "registry observer: Client has no properties; dropping" + ); + return; + }; + let Some(serial) = + parse_serial(obj.id, "Client", props.get("object.serial")) + else { + return; + }; + state_for_global + .borrow_mut() + .add(id, RegEvent::ClientAdded(ClientSnapshot { + serial, + id, + sec_pid: props + .get("pipewire.sec.pid") + .and_then(|value| value.parse::().ok()), + })); + } + ObjectType::Device => { + state_for_global + .borrow_mut() + .add(id, RegEvent::DeviceAdded { id }); + } + ObjectType::Link => { + let Some(props) = obj.props.as_ref() else { + tracing::warn!( + link_id = obj.id, + "registry observer: Link has no properties; dropping" + ); + return; + }; + let Some(serial) = + parse_serial(obj.id, "Link", props.get("object.serial")) + else { + return; + }; + let endpoints = link_endpoints_from_props(props); + state_for_global.borrow_mut().add( + id, + RegEvent::LinkAdded { + serial, + id, + endpoints, + }, + ); + if endpoints.is_some() { + return; + } + + let Some(registry) = registry_weak.upgrade() else { + return; + }; + let link: pw::link::Link = match registry.bind(obj) { + Ok(link) => link, + Err(e) => { + tracing::warn!( + link_id = obj.id, + "registry observer: failed to bind Link for endpoints: {e}" + ); + return; + } + }; + let resolved = Rc::new(Cell::new(false)); + let resolved_for_info = Rc::clone(&resolved); + let state_for_info = Rc::downgrade(&state_for_global); + let listener = link + .add_listener_local() + .info(move |info| { + if resolved_for_info.replace(true) { + return; + } + let endpoints = LinkEndpoints { + output_node: GlobalId(info.output_node_id()), + input_node: GlobalId(info.input_node_id()), + output_port: optional_global_id(info.output_port_id()), + input_port: optional_global_id(info.input_port_id()), + }; + if let Some(state) = state_for_info.upgrade() { + state.borrow_mut().apply(RegEvent::LinkEndpointsResolved { + serial, + endpoints, + }); + } + }) + .register(); + state_for_global + .borrow_mut() + .attach_bound_link(id, BoundLink { + _proxy: link, + _listener: listener, + }); + } + _ => {} + } + }) + .global_remove(move |id| { + let id = GlobalId(id); + let bound_link = state_for_remove.borrow_mut().remove_global(id); + state_for_remove + .borrow_mut() + .apply(RegEvent::Removed { id }); + drop(bound_link); + }) + .register(); + + pending_sync.set(Some( + core.sync(0) + .context("registry observer: initial core.sync failed")?, + )); + + let state_for_tick = Rc::clone(&state); + let timer = main_loop.loop_().add_timer(move |_| { + let now = u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX); + state_for_tick.borrow_mut().apply(RegEvent::Tick { now }); + }); + timer + .update_timer(Some(TICK_INTERVAL), Some(TICK_INTERVAL)) + .into_result() + .context("registry observer: failed to arm readiness timer")?; + + tracing::info!("registry observer: pw thread running"); + main_loop.run(); + tracing::info!("registry observer: pw thread exiting"); + Ok(()) +} + +fn parse_serial(id: u32, kind: &str, raw: Option<&str>) -> Option { + match raw.and_then(parse_object_serial) { + Some(serial) => Some(Serial(serial)), + None => { + tracing::warn!( + global_id = id, + object_type = kind, + serial = raw.unwrap_or(""), + "registry observer: global has no usable object.serial; dropping" + ); + None + } + } +} + +fn truthy(value: Option<&str>) -> bool { + value.is_some_and(|value| value != "false" && value != "0") +} + +fn link_endpoints_from_props(props: &pw::spa::utils::dict::DictRef) -> Option { + let output_node = props.get("link.output.node")?.parse::().ok()?; + let input_node = props.get("link.input.node")?.parse::().ok()?; + Some(LinkEndpoints { + output_node: GlobalId(output_node), + input_node: GlobalId(input_node), + output_port: props + .get("link.output.port") + .and_then(|value| value.parse::().ok()) + .map(GlobalId), + input_port: props + .get("link.input.port") + .and_then(|value| value.parse::().ok()) + .map(GlobalId), + }) +} + +fn optional_global_id(id: u32) -> Option { + (id != pw::constants::ID_ANY).then_some(GlobalId(id)) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::process::Command; + + struct PactlModule { + id: Option, + } + + impl PactlModule { + fn load(name: &str, args: &[String]) -> Self { + let output = Command::new("pactl") + .arg("load-module") + .arg(name) + .args(args) + .output() + .expect("pactl must be installed for the live observer test"); + assert!( + output.status.success(), + "pactl load-module {name} failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + ); + let id = String::from_utf8(output.stdout) + .expect("pactl module id must be UTF-8") + .trim() + .parse::() + .expect("pactl module id must be a u32"); + Self { id: Some(id) } + } + + fn unload(mut self) { + let id = self.id.take().expect("module must still be loaded"); + let output = Command::new("pactl") + .arg("unload-module") + .arg(id.to_string()) + .output() + .expect("pactl must be installed for the live observer test"); + assert!( + output.status.success(), + "pactl unload-module {id} failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + ); + } + } + + impl Drop for PactlModule { + fn drop(&mut self) { + if let Some(id) = self.id.take() { + let _ = Command::new("pactl") + .arg("unload-module") + .arg(id.to_string()) + .output(); + } + } + } + + fn wait_for( + observer: &RegistryObserverHandle, + predicate: impl Fn(&Projection) -> bool, + ) -> Projection { + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline { + if let Some(projection) = observer.latest() + && predicate(&projection) + { + return projection; + } + std::thread::sleep(Duration::from_millis(50)); + } + panic!("timed out waiting for the registry projection"); + } + + fn has_node(projection: &Projection, name: &str) -> bool { + projection + .snapshot + .nodes() + .any(|node| node.name.as_deref() == Some(name)) + } + + #[test] + #[ignore = "needs live pipewire"] + fn live_topology_diff_tracks_null_sink_and_loopback() { + pw::init(); + let observer = RegistryObserverHandle::spawn().expect("observer thread must spawn"); + let baseline = wait_for(&observer, |projection| projection.graph_ready); + let baseline_links = baseline.snapshot.links().count(); + + let unique = format!("pixelpass_observer_test_{}", std::process::id()); + let capture_name = format!("{unique}_capture"); + let playback_name = format!("{unique}_playback"); + let null_sink = PactlModule::load( + "module-null-sink", + &[format!("sink_name={unique}")], + ); + let with_sink = wait_for(&observer, |projection| has_node(projection, &unique)); + let sink_links = with_sink.snapshot.links().count(); + + let loopback = PactlModule::load( + "module-loopback", + &[ + format!("source={unique}.monitor"), + format!("sink={unique}"), + format!("source_output_properties=node.name={capture_name}"), + format!("sink_input_properties=node.name={playback_name}"), + ], + ); + wait_for(&observer, |projection| { + has_node(projection, &capture_name) + && has_node(projection, &playback_name) + && projection.snapshot.links().count() > sink_links + }); + + loopback.unload(); + null_sink.unload(); + wait_for(&observer, |projection| { + !has_node(projection, &unique) + && !has_node(projection, &capture_name) + && !has_node(projection, &playback_name) + && projection.snapshot.links().count() <= baseline_links + }); + } +} diff --git a/src/host/observer/mod.rs b/src/host/observer/mod.rs index cb0abbe..549accf 100644 --- a/src/host/observer/mod.rs +++ b/src/host/observer/mod.rs @@ -26,6 +26,7 @@ #![allow(dead_code)] // Wired by the phase-3 adapter (Codex's half) and consumed by later phases. +pub mod adapter; pub mod classify; pub mod pulse_pid; From 557c1030a725cfda1adf9507c40fb53e6bad6f6b Mon Sep 17 00:00:00 2001 From: Mollusk Date: Fri, 24 Jul 2026 13:28:31 -0400 Subject: [PATCH 3/4] host/observer: address Codex phase-3 review (2 P1 + P3s) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cross-review round: Codex adversarially reviewed my pure core, found two merge-blocking P1s and several P3s. Triaged each for reachability; fixes below, each mutation-verified (revert killed by its intended test). P1 finding 1 — graph_ready was sticky-once-Complete, so a Link added post-enumeration whose endpoints are still binding (an INVISIBLE edge, absent from the snapshot) left graph_ready=true and a candidate could be reported eligible over unseen tainted ancestry. graph_ready is now dynamic: Complete AND no outstanding obligations. Readiness::Complete stays sticky as the epoch marker. New regression test + flipped the old sticky-churn test. P1 finding 2 — snd_aloop presents with an allowlisted ALSA factory and device.api=alsa exactly like a real card but forwards audio through a kernel hop the Link graph cannot see; it was classified session_device=true, dropping its owner keys + backstop (leak). Added alsa.driver_name to DeviceClaim and a NON_TERMINAL_ALSA_DRIVERS denylist under the factory allowlist; adapter now populates it. Negative fixture added. P3 finding 5 — the BlueZ allowlist entries (api.bluez5.pcm.*) were invented; removed them (real names are api.bluez5.media.*). A BT sink now over-excludes (safe) pending a measured fixture. P3 finding 6 — strengthened the timeout test to assert TimedOut stays sticky through later DeviceAdded/sync/tick. Findings 3 (dropped-link unrepresented) and 4 (missed-removal generation ambiguity) documented as accepted low-reachability limitations (links carry object.serial — confirmed by the live gate; registry does not drop removals). Codex confirmed the pulse-PID matrix fails safe and the adapter add() FIFO is lockstep. 120 unit + live gate row 6 green, clippy + fmt clean. Co-Authored-By: Claude Opus 4.8 --- src/host/observer/adapter.rs | 61 +++++++++--------- src/host/observer/classify.rs | 45 +++++++++++--- src/host/observer/mod.rs | 48 ++++++++++++++- src/host/observer/tests.rs | 113 ++++++++++++++++++++++++++++------ 4 files changed, 209 insertions(+), 58 deletions(-) diff --git a/src/host/observer/adapter.rs b/src/host/observer/adapter.rs index 2fe7c89..de06f9e 100644 --- a/src/host/observer/adapter.rs +++ b/src/host/observer/adapter.rs @@ -39,7 +39,9 @@ impl RegistryObserverHandle { .name("pixelpass-pw-observer".to_string()) .spawn(move || { if let Err(e) = run_observer(latest_for_thread, shutdown_rx) { - tracing::warn!("registry observer: libpipewire thread exited with error: {e:#}"); + tracing::warn!( + "registry observer: libpipewire thread exited with error: {e:#}" + ); } }) .context("failed to spawn libpipewire registry observer thread")?; @@ -130,7 +132,10 @@ impl ObserverState { /// queues the same length per id — otherwise a phantom slot ahead of a /// bound Link would be popped on removal, leaking that Link's proxy. fn add(&mut self, id: GlobalId, event: RegEvent) { - self.live_globals.entry(id).or_default().push_back(LiveGlobal::default()); + self.live_globals + .entry(id) + .or_default() + .push_back(LiveGlobal::default()); self.apply(event); } @@ -217,8 +222,7 @@ fn run_observer( ); return; }; - let Some(serial) = - parse_serial(obj.id, "Node", props.get("object.serial")) + let Some(serial) = parse_serial(obj.id, "Node", props.get("object.serial")) else { return; }; @@ -251,6 +255,7 @@ fn run_observer( .map(GlobalId), device_api: props.get("device.api").map(str::to_owned), factory_name: props.get("factory.name").map(str::to_owned), + alsa_driver_name: props.get("alsa.driver_name").map(str::to_owned), }, }; state_for_global @@ -265,8 +270,7 @@ fn run_observer( ); return; }; - let Some(serial) = - parse_serial(obj.id, "Port", props.get("object.serial")) + let Some(serial) = parse_serial(obj.id, "Port", props.get("object.serial")) else { return; }; @@ -294,16 +298,17 @@ fn run_observer( return; } }; - state_for_global - .borrow_mut() - .add(id, RegEvent::PortAdded(PortSnapshot { + state_for_global.borrow_mut().add( + id, + RegEvent::PortAdded(PortSnapshot { serial, id, node, direction, exclusive: truthy(props.get("port.exclusive")), monitor: truthy(props.get("port.monitor")), - })); + }), + ); } ObjectType::Client => { let Some(props) = obj.props.as_ref() else { @@ -313,20 +318,20 @@ fn run_observer( ); return; }; - let Some(serial) = - parse_serial(obj.id, "Client", props.get("object.serial")) + let Some(serial) = parse_serial(obj.id, "Client", props.get("object.serial")) else { return; }; - state_for_global - .borrow_mut() - .add(id, RegEvent::ClientAdded(ClientSnapshot { + state_for_global.borrow_mut().add( + id, + RegEvent::ClientAdded(ClientSnapshot { serial, id, sec_pid: props .get("pipewire.sec.pid") .and_then(|value| value.parse::().ok()), - })); + }), + ); } ObjectType::Device => { state_for_global @@ -341,8 +346,7 @@ fn run_observer( ); return; }; - let Some(serial) = - parse_serial(obj.id, "Link", props.get("object.serial")) + let Some(serial) = parse_serial(obj.id, "Link", props.get("object.serial")) else { return; }; @@ -388,19 +392,19 @@ fn run_observer( input_port: optional_global_id(info.input_port_id()), }; if let Some(state) = state_for_info.upgrade() { - state.borrow_mut().apply(RegEvent::LinkEndpointsResolved { - serial, - endpoints, - }); + state + .borrow_mut() + .apply(RegEvent::LinkEndpointsResolved { serial, endpoints }); } }) .register(); - state_for_global - .borrow_mut() - .attach_bound_link(id, BoundLink { + state_for_global.borrow_mut().attach_bound_link( + id, + BoundLink { _proxy: link, _listener: listener, - }); + }, + ); } _ => {} } @@ -566,10 +570,7 @@ mod tests { let unique = format!("pixelpass_observer_test_{}", std::process::id()); let capture_name = format!("{unique}_capture"); let playback_name = format!("{unique}_playback"); - let null_sink = PactlModule::load( - "module-null-sink", - &[format!("sink_name={unique}")], - ); + let null_sink = PactlModule::load("module-null-sink", &[format!("sink_name={unique}")]); let with_sink = wait_for(&observer, |projection| has_node(projection, &unique)); let sink_links = with_sink.snapshot.links().count(); diff --git a/src/host/observer/classify.rs b/src/host/observer/classify.rs index 3639c27..51c8f08 100644 --- a/src/host/observer/classify.rs +++ b/src/host/observer/classify.rs @@ -39,15 +39,33 @@ use crate::host::taint::snapshot::GlobalId; /// `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. +/// +/// ⚠️ **ALSA only, and only these two, because they are the only factories +/// measured on the target box.** BlueZ was previously listed here as +/// `api.bluez5.pcm.{sink,source}` — those are invented; the real BlueZ +/// terminals are `api.bluez5.media.{sink,source}` with profile aliases +/// (Codex phase-3 review, finding 5). Rather than allowlist an unmeasured +/// guess, BlueZ is left off entirely: a real Bluetooth sink then keeps its +/// owner keys (over-exclusion — safe). Add BlueZ back only with a *measured* +/// factory name and a fixture. 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", ]; +/// ALSA drivers that expose a hardware-PCM `factory.name` but are **not** +/// passive terminals — audio written in reappears on their capture side +/// through a path the PipeWire Link graph cannot see, so classifying them +/// `session_device` (which drops owner keys and the fail-closed backstop) +/// would let tainted audio loop back untainted (Codex phase-3 review, +/// finding 2). `factory.name` alone cannot distinguish these from a real +/// card — `snd_aloop` presents as `api.alsa.pcm.{sink,source}` exactly like +/// `snd_hda_intel` — so this is a necessary denylist layered under the +/// factory allowlist, keyed on the one property that does distinguish them +/// (`alsa.driver_name`, measured present on real ALSA nodes). +const NON_TERMINAL_ALSA_DRIVERS: &[&str] = &["snd_aloop"]; + /// 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 @@ -67,6 +85,12 @@ pub struct DeviceClaim { /// `factory.name` — the discriminator. Only an allowlisted hardware-PCM /// factory earns `session_device`. pub factory_name: Option, + /// `alsa.driver_name` — the kernel driver behind an ALSA node (e.g. + /// `snd_hda_intel`, `snd_usb_audio`, `snd_aloop`). Needed because the + /// factory allowlist cannot tell a real card from a loopback driver that + /// shares the same factory; a driver on [`NON_TERMINAL_ALSA_DRIVERS`] + /// forfeits `session_device`. Absent on non-ALSA backends. + pub alsa_driver_name: Option, } /// The outcome of classifying one node's device claim. @@ -104,11 +128,16 @@ pub fn classify(claim: &DeviceClaim, device_resolved: bool) -> Classification { // 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)); + let on_factory_allowlist = claim + .factory_name + .as_deref() + .is_some_and(|f| HARDWARE_PCM_FACTORIES.contains(&f)); + let non_terminal_driver = claim + .alsa_driver_name + .as_deref() + .is_some_and(|d| NON_TERMINAL_ALSA_DRIVERS.contains(&d)); + let is_hardware_pcm = + claim.device_api.is_some() && on_factory_allowlist && !non_terminal_driver; if is_hardware_pcm { Classification::SessionDevice } else { diff --git a/src/host/observer/mod.rs b/src/host/observer/mod.rs index 549accf..1f3e367 100644 --- a/src/host/observer/mod.rs +++ b/src/host/observer/mod.rs @@ -23,6 +23,25 @@ //! whose Device we have not observed is held out of the snapshot entirely //! rather than admitted with a provisional `session_device` (see //! [`classify`]). +//! +//! **Two accepted limitations (Codex phase-3 review, findings 3 and 4), both +//! low-reachability, owed to a later hardening round:** +//! +//! - *A Link dropped for a missing `object.serial`/props is unrepresented.* +//! The adapter drops such a global before it reaches [`RegistryModel`], so +//! readiness can reach `Complete` while permanently omitting that Link — an +//! invisible edge that could hide tainted ancestry. **Not reachable in +//! practice:** every real Link global carries `object.serial` (confirmed by +//! the live gate, which only counts links the strict parser admits). A full +//! fix needs a pure "required-observation-failed" token that holds readiness +//! false; deferred rather than built for a case that does not occur. +//! - *Removal generation ordering assumes no removal is silently lost.* On a +//! recycled id with two live claimants, [`Self::on_removed`] retires the +//! oldest generation first; if the *first* generation's removal was never +//! delivered, a later removal is misattributed. PipeWire's registry does not +//! silently drop `global_remove`, so this needs callback loss to trigger. +//! The snapshot treats the two-claimant window as [`IdLookup::Ambiguous`] +//! (fail closed) meanwhile. #![allow(dead_code)] // Wired by the phase-3 adapter (Codex's half) and consumed by later phases. @@ -131,7 +150,12 @@ enum Slot { pub enum Readiness { /// The initial enumeration is still in flight. Waiting, - /// Server synced and every obligation resolved. `graph_ready` is true. + /// The initial enumeration finished at least once (server synced with no + /// obligations then outstanding). **Sticky** — later per-object + /// withholding does not revert it. Note this is *not* the same as + /// [`RegistryModel::graph_ready`], which additionally requires no *current* + /// obligation (Codex finding 1); `Complete` only records that the epoch + /// was reached. Complete, /// The bounded deadline passed with obligations outstanding. /// `graph_ready` stays false — fail closed. @@ -207,8 +231,28 @@ impl RegistryModel { self.readiness } + /// Whether the graph is trustworthy enough to make eligibility and sticky + /// **retirement** decisions right now. + /// + /// This is **dynamic**, not the sticky [`Readiness::Complete`] flag: it is + /// true only when the initial enumeration has completed **and** there are + /// no current obligations outstanding (a node withheld on an unresolved + /// Device, or a Link still being bound). The distinction is the fix for + /// Codex phase-3 review finding 1: a Link whose endpoints are still + /// resolving is an **invisible edge** — it is absent from the snapshot, + /// not merely dangling — so a decision made while one exists can miss real + /// tainted ancestry and wrongly report a candidate eligible. Unresolved + /// ancestry ⇒ fail closed is the governing invariant (v3.4 §6.1), and an + /// unresolved Link is unresolved ancestry, so `graph_ready` must drop back + /// to false whenever one is pending — even after the initial epoch. + /// + /// [`Readiness::Complete`] stays sticky (it records that the initial + /// enumeration happened, for logging and to distinguish "not started" from + /// "momentarily churning"); `graph_ready` layers the dynamic obligation + /// check on top. Downstream (phase 6) may debounce the brief blips a + /// normal Link bind causes; the observer's job is to report the truth. pub fn graph_ready(&self) -> bool { - matches!(self.readiness, Readiness::Complete) + matches!(self.readiness, Readiness::Complete) && !self.obligations_outstanding() } /// The pulse-PID candidate the adapter should be probing (`None` = no diff --git a/src/host/observer/tests.rs b/src/host/observer/tests.rs index a5eae36..014e9de 100644 --- a/src/host/observer/tests.rs +++ b/src/host/observer/tests.rs @@ -37,6 +37,7 @@ fn hw_claim(device_id: u32, api: &str, factory: &str) -> DeviceClaim { device_id: Some(gid(device_id)), device_api: Some(api.to_string()), factory_name: Some(factory.to_string()), + alsa_driver_name: Some("snd_hda_intel".to_string()), } } @@ -114,25 +115,53 @@ fn classify_unresolved_device_withholds() { #[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" - }; + // Only the measured ALSA factories are allowlisted (finding 5: the BlueZ + // entries were invented and were removed). + for factory in ["api.alsa.pcm.sink", "api.alsa.pcm.source"] { assert_eq!( - classify(&hw_claim(7, api, factory), true), + classify(&hw_claim(7, "alsa", factory), true), Classification::SessionDevice, "factory {factory} should be a session device" ); } } +#[test] +fn classify_invented_bluez_factories_are_not_session_devices() { + // Finding 5: `api.bluez5.pcm.*` is not a real factory name; whatever it is, + // it is not on the measured allowlist, so it fails closed to false + // (over-exclusion, safe) rather than being trusted. + for factory in ["api.bluez5.pcm.sink", "api.bluez5.pcm.source"] { + let claim = DeviceClaim { + device_id: Some(gid(7)), + device_api: Some("bluez5".to_string()), + factory_name: Some(factory.to_string()), + alsa_driver_name: None, + }; + assert_eq!(classify(&claim, true), Classification::NotSessionDevice); + } +} + +#[test] +fn classify_snd_aloop_is_not_a_session_device() { + // Finding 2: an ALSA loopback presents with an allowlisted factory and + // device.api=alsa exactly like a real card, but forwards audio through a + // kernel hop the Link graph cannot see. It must NOT earn session_device. + for factory in ["api.alsa.pcm.sink", "api.alsa.pcm.source"] { + let claim = DeviceClaim { + device_id: Some(gid(7)), + device_api: Some("alsa".to_string()), + factory_name: Some(factory.to_string()), + alsa_driver_name: Some("snd_aloop".to_string()), + }; + assert_eq!( + classify(&claim, true), + Classification::NotSessionDevice, + "snd_aloop {factory} must fail closed" + ); + } +} + #[test] fn classify_resolved_but_not_hardware_pcm_fails_closed() { // A null sink, a loopback, and an unknown factory are all forwarders, not @@ -154,6 +183,7 @@ fn classify_missing_device_api_fails_closed() { device_id: Some(gid(7)), device_api: None, factory_name: Some("api.alsa.pcm.sink".to_string()), + alsa_driver_name: Some("snd_hda_intel".to_string()), }; assert_eq!(classify(&claim, true), Classification::NotSessionDevice); } @@ -488,6 +518,15 @@ fn model_readiness_times_out_fail_closed() { m.apply(RegEvent::Tick { now: 5000 }); assert_eq!(m.readiness(), Readiness::TimedOut); assert!(!m.graph_ready(), "timeout fails closed"); + + // Finding 6: TimedOut must be sticky. Resolving the obligation, syncing + // again, and ticking further must NOT flip it to Complete — a timed-out + // observer stays fail-closed for its lifetime. + m.apply(RegEvent::DeviceAdded { id: gid(42) }); + m.apply(RegEvent::ServerSynced); + m.apply(RegEvent::Tick { now: 6000 }); + assert_eq!(m.readiness(), Readiness::TimedOut, "timeout is sticky"); + assert!(!m.graph_ready()); } #[test] @@ -504,23 +543,61 @@ fn model_tick_before_deadline_does_not_time_out() { } #[test] -fn model_complete_is_sticky_across_later_churn() { +fn model_complete_epoch_is_sticky_but_graph_ready_is_dynamic() { 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). + assert!(m.graph_ready()); + // A node withheld AFTER completion does not revert the sticky EPOCH... 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. + assert_eq!(m.readiness(), Readiness::Complete, "epoch stays sticky"); + // ...but graph_ready DOES drop while the obligation is outstanding + // (Codex finding 1: unresolved ancestry ⇒ fail closed, even post-epoch). + assert!( + !m.graph_ready(), + "an outstanding obligation makes decisions unsafe" + ); + // A late timeout Tick is inert once Complete. m.apply(RegEvent::Tick { now: 100_000 }); assert_eq!(m.readiness(), Readiness::Complete); + // Resolving the obligation restores graph_ready. + m.apply(RegEvent::DeviceAdded { id: gid(42) }); + assert!(m.graph_ready()); +} + +#[test] +fn model_pending_link_drops_graph_ready_after_completion() { + // Codex finding 1, the leak that mattered: a real Link added post-epoch + // whose endpoints are still binding is an INVISIBLE edge (absent from the + // snapshot, not dangling). graph_ready must go false until it resolves, + // or a candidate can be reported eligible while tainted ancestry it cannot + // see already carries call audio. + let mut m = model(); + m.apply(RegEvent::ServerSynced); + assert!(m.graph_ready()); + + m.apply(RegEvent::LinkAdded { + serial: ser(300), + id: gid(90), + endpoints: None, + }); + assert!(!m.graph_ready(), "an unresolved link must gate decisions"); + // The snapshot genuinely omits it, which is exactly why graph_ready must + // compensate. + assert_eq!(m.project().snapshot.links().count(), 0); + assert!(!m.project().graph_ready); + + m.apply(RegEvent::LinkEndpointsResolved { + serial: ser(300), + endpoints: endpoints(50, 55), + }); + assert!(m.graph_ready(), "resolved ⇒ decisions safe again"); + assert_eq!(m.project().snapshot.links().count(), 1); } #[test] From f90bee63f75ea24d6d5e525693e9660a99a8433e Mon Sep 17 00:00:00 2001 From: Mollusk Date: Fri, 24 Jul 2026 13:40:47 -0400 Subject: [PATCH 4/4] host/observer: close the snd_aloop absent-driver leak (Codex re-review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex's re-review of the phase-3 fixes confirmed finding 1/5/6 closed but found the finding-2 fix incomplete: the denylist only rejected a *present* snd_aloop driver, so an snd_aloop node whose alsa.driver_name was not copied onto the node still classified session_device=true — the original leak. The absence is reachable: PipeWire >=1.2.6 stopped overwriting node props with card props, and WirePlumber only began copying alsa.* onto nodes in 0.5.13. Fix: session_device now requires a PRESENT, non-denied ALSA driver; a missing alsa.driver_name fails closed to NotSessionDevice (a real card without the prop is over-excluded — safe; recovering it needs reading the driver from the backing Device global, owed to a later round). Mutation-verified: reverting to fail-open on absence is killed by classify_alsa_without_driver_name_fails_closed. Also: corrected the finding-3 limitation doc to cite PipeWire's object.serial identity contract rather than overclaiming the live gate proves it (Codex P3, non-blocking). 121 unit + live gate row 6 green, clippy clean, observer files fmt-clean. Co-Authored-By: Claude Opus 4.8 --- src/host/observer/classify.rs | 31 ++++++++++++++++++++++--------- src/host/observer/mod.rs | 11 +++++++---- src/host/observer/tests.rs | 20 ++++++++++++++++++++ 3 files changed, 49 insertions(+), 13 deletions(-) diff --git a/src/host/observer/classify.rs b/src/host/observer/classify.rs index 51c8f08..37080f6 100644 --- a/src/host/observer/classify.rs +++ b/src/host/observer/classify.rs @@ -61,9 +61,10 @@ const HARDWARE_PCM_FACTORIES: &[&str] = &[ /// would let tainted audio loop back untainted (Codex phase-3 review, /// finding 2). `factory.name` alone cannot distinguish these from a real /// card — `snd_aloop` presents as `api.alsa.pcm.{sink,source}` exactly like -/// `snd_hda_intel` — so this is a necessary denylist layered under the -/// factory allowlist, keyed on the one property that does distinguish them -/// (`alsa.driver_name`, measured present on real ALSA nodes). +/// `snd_hda_intel` — so a real ALSA terminal must present an `alsa.driver_name` +/// that is **present and not on this denylist**; a missing driver fails closed +/// (see [`classify`]). `snd_dummy` is intentionally absent: it is virtual but +/// does not couple playback to capture, so it is not a loopback hazard. const NON_TERMINAL_ALSA_DRIVERS: &[&str] = &["snd_aloop"]; /// The three node properties the classifier reads, exactly as the adapter @@ -88,8 +89,11 @@ pub struct DeviceClaim { /// `alsa.driver_name` — the kernel driver behind an ALSA node (e.g. /// `snd_hda_intel`, `snd_usb_audio`, `snd_aloop`). Needed because the /// factory allowlist cannot tell a real card from a loopback driver that - /// shares the same factory; a driver on [`NON_TERMINAL_ALSA_DRIVERS`] - /// forfeits `session_device`. Absent on non-ALSA backends. + /// shares the same factory. `session_device` requires this to be + /// **present and not** on [`NON_TERMINAL_ALSA_DRIVERS`]; a driver on the + /// denylist, or an absent value, both fail closed (see [`classify`]). + /// May be absent on non-ALSA backends or on version pairings that do not + /// copy `alsa.*` onto the node. pub alsa_driver_name: Option, } @@ -132,12 +136,21 @@ pub fn classify(claim: &DeviceClaim, device_resolved: bool) -> Classification { .factory_name .as_deref() .is_some_and(|f| HARDWARE_PCM_FACTORIES.contains(&f)); - let non_terminal_driver = claim + // A **present, non-denied** ALSA driver is required — absence fails closed + // (Codex phase-3 re-review). `alsa.driver_name` is not copied onto the + // node on every PipeWire/WirePlumber version pairing (PipeWire ≥1.2.6 + // stopped overwriting node props with card props; WirePlumber only began + // copying `alsa.*` onto nodes in 0.5.13), so a *missing* value must not be + // read as "not a loopback" — that is exactly the hole an `snd_aloop` node + // without the property would slip through. A real card whose node lacks + // the driver is instead over-excluded (keeps its owner keys — safe); + // recovering `session_device` for it needs reading the driver from the + // backing Device global, which is owed to a later round. + let driver_ok = claim .alsa_driver_name .as_deref() - .is_some_and(|d| NON_TERMINAL_ALSA_DRIVERS.contains(&d)); - let is_hardware_pcm = - claim.device_api.is_some() && on_factory_allowlist && !non_terminal_driver; + .is_some_and(|d| !NON_TERMINAL_ALSA_DRIVERS.contains(&d)); + let is_hardware_pcm = claim.device_api.is_some() && on_factory_allowlist && driver_ok; if is_hardware_pcm { Classification::SessionDevice } else { diff --git a/src/host/observer/mod.rs b/src/host/observer/mod.rs index 1f3e367..ced862f 100644 --- a/src/host/observer/mod.rs +++ b/src/host/observer/mod.rs @@ -31,10 +31,13 @@ //! The adapter drops such a global before it reaches [`RegistryModel`], so //! readiness can reach `Complete` while permanently omitting that Link — an //! invisible edge that could hide tainted ancestry. **Not reachable in -//! practice:** every real Link global carries `object.serial` (confirmed by -//! the live gate, which only counts links the strict parser admits). A full -//! fix needs a pure "required-observation-failed" token that holds readiness -//! false; deferred rather than built for a case that does not occur. +//! practice:** PipeWire's native protocol defines `object.serial` as the +//! unique identity every global carries, so a Link without one requires a +//! protocol/server failure, not ordinary churn. (The live gate is +//! consistent with this but does not *prove* it — it only counts Links the +//! strict parser already admitted.) A full fix needs a pure +//! "required-observation-failed" token that holds readiness false; deferred +//! rather than built for a case that does not occur. //! - *Removal generation ordering assumes no removal is silently lost.* On a //! recycled id with two live claimants, [`Self::on_removed`] retires the //! oldest generation first; if the *first* generation's removal was never diff --git a/src/host/observer/tests.rs b/src/host/observer/tests.rs index 014e9de..61db94d 100644 --- a/src/host/observer/tests.rs +++ b/src/host/observer/tests.rs @@ -142,6 +142,26 @@ fn classify_invented_bluez_factories_are_not_session_devices() { } } +#[test] +fn classify_alsa_without_driver_name_fails_closed() { + // Codex re-review: a missing `alsa.driver_name` must NOT grant + // session_device — an snd_aloop node whose driver prop was not copied onto + // the node would otherwise slip through. Absence fails closed. + for factory in ["api.alsa.pcm.sink", "api.alsa.pcm.source"] { + let claim = DeviceClaim { + device_id: Some(gid(7)), + device_api: Some("alsa".to_string()), + factory_name: Some(factory.to_string()), + alsa_driver_name: None, + }; + assert_eq!( + classify(&claim, true), + Classification::NotSessionDevice, + "absent driver on {factory} must fail closed" + ); + } +} + #[test] fn classify_snd_aloop_is_not_a_session_device() { // Finding 2: an ALSA loopback presents with an allowlisted factory and