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]