peerspeak/src/audio/ownership.rs

//! Ownership tagging for every audio node peerspeak puts on the graph.
//!
//! Zero behaviour change on its own. Its whole purpose is to make peerspeak's
//! own playback *identifiable from outside the process*, so that pixelpass's
//! screenshare audio exclusion engine can refuse to fan our own call audio
//! back into the stream the far end is already hearing (the echo it exists to
//! prevent). See `docs/screenshare-audio-exclusion-plan.md` §5.1 and
//! `docs/screenshare-audio-exclusion-impl-plan.md` §3.
//!
//! # The wire contract
//!
//! The literals below are a **cross-repo contract**, not local naming.
//! pixelpass matches on them and does not depend on this crate (nor this
//! crate on it) — the coupling is the PipeWire graph itself. The values are
//! committed in both repos as `tests/fixtures/ownership-tag-contract.txt`,
//! and [`contract_matches_fixture`](tests::contract_matches_fixture) asserts
//! these constants against that file. pixelpass's matching constants are in
//! `src/host/taint/mod.rs` (`PEERSPEAK_OWNED_PROP`,
//! `PEERSPEAK_OWNED_NODE_PREFIX`), asserted against its own copy of the same
//! fixture.
//!
//! # Why two carriers
//!
//! A node is owned if **either** carrier matches — a union, which is the
//! fail-closed direction (a missed tag means our audio gets shared; a
//! spurious tag only over-excludes). Round 8 of the design added the second
//! carrier after the phase-5 audit found that a node *property* is invisible
//! to the PipeWire registry's `global` event and recoverable only by binding
//! the node, while `node.name` is announced directly. The primary taint root
//! should not rest on a single observation mechanism.
//!
//! `node.description` is deliberately left alone, so mixers still show "mpv"
//! rather than an internal identifier.
//!
//! # Known limitation (accepted for v1)
//!
//! The child environment is inherited by *grandchildren*, so a player that
//! spawns its own audio helper tags it too. That over-excludes rather than
//! leaks, and is the accepted v1 limitation recorded in plan §5.1.

use std::process::Command;

/// Carrier 1, key. Set on every peerspeak-owned node. Read by pixelpass only
/// after binding the node — see the module docs.
pub const OWNED_PROP_KEY: &str = "peerspeak.owned";

/// Carrier 1, value. pixelpass treats anything other than `false`/`0` as
/// truthy (fail-closed), but we always emit exactly this.
pub const OWNED_PROP_VALUE: &str = "1";

/// Carrier 2. The `node.name` prefix pixelpass matches; the remainder of the
/// name is diagnostic only. Format: `peerspeak_owned_<role>_<pid>`.
pub const OWNED_NODE_NAME_PREFIX: &str = "peerspeak_owned_";

/// The environment variable through which a PulseAudio-API client (mpv/VLC on
/// `--ao=pulse`, `paplay`) receives extra stream properties.
pub const PULSE_PROP_ENV: &str = "PULSE_PROP";

/// The environment variable through which a PipeWire-native client
/// (`pw-play`, mpv on `--ao=pipewire`) receives extra stream properties.
pub const PIPEWIRE_PROPS_ENV: &str = "PIPEWIRE_PROPS";

/// Role of the native call-playback stream — the node carrying the far end's
/// voice.
pub const NATIVE_PLAYBACK_ROLE: &str = "call";

/// Role of a notification-sound player (`pw-play`/`paplay`/`aplay`).
pub const NOTIFICATION_ROLE: &str = "notify";

/// Build the `node.name` for a node this process owns.
///
/// The pid is **peerspeak's own**, not the child's: for a spawned player we
/// set the name before the fork, so the child's pid does not exist yet — and
/// the owning process is the more useful thing to record anyway.
pub fn owned_node_name(role: &str) -> String {
    format!(
        "{OWNED_NODE_NAME_PREFIX}{}_{}",
        sanitize_role(role),
        std::process::id()
    )
}

/// Keep a role to `[a-z0-9-]`, lowercased, so the generated `node.name` is a
/// single shell-safe, whitespace-free token.
///
/// Not defensive theatre: [`PULSE_PROP_ENV`] is a space-separated
/// `key=value` encoding, so a role containing a space would silently split
/// into a second bogus property. Every current caller passes a literal, and
/// this keeps that true no matter who calls next.
fn sanitize_role(role: &str) -> String {
    let cleaned: String = role
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() {
                c.to_ascii_lowercase()
            } else {
                '-'
            }
        })
        .collect();
    let trimmed = cleaned.trim_matches('-');
    if trimmed.is_empty() {
        "child".to_string()
    } else {
        trimmed.to_string()
    }
}

/// The `PULSE_PROP` value carrying both carriers.
fn pulse_prop_value(node_name: &str) -> String {
    format!("{OWNED_PROP_KEY}={OWNED_PROP_VALUE} node.name={node_name}")
}

/// The `PIPEWIRE_PROPS` value carrying both carriers.
///
/// SPA-JSON; keys are quoted because `peerspeak.owned` contains a dot.
fn pipewire_props_value(node_name: &str) -> String {
    format!("{{ \"{OWNED_PROP_KEY}\" = \"{OWNED_PROP_VALUE}\", \"node.name\" = \"{node_name}\" }}")
}

/// Tag a child process that will play audio, so its resulting node carries
/// both ownership carriers.
///
/// Both environment variables are set unconditionally, because we cannot know
/// from here which audio API the child will pick (mpv chooses at runtime, and
/// VLC and `paplay` differ from `pw-play`). Setting the one the child ignores
/// costs nothing.
///
/// Any inherited value of either variable is **replaced**, not appended: the
/// tag is a correctness mechanism, and a half-merged property string that
/// fails to parse would drop it silently.
pub fn tag_child(command: &mut Command, role: &str) {
    let node_name = owned_node_name(role);
    command.env(PULSE_PROP_ENV, pulse_prop_value(&node_name));
    command.env(PIPEWIRE_PROPS_ENV, pipewire_props_value(&node_name));
}

/// Live-test support for the phase-1 exit gate, shared by the two modules
/// that spawn tagged players (`notify`, `screenshare`).
///
/// Lives here rather than in either test module because the gate is one
/// property — "the carriers reach a live node" — asked of two call sites, and
/// two copies of the poll would drift.
#[cfg(all(test, not(windows)))]
pub mod live_test {
    use std::process::Command;
    use std::time::{Duration, Instant};

    /// Poll `pw-dump` for a `Stream/Output/Audio` node whose `node.name`
    /// starts with `prefix`; returns its name and `peerspeak.owned` value.
    ///
    /// Polls rather than sleeping once because node creation races the
    /// player process starting up: a fixed sleep either flakes or is slow.
    pub fn poll_for_owned_node(
        prefix: &str,
        timeout: Duration,
    ) -> Option<(String, Option<String>)> {
        let deadline = Instant::now() + timeout;
        while Instant::now() < deadline {
            if let Some(found) = scan_once(prefix) {
                return Some(found);
            }
            std::thread::sleep(Duration::from_millis(100));
        }
        None
    }

    fn scan_once(prefix: &str) -> Option<(String, Option<String>)> {
        let out = Command::new("pw-dump").output().ok()?;
        let dump: serde_json::Value = serde_json::from_slice(&out.stdout).ok()?;
        for object in dump.as_array()? {
            let props = &object["info"]["props"];
            // `continue`, never `?`: most objects in a dump are not nodes and
            // have no `node.name` at all, and propagating that as `None`
            // would abandon the scan at the first one.
            let Some(name) = props["node.name"].as_str() else {
                continue;
            };
            if !name.starts_with(prefix) {
                continue;
            }
            // PipeWire may hand the property back as a JSON string or a
            // number depending on how the client set it; both are the same
            // property, and reading only one would make the assertion
            // accidentally depend on the producer's encoding.
            let owned = props[super::OWNED_PROP_KEY]
                .as_str()
                .map(str::to_string)
                .or_else(|| props[super::OWNED_PROP_KEY].as_i64().map(|n| n.to_string()));
            return Some((name.to_string(), owned));
        }
        None
    }

    /// A zero-filled 48 kHz stereo s16 WAV of `seconds` length — long enough
    /// for the poll to catch the node, silent so the gate never makes noise.
    pub fn silent_wav(seconds: u32) -> Vec<u8> {
        let data_len = 48_000 * seconds * 4;
        let mut wav = Vec::with_capacity(44 + data_len as usize);
        wav.extend_from_slice(b"RIFF");
        wav.extend_from_slice(&(36 + data_len).to_le_bytes());
        wav.extend_from_slice(b"WAVEfmt ");
        wav.extend_from_slice(&16u32.to_le_bytes());
        wav.extend_from_slice(&1u16.to_le_bytes()); // PCM
        wav.extend_from_slice(&2u16.to_le_bytes()); // stereo
        wav.extend_from_slice(&48_000u32.to_le_bytes());
        wav.extend_from_slice(&(48_000u32 * 4).to_le_bytes()); // byte rate
        wav.extend_from_slice(&4u16.to_le_bytes()); // block align
        wav.extend_from_slice(&16u16.to_le_bytes()); // bits per sample
        wav.extend_from_slice(b"data");
        wav.extend_from_slice(&data_len.to_le_bytes());
        wav.resize(44 + data_len as usize, 0);
        wav
    }

    /// The `node.name` prefix a node tagged by *this* process with `role`
    /// must carry. Narrower than [`super::OWNED_NODE_NAME_PREFIX`] on
    /// purpose: it pins the match to this test run, so a stray tagged node
    /// left by anything else cannot make a gate pass.
    pub fn expected_prefix(role: &str) -> String {
        format!(
            "{}{role}_{}",
            super::OWNED_NODE_NAME_PREFIX,
            std::process::id()
        )
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::BTreeMap;

    /// The contract file, byte-identical to pixelpass's copy.
    const FIXTURE: &str = include_str!("../../tests/fixtures/ownership-tag-contract.txt");

    fn fixture() -> BTreeMap<String, String> {
        FIXTURE
            .lines()
            .map(str::trim)
            .filter(|line| !line.is_empty() && !line.starts_with('#'))
            .map(|line| {
                let (key, value) = line
                    .split_once('=')
                    .unwrap_or_else(|| panic!("fixture line is not key=value: {line:?}"));
                (key.to_string(), value.to_string())
            })
            .collect()
    }

    fn fixture_get(key: &str) -> String {
        fixture()
            .get(key)
            .unwrap_or_else(|| panic!("fixture has no key {key:?}"))
            .clone()
    }

    /// Half of the cross-repo contract test: our constants are exactly what
    /// the shared fixture pins. pixelpass runs the mirror of this against its
    /// own byte-identical copy.
    #[test]
    fn contract_matches_fixture() {
        assert_eq!(OWNED_PROP_KEY, fixture_get("prop_key"));
        assert_eq!(OWNED_PROP_VALUE, fixture_get("prop_value"));
        assert_eq!(OWNED_NODE_NAME_PREFIX, fixture_get("node_name_prefix"));
    }

    /// The fixture's documented format and example must agree with the
    /// generator, so a future reader can trust the example.
    #[test]
    fn fixture_example_is_well_formed() {
        let example = fixture_get("node_name_example");
        assert!(example.starts_with(&fixture_get("node_name_prefix")));
        let format = fixture_get("node_name_format");
        assert_eq!(format, "peerspeak_owned_<role>_<pid>");
        // Same shape our generator emits: prefix, role, pid.
        let generated = owned_node_name("mpv");
        let suffix = generated
            .strip_prefix(OWNED_NODE_NAME_PREFIX)
            .expect("generated name carries the prefix");
        let (role, pid) = suffix.split_once('_').expect("role_pid suffix");
        assert_eq!(role, "mpv");
        assert_eq!(pid.parse::<u32>().unwrap(), std::process::id());
    }

    /// The black-box half: build a real `Command` exactly as a call site
    /// does, read back the environment it *would* carry across the fork, and
    /// assert the property string pixelpass matches on is present in it.
    ///
    /// Deliberately inspects the `Command`, not our own formatter — a helper
    /// that returns the right string but is wired into the wrong variable
    /// would pass a formatter test and fail in the field.
    #[test]
    fn tagged_child_command_carries_both_carriers() {
        let mut command = Command::new("mpv");
        tag_child(&mut command, "mpv");

        let env: BTreeMap<String, String> = command
            .get_envs()
            .map(|(key, value)| {
                (
                    key.to_string_lossy().into_owned(),
                    value
                        .expect("tagging sets values, never removals")
                        .to_string_lossy()
                        .into_owned(),
                )
            })
            .collect();

        let expected_name = owned_node_name("mpv");
        let prop = format!("{}={}", fixture_get("prop_key"), fixture_get("prop_value"));

        let pulse = env.get(PULSE_PROP_ENV).expect("PULSE_PROP set");
        assert!(
            pulse.contains(&prop),
            "PULSE_PROP carries carrier 1: {pulse}"
        );
        assert!(
            pulse.contains(&format!("node.name={expected_name}")),
            "PULSE_PROP carries carrier 2: {pulse}"
        );

        let pw = env.get(PIPEWIRE_PROPS_ENV).expect("PIPEWIRE_PROPS set");
        assert!(
            pw.contains(&format!(
                "\"{}\" = \"{}\"",
                fixture_get("prop_key"),
                fixture_get("prop_value")
            )),
            "PIPEWIRE_PROPS carries carrier 1: {pw}"
        );
        assert!(
            pw.contains(&format!("\"node.name\" = \"{expected_name}\"")),
            "PIPEWIRE_PROPS carries carrier 2: {pw}"
        );
        assert!(
            pw.starts_with('{') && pw.ends_with('}'),
            "PIPEWIRE_PROPS is a SPA-JSON object: {pw}"
        );
    }

    #[test]
    fn every_generated_name_matches_the_prefix_pixelpass_looks_for() {
        for role in ["call", "mpv", "vlc", "notify"] {
            let name = owned_node_name(role);
            assert!(
                name.starts_with(OWNED_NODE_NAME_PREFIX),
                "{name} must match the prefix"
            );
        }
    }

    #[test]
    fn role_is_reduced_to_a_single_safe_token() {
        // A space would split PULSE_PROP into a bogus second property.
        assert_eq!(sanitize_role("my player"), "my-player");
        assert_eq!(sanitize_role("MPV"), "mpv");
        assert_eq!(sanitize_role("vlc/../.."), "vlc");
        assert_eq!(sanitize_role(""), "child");
        assert_eq!(sanitize_role("---"), "child");
        assert!(!owned_node_name("my player").contains(' '));
    }

    #[test]
    fn pulse_prop_pairs_are_whitespace_separated_and_parse_back() {
        let value = pulse_prop_value(&owned_node_name("notify"));
        let pairs: BTreeMap<&str, &str> = value
            .split_whitespace()
            .map(|pair| pair.split_once('=').expect("key=value"))
            .collect();
        assert_eq!(pairs.len(), 2, "exactly two properties: {value}");
        assert_eq!(pairs.get(OWNED_PROP_KEY), Some(&OWNED_PROP_VALUE));
        assert!(
            pairs
                .get("node.name")
                .is_some_and(|name| name.starts_with(OWNED_NODE_NAME_PREFIX))
        );
    }
}