mollusk/pixelpass
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge phase 3: registry observer (pure core + I/O adapter)

Browse Source 
Split-seam mutual-review build: pure reducer/classifiers (Claude) + libpipewire
adapter (Codex), each reviewed by the other. Two review rounds closed 3 P1s
(dynamic graph_ready over invisible edges; snd_aloop absent-driver fail-closed;
FIFO lockstep). Exit gate incl. live topology-diff row passes on the host.

Additive/read-only — does not yet replace the audio.rs router (integration
phase). DAG: 0a -> 2 -> 3 done; next is Phase 4 (AEC validation state machine).

Co-Authored-By: Codex (gpt-5.6-sol) <codex@openai.com>
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
mollusk
2026-07-24 14:01:32 -04:00
co-authored by Codex Claude Opus 4.8
parent ab597c332d f90bee63f7
commit 8973e5dc19
7 changed files with 2082 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/host/audio.rs
+1 -1
View File
@@ -615,7 +615,7 @@ fn run_router(
/// (empty, signed, whitespace-padded, non-numeric, overflowing) is a /// (empty, signed, whitespace-padded, non-numeric, overflowing) is a
/// property we do not understand and must not guess at. Leading zeroes /// property we do not understand and must not guess at. Leading zeroes
/// are accepted — they are unambiguous and parse to the same value. /// are accepted — they are unambiguous and parse to the same value.
fn parse_object_serial(raw: &str) -> Option<u64> { pub(crate) fn parse_object_serial(raw: &str) -> Option<u64> {
if raw.is_empty() || !raw.bytes().all(|b| b.is_ascii_digit()) { if raw.is_empty() || !raw.bytes().all(|b| b.is_ascii_digit()) {
return None; return None;
} }
src/host/mod.rs
+1
View File
@@ -1,5 +1,6 @@
pub mod audio; pub mod audio;
mod capture; mod capture;
mod observer;
mod pipeline; mod pipeline;
mod quality; mod quality;
mod serve; mod serve;
src/host/observer/adapter.rs
+601
View File
@@ -0,0 +1,601 @@
//! 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<Mutex<Option<Projection>>>,
shutdown_tx: pw::channel::Sender<()>,
thread: Option<JoinHandle<()>>,
}
impl RegistryObserverHandle {
/// Spawn the read-only PipeWire registry observer.
pub fn spawn() -> Result<Self> {
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<Projection> {
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<BoundLink>,
}
struct ObserverState {
model: RegistryModel,
latest: Arc<Mutex<Option<Projection>>>,
last_candidate: Option<u32>,
live_globals: BTreeMap<GlobalId, VecDeque<LiveGlobal>>,
}
impl ObserverState {
fn new(latest: Arc<Mutex<Option<Projection>>>) -> 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<BoundLink> {
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<Mutex<Option<Projection>>>,
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::<u64>().ok()),
link_group: props.get("node.link-group").map(str::to_owned),
client_id: props
.get("client.id")
.and_then(|value| value.parse::<u32>().ok())
.map(GlobalId),
process_id: props
.get("application.process.id")
.and_then(|value| value.parse::<u32>().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::<u32>().ok())
.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
.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::<u32>().ok())
.map(GlobalId)
else {
tracing::warn!(
port_id = obj.id,
node_id = props.get("node.id").unwrap_or("<absent>"),
"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("<absent>"),
"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::<u32>().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<Serial> {
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("<absent>"),
"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<LinkEndpoints> {
let output_node = props.get("link.output.node")?.parse::<u32>().ok()?;
let input_node = props.get("link.input.node")?.parse::<u32>().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::<u32>().ok())
.map(GlobalId),
input_port: props
.get("link.input.port")
.and_then(|value| value.parse::<u32>().ok())
.map(GlobalId),
})
}
fn optional_global_id(id: u32) -> Option<GlobalId> {
(id != pw::constants::ID_ANY).then_some(GlobalId(id))
}
#[cfg(test)]
mod tests {
use super::*;
use std::process::Command;
struct PactlModule {
id: Option<u32>,
}
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::<u32>()
.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
});
}
}
src/host/observer/classify.rs
+161
View File
@@ -0,0 +1,161 @@
//! 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.
///
/// ⚠️ **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",
];
/// 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 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
/// 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<GlobalId>,
/// `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<String>,
/// `factory.name` — the discriminator. Only an allowlisted hardware-PCM
/// factory earns `session_device`.
pub factory_name: Option<String>,
/// `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. `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<String>,
}
/// 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 on_factory_allowlist = claim
.factory_name
.as_deref()
.is_some_and(|f| HARDWARE_PCM_FACTORIES.contains(&f));
// 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 && driver_ok;
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
}
}
src/host/observer/mod.rs
+512
View File
@@ -0,0 +1,512 @@
//! 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`]).
//!
//! **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:** 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
//! 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.
pub mod adapter;
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<String>,
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<GlobalId>,
pub input_port: Option<GlobalId>,
}
/// 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<LinkEndpoints>,
},
/// The bind-`LinkInfoRef` fallback resolved a Link's endpoints.
LinkEndpointsResolved {
serial: Serial,
endpoints: LinkEndpoints,
},
/// The adapter read `/proc/<pid>/comm` (`None` = the read failed / the
/// process is gone). Validates the pulse-PID candidate.
ProcCommProbed { pid: u32, comm: Option<String> },
/// 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,
/// 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.
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<u32>,
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<Serial, NodeSnapshot>,
ports: BTreeMap<Serial, PortSnapshot>,
links: BTreeMap<Serial, LinkSnapshot>,
clients: BTreeMap<Serial, ClientSnapshot>,
/// Nodes held out of the snapshot pending their Device's resolution.
withheld: BTreeMap<Serial, NodeObservation>,
/// Links whose endpoints the adapter is still binding; the id is kept so
/// removal and resolution can find them.
pending_links: BTreeMap<Serial, GlobalId>,
/// Live Device global ids, ref-counted so a recycled id is only
/// considered resolved while a Device actually holds it.
resolved_devices: BTreeMap<GlobalId, usize>,
/// Insertion-ordered holders of each live global id. `global_remove`
/// accounts for the oldest generation first (v3.4 §6.1.3).
live_ids: BTreeMap<GlobalId, VecDeque<Slot>>,
/// `/proc/<pid>/comm` reads keyed by pid, for pulse-PID validation.
probed_comm: BTreeMap<u32, Option<String>>,
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
}
/// 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) && !self.obligations_outstanding()
}
/// 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<u32> {
let clients: Vec<ClientSnapshot> = 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<Serial> = 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<LinkEndpoints>) {
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<u32> {
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,
}
}
src/host/observer/pulse_pid.rs
+89
View File
@@ -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/<pid>/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/<pid>/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<u32> {
let mut counts: BTreeMap<u32, usize> = 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/<candidate>/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<u32> {
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<String>) -> Option<u32> {
let candidate = candidate(clients)?;
validate(candidate, comm_of(candidate).as_deref())
}
src/host/observer/tests.rs
+717
View File
@@ -0,0 +1,717 @@
//! 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()),
alsa_driver_name: Some("snd_hda_intel".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<u32>) -> 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() {
// 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, "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_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
// 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
// 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()),
alsa_driver_name: Some("snd_hda_intel".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<u32>]) -> Vec<ClientSnapshot> {
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");
// 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]
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_epoch_is_sticky_but_graph_ready_is_dynamic() {
let mut m = model();
m.apply(RegEvent::ServerSynced);
assert_eq!(m.readiness(), Readiness::Complete);
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, "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]
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"
);
}