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

repair: read and unload Pulse modules through libpulse, not pactl text

Browse Source 
Third review round found three more blocking P2s, and they shared one root cause:
`pactl`'s text output cannot carry the guarantees repair was claiming. All three
are closed by talking to the protocol instead. New dependency taken with the
user's explicit sign-off after vetting.

**Record boundaries were unprovable.** `pactl list short modules` prints a
module's argument raw into a tab/newline-delimited format with no escaping. A
*genuine* module whose argument contains a newline renders a first line that reads
byte-exactly like one of our canonical forms, with the remainder dropped as an
unparseable continuation — no forged index, so the duplicate-index check could not
see it. Repair would have classified and unloaded a module it never saw in full.
**Field-confirmed on the live server**, because this needed no adversary: loading a
loopback whose argument is canonical-then-newline-then-`remix=false` (a real
loopback option) produces exactly that listing. A tab in the same position is
worse: it hid a sink reference from the gate that protects a still-referenced sink.

**Index and argument could be mis-paired.** The listing carrying exact arguments
(`-f json`) carries no index at all on pactl 17; the one carrying the index cannot
carry the argument faithfully. Correlating them by position — which the previous
commit did — is unsound whenever module names repeat: another client loading one
module and unloading another between the two calls leaves counts and names aligned
while every argument has shifted by one, so a foreign module inherits a canonical
fingerprint. The name check cannot see it and the retry never fires.

**Locality was a guess.** `PULSE_SERVER` is a fallback *list*, so
`unix:/missing tcp:remote:4713` passes any "starts with unix:" test and then
connects to another machine, where local pids mean nothing and a live remote host's
modules look dead. A remote server can also be selected by client config with the
variable unset entirely.

New `repair/introspect.rs` owns one verified-local connection: `pa_module_info`
gives index, name and exact argument in a single record, `pa_context_is_local()`
answers locality about the connection actually established, and unloading goes back
through that same connection so listing and destruction cannot disagree about which
server they mean. It holds no policy beyond refusing the wrong server; every
decision stays in the pure planner.

⚠️ **The field test caught a real bug that no unit test could have.** The first
version did its work correctly and then aborted on the way out:

    Assertion '!e->dead' failed at ../pulseaudio/src/pulse/mainloop.c:207,
    function mainloop_io_free(). Aborting.

SIGABRT, core dumped, exit 134 — a fully successful repair reporting failure to its
caller. Cause: Rust drops fields in declaration order and the context's teardown
frees IO events living in the mainloop, which I had declared first. **This is
exactly the invariant phase 0b exists for, met again one layer down.** Fixed, and
then hardened past the fix: `Drop` explicitly takes and destroys the context before
the mainloop, so the ordering no longer depends on where the fields are written.

Liveness keeps its `NSpid`/container checks but the claim is corrected: `NSpid > 1`
means "definitely nested", while `NSpid == 1` is NOT proof of the initial namespace
— its leftmost value is relative to the procfs that was mounted, so a nested
namespace with its own `/proc` reports one entry legitimately. These are negative
signals that fail closed, not a proof of trustworthy pids. Closing that properly
needs modules to carry an owner token (machine/boot plus pid-namespace identity),
which changes what pixelpass writes into the graph and how far back `--repair` can
clean up: recorded as a design decision, not guessed at.

libpulse-binding 2.30.1 vetted before use: MIT/Apache-2.0, 5.5M downloads, 3 new
crates total, build script does nothing but probe pkg-config, no network or
subprocess use anywhere in the sources, and all three historical RustSec advisories
(2018-0020, 2018-0021, 2019-0038) were fixed by 2.6.0. The reasoning is recorded in
Cargo.toml beside the dependency.

247 tests, clippy clean, fmt clean apart from the pre-existing taint/tests.rs:2683.
The text parser's tests are gone with the parser; the liveness probe keeps its own,
and the live field gates — A/B orphan removal, the reference/unrecognised fixture,
and the newline fixture — all pass with exit 0.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
mollusk
2026-07-27 01:23:23 -04:00
co-authored by Claude Opus 5
parent 01c582427b
commit 45464b4af5
5 changed files with 405 additions and 277 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+39
View File
@@ -2958,6 +2958,33 @@ version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
[[package]]
name = "libpulse-binding"
version = "2.30.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "909eb3049e16e373680fe65afe6e2a722ace06b671250cc4849557bc57d6a397"
dependencies = [
"bitflags 2.13.0",
"libc",
"libpulse-sys",
"num-derive",
"num-traits",
"winapi",
]
[[package]]
name = "libpulse-sys"
version = "1.23.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d74371848b22e989f829cc1621d2ebd74960711557d8b45cfe740f60d0a05e61"
dependencies = [
"libc",
"num-derive",
"num-traits",
"pkg-config",
"winapi",
]
[[package]]
name = "libredox"
version = "0.1.18"
@@ -3539,6 +3566,17 @@ version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441"
[[package]]
name = "num-derive"
version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed3955f1a9c7c0c15e092f9c887db08b1fc683305fdf6eb6684f22555355e202"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "num-traits"
version = "0.2.19"
@@ -4163,6 +4201,7 @@ dependencies = [
"iroh",
"iroh-tickets",
"ksni",
"libpulse-binding",
"nix 0.30.1",
"notify-rust",
"pipewire",
Cargo.toml
+10
View File
@@ -46,6 +46,16 @@ serde_json = "1"
directories = "5"
ashpd = { version = "0.9", default-features = false, features = ["tokio"] }
pipewire = "0.9"
# `--repair` reads and unloads Pulse modules through libpulse introspection rather
# than by parsing `pactl` output. `pa_module_info` carries index, name and the exact
# argument in one record, and `pa_context_is_local()` answers whether the server we
# actually reached is local — neither of which the text listings can do (an argument
# may contain tabs and newlines that the short format cannot escape, the JSON
# listing carries no module index at all, and `PULSE_SERVER` is a fallback list, so
# it never proved locality). Vetted at 2.30.1: MIT/Apache-2.0, no build script
# beyond a pkg-config probe, no network or subprocess use, and all three historical
# RustSec advisories (2018-0020, 2018-0021, 2019-0038) fixed by 2.6.0.
libpulse-binding = "2.30"
x11rb = { version = "0.13", default-features = false, features = ["allow-unsafe-code"] }
uuid = { version = "1", features = ["v4"] }
iroh-tickets = "1.0.0"
src/repair/introspect.rs
+286
View File
@@ -0,0 +1,286 @@
//! Structured Pulse introspection: the observation and destruction layer for
//! `--repair`.
//!
//! # Why this replaced parsing `pactl`
//!
//! Three separate defects, all of them consequences of reading a human-oriented
//! text format rather than the protocol:
//!
//! 1. **Record boundaries were unprovable.** `pactl list short modules` prints a
//! module's argument raw into a tab-and-newline-delimited format with no
//! escaping. A *genuine* module whose argument contains a newline — say
//! `…latency_msec=20\nremix=false`, and `remix` is a real loopback option —
//! renders a first line that reads byte-exactly like one of our canonical
//! forms, with the remainder dropped as an unparseable continuation. No index
//! is forged, so no duplicate-index check can see it: repair would classify and
//! unload a module it had never actually seen in full. A tab in the same
//! position instead hides a sink reference, which is worse, because the gate
//! that protects a still-referenced sink then cannot see the reference.
//! 2. **Index and argument could be mis-paired.** The one listing that carries the
//! exact argument (`-f json`) carries **no index** at all on pactl 17, and the
//! one that carries the index cannot carry the argument faithfully. Combining
//! them by position is unsound whenever module names repeat: another client
//! loading one module and unloading another between the two calls leaves the
//! counts and names aligned while every argument has shifted by one.
//! 3. **Locality was a guess.** `PULSE_SERVER` is a *fallback list*, so
//! `unix:/missing tcp:remote:4713` passes any "starts with unix:" test and then
//! connects to another machine — where our local pids mean nothing and a live
//! remote host's modules look dead.
//!
//! `pa_module_info` carries index, name and argument together in one structured
//! record, so (1) and (2) cannot arise. `pa_context_is_local()` answers (3) about
//! the connection that actually got established rather than about a string we
//! hoped described it. And because unloading goes back through the *same*
//! connection, there is no window in which listing and destruction could disagree
//! about which server they are talking to.
//!
//! # What is deliberately not here
//!
//! No decisions. This module observes and destroys; every judgement about what may
//! be destroyed lives in [`super::plan`], which is pure and needs no Pulse server
//! to test. The one policy this layer owns is *refusing to talk to the wrong
//! server at all*.
use anyhow::{Context as _, Result, bail};
use libpulse_binding::callbacks::ListResult;
use libpulse_binding::context::{Context, FlagSet as ContextFlagSet, State as ContextState};
use libpulse_binding::mainloop::standard::{IterateResult, Mainloop};
use libpulse_binding::operation::{Operation, State as OperationState};
use libpulse_binding::proplist::{Proplist, properties};
use std::cell::RefCell;
use std::rc::Rc;
use std::time::{Duration, Instant};
use super::plan::ModuleObservation;
/// How long to wait for the connection to reach `Ready`. A one-shot CLI must not
/// hang on an unresponsive server; failing closed here costs the user a re-run.
const CONNECT_BUDGET: Duration = Duration::from_secs(3);
/// How long any single introspection request may take.
const REQUEST_BUDGET: Duration = Duration::from_secs(3);
/// How long to sleep between mainloop iterations while waiting. Non-blocking
/// iteration plus a short sleep keeps the deadline enforceable, which
/// `iterate(true)` would not.
const POLL_INTERVAL: Duration = Duration::from_millis(2);
/// A live, verified-local connection to the Pulse server.
///
/// Both listing and unloading run through this one connection, so everything
/// repair sees and everything it destroys provably belong to the same server.
///
/// ⚠️ **Field order is load-bearing, and this was not theoretical.** Rust drops
/// fields in declaration order, and the context's teardown frees IO events that
/// live *in* the mainloop. With `mainloop` declared first, `--repair` did its work
/// correctly and then died on the way out:
///
/// ```text
/// Assertion '!e->dead' failed at ../pulseaudio/src/pulse/mainloop.c:207,
/// function mainloop_io_free(). Aborting.
/// ```
///
/// SIGABRT, a core dump, and exit 134 — so a completely successful repair reported
/// failure to its caller. This is the same invariant phase 0b's
/// `ScreenshareTeardown` exists for, met again one layer down.
///
/// Rather than leave that resting on where the fields happen to be written, [`Drop`]
/// **explicitly** takes and drops the context first, so the ordering survives a
/// future reorder of this struct. The declaration order below is still correct, and
/// now it is also not load-bearing.
pub struct PulseSession {
/// `Option` only so that `Drop` can `take()` it and destroy it *before* the
/// mainloop. Always `Some` for the whole of the session's usable life.
context: Option<Context>,
mainloop: Mainloop,
}
impl Drop for PulseSession {
fn drop(&mut self) {
// Disconnect, then destroy the context while the mainloop it registered IO
// events with is still alive, then let the mainloop go. Bounded and
// best-effort: this runs on the way out, with nobody left to report to.
if let Some(mut context) = self.context.take() {
context.disconnect();
drop(context);
}
for _ in 0..8 {
if matches!(self.mainloop.iterate(false), IterateResult::Success(_)) {
continue;
}
break;
}
}
}
impl PulseSession {
/// The live context. Infallible in practice: only `Drop` ever clears it, and
/// nothing can call this afterwards.
fn context(&mut self) -> &mut Context {
self.context
.as_mut()
.expect("the context is only taken during Drop")
}
/// Connect, wait for readiness, and refuse anything but a local server.
pub fn connect() -> Result<Self> {
let mut proplist = Proplist::new().context("could not allocate a Pulse proplist")?;
// `set_str` fails only on an invalid key, and these keys are constants.
let _ = proplist.set_str(properties::APPLICATION_NAME, "pixelpass --repair");
let _ = proplist.set_str(properties::APPLICATION_ID, "xyz.pixelpass.repair");
let mut mainloop = Mainloop::new().context("could not create a Pulse mainloop")?;
let mut context = Context::new_with_proplist(&mainloop, "pixelpass --repair", &proplist)
.context("could not create a Pulse context")?;
context
.connect(None, ContextFlagSet::NOFLAGS, None)
.context("could not connect to the Pulse server")?;
let deadline = Instant::now() + CONNECT_BUDGET;
loop {
iterate_once(&mut mainloop)?;
match context.get_state() {
ContextState::Ready => break,
ContextState::Failed => {
bail!("the Pulse server refused the connection");
}
ContextState::Terminated => {
bail!("the Pulse connection terminated before it was ready");
}
_ => {
if Instant::now() >= deadline {
bail!(
"the Pulse server did not become ready within {:?}",
CONNECT_BUDGET
);
}
std::thread::sleep(POLL_INTERVAL);
}
}
}
// The connection is established, so this is about the server we actually
// reached — not about what a server string appeared to promise. A remote
// server's module table belongs to another machine's processes, where our
// pids mean nothing, so repair must not touch it.
match context.is_local() {
Some(true) => {}
Some(false) => bail!(
"connected to a REMOTE Pulse server; --repair only ever operates on the local \
server, because it decides what to unload from local process liveness"
),
None => bail!(
"could not determine whether the Pulse server is local; refusing to unload \
anything"
),
}
Ok(Self {
context: Some(context),
mainloop,
})
}
/// Every loaded module, with its exact argument.
pub fn list_modules(&mut self) -> Result<Vec<ModuleObservation>> {
// `Rc<RefCell<…>>` because the callback is owned by the C library and may
// be invoked many times before the operation completes.
let collected: Rc<RefCell<Vec<ModuleObservation>>> = Rc::new(RefCell::new(Vec::new()));
let failed: Rc<RefCell<bool>> = Rc::new(RefCell::new(false));
let sink = Rc::clone(&collected);
let error_flag = Rc::clone(&failed);
let op = self
.context()
.introspect()
.get_module_info_list(move |result| match result {
ListResult::Item(info) => {
// A module with no name is not one we can identify, and an
// argumentless module is simply one loaded without arguments.
let name = info.name.as_deref().unwrap_or_default();
let args = info.argument.as_deref().unwrap_or_default();
sink.borrow_mut()
.push(ModuleObservation::new(info.index, name, args));
}
ListResult::End => {}
ListResult::Error => *error_flag.borrow_mut() = true,
});
self.run_to_completion(op, "list modules")?;
if *failed.borrow() {
bail!("the Pulse server returned an error while listing modules");
}
let modules = collected.borrow().clone();
// Impossible per the protocol — an index identifies one module — so this is
// a sanity check on external input, not a safety boundary. It fails closed
// because an ambiguous index is one we could unload wrongly.
for (i, module) in modules.iter().enumerate() {
if modules[..i].iter().any(|earlier| earlier.id == module.id) {
bail!(
"the Pulse server reported module index #{} twice; refusing to unload \
anything",
module.id
);
}
}
Ok(modules)
}
/// Unload one module, over the same connection it was observed on.
pub fn unload_module(&mut self, index: u32) -> Result<()> {
let succeeded: Rc<RefCell<Option<bool>>> = Rc::new(RefCell::new(None));
let outcome = Rc::clone(&succeeded);
let op = self
.context()
.introspect()
.unload_module(index, move |success| *outcome.borrow_mut() = Some(success));
self.run_to_completion(op, "unload module")?;
match *succeeded.borrow() {
Some(true) => Ok(()),
Some(false) => bail!("the Pulse server rejected unloading module #{index}"),
// The operation completed without the callback running, which we cannot
// read as success.
None => bail!("no result was reported for unloading module #{index}"),
}
}
/// Drive the mainloop until `op` finishes, or the budget expires.
fn run_to_completion<T: ?Sized>(&mut self, op: Operation<T>, what: &str) -> Result<()> {
let deadline = Instant::now() + REQUEST_BUDGET;
loop {
iterate_once(&mut self.mainloop)?;
match op.get_state() {
OperationState::Done => return Ok(()),
OperationState::Cancelled => {
bail!("the Pulse server cancelled the request to {what}");
}
OperationState::Running => {
// A connection that dies mid-request would otherwise be waited
// out to the full budget.
match self.context().get_state() {
ContextState::Ready => {}
state => {
bail!("the Pulse connection became {state:?} while trying to {what}")
}
}
if Instant::now() >= deadline {
bail!("the Pulse server did not {what} within {REQUEST_BUDGET:?}");
}
std::thread::sleep(POLL_INTERVAL);
}
}
}
}
}
/// One non-blocking mainloop iteration, with quit and error surfaced as errors.
fn iterate_once(mainloop: &mut Mainloop) -> Result<()> {
match mainloop.iterate(false) {
IterateResult::Success(_) => Ok(()),
IterateResult::Quit(code) => {
bail!("the Pulse mainloop quit unexpectedly (code {})", code.0)
}
IterateResult::Err(e) => Err(e).context("the Pulse mainloop failed"),
}
}
src/repair/mod.rs
+65 -249
View File
@@ -16,47 +16,21 @@
//! connection-owned and removes itself when its host dies, so there is nothing
//! there for repair to do and no safe way for it to help.
//!
//! # Where the observations come from: one listing, atomically
//! # Where the observations come from
//!
//! `pactl list short modules`, in a single invocation, and nothing else. Every
//! observation's `(index, name, argument)` therefore comes from **one** server
//! response and cannot be mis-assembled.
//!
//! This replaced a two-listing scheme (indices from the short listing, exact
//! arguments from `pactl -f json list modules`, correlated by position). The JSON
//! listing was needed because it carries the exact argument where the short
//! listing's is tab-delimited text — but on pactl 17 its records carry **no module
//! index at all** (`"index": null`) while `unload-module` accepts only an index, so
//! it can never stand alone. Correlating the two by position is unsound with
//! repeated module names: if another client loads one module and unloads another
//! between the two calls, the counts and names still line up while the arguments
//! have shifted by one, and a *foreign* module can inherit a canonical
//! fingerprint. The name check cannot see that, and the retry never fires because
//! correlation "succeeded".
//!
//! Two consequences of using the short listing alone, both handled rather than
//! hoped away:
//!
//! - **A tab inside an argument is invisible to this format.** Such a row is marked
//! `args_complete: false`; [`plan::classify`] refuses it outright (a truncation
//! could otherwise coincide with a canonical form) while the reference gate can
//! still see that the row names a sink.
//! - **A crafted argument containing a newline can fabricate a row.** A fabricated
//! row only does damage if it names a *real* module's index — which makes that
//! index appear twice — so a duplicated index refuses the whole run.
//!
//! The remaining exact route is libpulse introspection (`pa_module_info` carries
//! index, name and argument in one record). That is a new dependency plus a
//! mainloop in a one-shot CLI path, so it is recorded as the upgrade rather than
//! taken now; see the deferred item in the impl plan.
//! Structured Pulse introspection over one verified-local connection — see
//! [`introspect`], which also documents the three defects that parsing `pactl`'s
//! text output turned out to have. Listing *and* unloading both go through that
//! same connection.
pub mod introspect;
pub mod plan;
use anyhow::{Context, Result, bail};
use std::path::Path;
use std::process::Command;
use plan::{Fingerprint, Liveness, ModuleObservation, Shape};
use introspect::PulseSession;
use plan::{Fingerprint, Liveness, Shape};
pub async fn run() -> Result<()> {
let liveness = LivenessProbe::new();
@@ -67,7 +41,10 @@ pub async fn run() -> Result<()> {
);
}
let modules = snapshot().context("failed to observe pactl modules")?;
let mut pulse = PulseSession::connect().context("could not observe the Pulse module table")?;
let modules = pulse
.list_modules()
.context("could not list Pulse modules")?;
// Say so loudly when something names our sinks but matches no shape we know:
// that is either a third party using our names, or a newer pixelpass whose
@@ -123,7 +100,9 @@ pub async fn run() -> Result<()> {
// Fresh snapshot per action. Deliberately not hoisted out of the loop:
// each unload changes the module table, and the point is to decide
// against the table as it is *now*, not as it was when the plan was made.
let current = snapshot().context("failed to re-observe pactl modules")?;
let current = pulse
.list_modules()
.context("could not re-list Pulse modules")?;
let Some(obs) = current.iter().find(|m| m.id == fp.id) else {
println!(
"[pixelpass] --repair: module #{} is already gone; skipping",
@@ -182,7 +161,7 @@ pub async fn run() -> Result<()> {
}
}
match unload_module(fp.id) {
match pulse.unload_module(fp.id) {
Ok(()) => {
println!("[pixelpass] --repair: {}", describe(fp));
unloaded += 1;
@@ -234,26 +213,36 @@ fn describe(fp: &Fingerprint) -> String {
/// Answers "is this pid still around", and knows when it must refuse to answer.
///
/// Two separate hazards, and the first one is the dangerous one:
/// **An error is not an absence.** `Path::exists()` folds permission errors and a
/// missing `/proc` into `false`, which here would read as "dead, go ahead and
/// unload". Liveness is asked with `kill(pid, 0)` instead, where `EPERM` *proves*
/// existence.
///
/// 1. **A pid can be alive and invisible.** Inside a pid namespace — a container,
/// a distrobox — `/proc/self` is perfectly visible while every process in the
/// *parent* namespace is not. Repair there can reach the host's Pulse socket,
/// see a live host's modules, conclude its pid is dead, and unload a running
/// host's audio. `hidepid` has the same self-visible/others-invisible shape. So
/// a "can I see myself" preflight proves nothing; what is needed is positive
/// confidence that our pid numbers mean the same thing as the ones in the
/// module names. `NSpid` in `/proc/self/status` answers that directly: more than
/// one entry means we are nested, and every verdict becomes `Unknown`.
/// 2. **An error is not an absence.** `Path::exists()` maps permission errors and a
/// missing `/proc` to `false`, which here reads as "dead, go ahead". Liveness is
/// asked with `kill(pid, 0)` instead, where `EPERM` *proves* existence.
/// # The limit of what this can prove, stated rather than papered over
///
/// A remote Pulse server also fails closed: the module table then belongs to
/// another machine's processes, where our pids mean nothing at all.
/// A pid can be alive and invisible. Inside a pid namespace — a container, a
/// distrobox — `/proc/self` is perfectly visible while every process in the
/// *parent* namespace is not, and `hidepid` has the same self-visible,
/// others-invisible shape. Repair in such a place can reach the host's Pulse
/// socket, see a live host's modules, get `ESRCH` for its pid and unload a running
/// host's audio.
///
/// The signals below are **negative** ones: they detect *some* cases where pid
/// numbers cannot be trusted, and every one of them fails closed. What they cannot
/// do is prove the converse. `NSpid` reports this process's pid in each namespace
/// that its procfs can see, and its leftmost value is relative to the pid namespace
/// that mounted that procfs — so a nested namespace with its own `/proc` reports a
/// single entry quite legitimately. `NSpid > 1` therefore means "definitely
/// nested", while `NSpid == 1` means only "not detectably nested".
///
/// Closing that properly needs the module itself to carry an owner token (machine
/// and boot identity plus pid-namespace identity) written at load time, with
/// token-less modules treated as `Unknown`. That changes what pixelpass writes into
/// the graph and how far back `--repair` can clean up, so it is a design decision
/// recorded in the impl plan rather than guessed at here.
struct LivenessProbe {
/// `None` when pid numbers here are trustworthy; `Some(reason)` when every
/// answer must be [`Liveness::Unknown`].
/// `None` when no signal says pid numbers are untrustworthy; `Some(reason)`
/// when every answer must be [`Liveness::Unknown`].
degraded: Option<String>,
}
@@ -265,18 +254,12 @@ impl LivenessProbe {
}
fn detect_degradation() -> Option<String> {
// A remote server's modules belong to another machine's pids.
if let Some(server) = std::env::var_os("PULSE_SERVER") {
let server = server.to_string_lossy().to_string();
let local =
server.starts_with("unix:") || server.starts_with('/') || server.starts_with("{");
if !local {
return Some(format!("PULSE_SERVER={server} is not a local socket"));
}
}
// The authoritative namespace question: NSpid lists this process's pid in
// every namespace it is visible in, outermost first. More than one entry
// means our pid numbers are not the ones the outer namespace uses.
// Locality is deliberately NOT checked here. `PULSE_SERVER` is a fallback
// *list*, so `unix:/missing tcp:remote:4713` starts with "unix:" and still
// connects to another machine, and a remote server can be selected by client
// configuration with the variable unset entirely. The authoritative answer
// comes from `pa_context_is_local()` on the connection that actually got
// established — see `introspect::PulseSession::connect`.
match std::fs::read_to_string("/proc/self/status") {
Ok(status) => {
let nspid = status
@@ -290,6 +273,8 @@ impl LivenessProbe {
so pids in module names may belong to processes it cannot see"
));
}
// NB: a single entry is not proof of the initial namespace — see
// the type's doc comment. It only means nothing detected it.
// A kernel too old to report NSpid cannot rule nesting out.
None => {
return Some(
@@ -336,191 +321,20 @@ impl LivenessProbe {
}
}
// ──────────────────────────────────────────────────────────────────────
// Observation
// ──────────────────────────────────────────────────────────────────────
/// One observation of the whole module table, from a single `pactl` invocation.
fn snapshot() -> Result<Vec<ModuleObservation>> {
let output = Command::new("pactl")
.args(["list", "short", "modules"])
.output()
.context("failed to run pactl")?;
if !output.status.success() {
bail!(
"pactl list short modules failed: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
let text = String::from_utf8(output.stdout).context("pactl returned non-UTF-8")?;
parse_short(&text)
}
/// Parse `pactl list short modules`: `<index>\t<name>\t<argument>` per module, one
/// server response, so a row's three fields always belong together.
///
/// Two format hazards, both handled rather than assumed away:
///
/// - **Tabs.** pactl emits a trailing tab after the argument, so a well-formed row
/// has at most one empty field after it. Any *non-empty* content past the
/// argument means the argument itself contained a tab and this format cannot
/// show it in full: the row is marked `args_complete: false`, never classified,
/// but still visible to the reference gate.
/// - **Newlines.** Some modules render arguments as a multi-line `{ … }` block; the
/// continuation lines never parse as a u32 index, so they are dropped. A crafted
/// argument *could* still fabricate a row that does parse — but to do damage it
/// must claim a real module's index, which then appears twice. A duplicated index
/// therefore refuses the whole run.
fn parse_short(text: &str) -> Result<Vec<ModuleObservation>> {
let mut rows: Vec<ModuleObservation> = Vec::new();
for line in text.lines() {
let fields: Vec<&str> = line.split('\t').collect();
let Some(id_str) = fields.first() else {
continue;
};
let Ok(id) = id_str.parse::<u32>() else {
continue;
};
let Some(name) = fields.get(1) else { continue };
let args = fields.get(2).copied().unwrap_or("");
let complete = fields.iter().skip(3).all(|extra| extra.is_empty());
rows.push(if complete {
ModuleObservation::new(id, name, args)
} else {
ModuleObservation::truncated(id, name, args)
});
}
// A repeated index is the one shape a fabricated row needs in order to make us
// unload a module that exists. It is also never legitimate.
for (i, row) in rows.iter().enumerate() {
if rows[..i].iter().any(|earlier| earlier.id == row.id) {
bail!(
"module index #{} appears more than once in `pactl list short modules`; \
refusing to unload anything",
row.id
);
}
}
Ok(rows)
}
fn unload_module(id: u32) -> Result<()> {
let output = Command::new("pactl")
.arg("unload-module")
.arg(id.to_string())
.output()
.context("failed to run pactl unload-module")?;
if !output.status.success() {
bail!(
"pactl unload-module #{id}: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
/// The whole row comes from one response: index, name and the exact argument,
/// including pactl's trailing tab.
#[test]
fn parses_index_name_and_exact_argument() {
let text = "5\tmodule-null-sink\tsink_name=pixelpass_capture_42\t\n\
10\tmodule-loopback\tsource=@DEFAULT_SINK@.monitor \
sink=pixelpass_capture_42 latency_msec=20\t\n\
7\tmodule-always-sink\n";
let rows = parse_short(text).expect("well-formed");
assert_eq!(rows.len(), 3);
assert_eq!(rows[0].id, 5);
assert_eq!(rows[0].args, "sink_name=pixelpass_capture_42");
assert!(rows[0].args_complete, "a trailing tab is not truncation");
assert!(rows[1].args.ends_with("latency_msec=20"));
assert_eq!(rows[2].args, "", "a module without arguments still parses");
}
// The `pactl`-text parser that used to live here is gone, and so are its
// tests: `introspect` gets index, name and argument as structured fields, so
// there is no format left to mis-parse. What replaced those tests is the live
// field gate, since the remaining risk is in talking to the server, which no
// unit test can exercise. The decisions all live in `plan`, which is pure and
// tested there.
/// An argument containing a tab cannot be shown in full by this format. Such a
/// row must never be classified — a truncation could coincide with a canonical
/// form — but it must still be *present*, or the reference gate would not see
/// that something names our sink.
#[test]
fn a_tab_inside_an_argument_marks_the_row_incomplete() {
let canonical = "source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_42 latency_msec=20";
let text = format!("10\tmodule-loopback\t{canonical}\tremix=false\n");
let rows = parse_short(&text).expect("well-formed");
assert_eq!(rows.len(), 1);
assert!(!rows[0].args_complete);
assert_eq!(
plan::classify(&rows[0]),
None,
"a truncated argument that happens to match must not be ours"
);
// Still visible to the gate that protects a sink from being destroyed.
assert_eq!(
plan::sink_still_referenced(&rows, 42, 999),
Some(10),
"an unreadable argument still counts as a reference"
);
}
/// Modules whose arguments render as a `{ … }` block wrap onto continuation
/// lines; swallowing one as a module would fabricate an entry — and a
/// fabricated entry is an index we might later unload.
#[test]
fn ignores_continuation_lines_of_multi_line_argument_blocks() {
let text = "1\tlibpipewire-module-rt\t{\n\
\x20 nice.level = -11\n\
\x20 rt.prio = 88\n\
}\n\
5\tmodule-null-sink\tsink_name=pixelpass_capture_42\n";
let rows = parse_short(text).expect("well-formed");
assert_eq!(rows.len(), 2, "{rows:?}");
assert_eq!(rows[1].id, 5);
}
/// A crafted argument can fabricate a row that parses. It only does damage if
/// it claims a *real* module's index — which makes that index appear twice — so
/// a duplicate index refuses the entire run rather than unloading a stranger.
#[test]
fn a_duplicated_index_refuses_the_whole_run() {
// As it would arrive: a module whose argument embeds a newline and a row
// that re-uses index 5, which really belongs to something else.
let text = "5\tmodule-real-thing\tconfig={\n\
5\tmodule-null-sink\tsink_name=pixelpass_capture_4242\n";
let err = parse_short(text).expect_err("a repeated index is never legitimate");
assert!(
format!("{err:#}").contains("more than once"),
"unexpected error: {err:#}"
);
}
/// End-to-end from raw pactl text: the parser feeding the planner. Without
/// this, a parser that dropped every argument would leave the entire planner
/// suite green while real `--repair` recognised nothing at all.
#[test]
fn raw_pactl_output_becomes_a_plan() {
let text = "1\tlibpipewire-module-rt\t{\n\
\x20 nice.level = -11\n\
}\n\
10\tmodule-loopback\tsource=@DEFAULT_SINK@.monitor \
sink=pixelpass_capture_4242 latency_msec=20\t\n\
11\tmodule-loopback\tsource=pixelpass_capture_4242.monitor \
sink=@DEFAULT_SINK@ latency_msec=20\t\n";
let observations = parse_short(text).expect("well-formed");
let planned = plan::plan(&observations, |_| Liveness::Dead);
assert_eq!(
planned.unload.iter().map(|fp| fp.id).collect::<Vec<_>>(),
vec![10, 11],
"both orphan loopbacks must be planned from raw output: {planned:?}"
);
assert_eq!(planned.dead_pids.len(), 1);
}
/// The probe must refuse to call anything dead when pid numbers here may not
/// mean what the module names mean. On this host it should be confident; the
/// point of the assertion is that `of()` never returns `Dead` while degraded.
/// The probe must never say `Dead` when it cannot see the whole pid space, and
/// must never ask `kill` about a pid that would signal something other than one
/// process.
#[test]
fn a_degraded_probe_never_reports_dead() {
let degraded = LivenessProbe {
@@ -530,14 +344,16 @@ mod tests {
assert_eq!(degraded.of(u32::MAX), Liveness::Unknown);
let probe = LivenessProbe::new();
// Our own pid is alive by construction — unless this test itself runs in a
// pid namespace, which is exactly when the probe must abstain.
// Our own pid is alive by construction — unless this test itself runs
// somewhere the probe must abstain, which is exactly the other branch.
let me = std::process::id();
match probe.degraded_reason() {
None => assert_eq!(probe.of(me), Liveness::Alive),
Some(_) => assert_eq!(probe.of(me), Liveness::Unknown),
}
// pid 0 would signal our whole process group; it is never askable.
// `kill(0, …)` would signal our whole process group, and a pid past
// `i32::MAX` cannot be expressed to `kill` at all.
assert_eq!(probe.of(0), Liveness::Unknown);
assert_eq!(probe.of(u32::MAX), Liveness::Unknown);
}
}
src/repair/plan.rs
+5 -28
View File
@@ -97,24 +97,16 @@ pub fn recorded_argument(args: &[String]) -> String {
/// One module exactly as the server reported it.
///
/// `args` is the **exact** recorded argument string, not a normalised one. Within
/// a single repair invocation every snapshot comes from the same server, so
/// re-rendering is not a thing that happens, and normalising would only make two
/// genuinely different arguments compare equal (whitespace inside a quoted
/// property value is not layout).
/// `args` is the **exact** argument string from `pa_module_info`, not a normalised
/// one, and not a reconstruction from a text listing. Normalising would only make
/// two genuinely different arguments compare equal — whitespace inside a quoted
/// property value is not layout — and every snapshot within one invocation comes
/// from the same connection, so there is no re-rendering to absorb.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ModuleObservation {
pub id: u32,
pub name: String,
pub args: String,
/// False when the observation format could not carry the whole argument (the
/// short listing is tab-delimited, and an argument may itself contain a tab).
///
/// A truncated argument is never ours — [`classify`] refuses it outright,
/// because a truncation could otherwise coincide with a canonical form. It is
/// still kept in the snapshot: [`sink_still_referenced`] must be able to see
/// that *something* names a sink even when it cannot read the whole argument.
pub args_complete: bool,
}
impl ModuleObservation {
@@ -123,15 +115,6 @@ impl ModuleObservation {
id,
name: name.to_string(),
args: args.to_string(),
args_complete: true,
}
}
/// An observation whose argument the format could not fully carry.
pub fn truncated(id: u32, name: &str, args: &str) -> Self {
Self {
args_complete: false,
..Self::new(id, name, args)
}
}
}
@@ -357,12 +340,6 @@ impl Plan {
/// grows an argument, changes a latency, or repeats the pid cannot leave a matcher
/// quietly accepting the old form.
pub fn classify(obs: &ModuleObservation) -> Option<Fingerprint> {
// An argument we could not read in full is never ours: a truncation could
// coincide with a canonical form, and unloading on a coincidence is the one
// outcome none of these rules tolerate.
if !obs.args_complete {
return None;
}
for shape in ALL_SHAPES {
let template = shape.template();
if obs.name != template.module_name {