Compare commits

..
Author SHA1 Message Date
molluskandClaude Opus 4.8 47c58047ce Windows port Phase 1: real cpal/WASAPI audio backend
windows-build / windows-build (push) Has been cancelled
cargo-deny / cargo-deny (pull_request) Has been cancelled
windows-build / windows-build (pull_request) Has been cancelled
Replace the Phase 0 no-op CpalBackend stub with a working cpal backend
(WASAPI on Windows), preserving the exact PipeWire AudioBackend contract
so the mixer/encoder/jitter pipeline is unchanged.

- Capture: input stream -> downmix to mono -> 960-sample (20ms) i16 frames
  -> tx, matching the encoder/jitter frame size.
- Playback: 200ms stereo ring prefilled to PLAYBACK_TARGET_SAMPLES; the
  output callback drains it (silence on underrun) while the owning thread
  feeds it from rx. ring_fill is the exact delta-maintained occupancy
  counter (fetch_add on push, fetch_sub on pop), preserving the clock-paced
  production design (not ringbuf's stale occupied_len).
- cpal::Stream is !Send, but AudioBackend is Send+Sync and shared via Arc,
  so each stream lives on its own owning thread (built/played/dropped
  there); the struct holds only the running flag + JoinHandle. stop()
  flips the flag and joins.
- Generic over F32/I16/U16 sample formats; device selected by name else
  default; requires a native 48kHz config (clear error otherwise, no
  resampling yet). Mirrors the PipeWire drain_loop and playout-health line.
- Cargo.toml: add cpal 0.15 under cfg(windows).

Verified by temporarily compiling cpal_impl against real cpal on Linux/ALSA:
build + clippy clean, 6/6 cpal_impl unit tests pass. Reverted to windows-only
gating; shipped Linux state green (316/316). Runtime/WASAPI end-to-end is
unverified and pending a Windows host (plan M2).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 17:15:56 -04:00
molluskandClaude Opus 4.8 85b12a26c9 Windows port Phase 0: platform-select the audio backend
windows-build / windows-build (push) Has been cancelled
cargo-deny / cargo-deny (pull_request) Has been cancelled
windows-build / windows-build (pull_request) Has been cancelled
Make the tree compile for Windows without touching core logic, by
confining all Linux/PipeWire assumptions behind cfg gates and a single
platform-selected backend alias. No new dependencies — the cpal/WASAPI
backend lands in Phase 1; this ships a no-op stub.

- Cargo.toml: move pipewire + rfd(xdg-portal) under cfg(unix); add a
  cfg(windows) rfd using the Win32 dialog backend.
- audio: gate pipewire_impl to unix, add a cpal_impl stub for windows,
  and select between them via the new PlatformAudioBackend alias.
- core: use PlatformAudioBackend instead of the concrete PipeWireBackend.
- lib: gate the unix-only 0o600 log-file mode code (+ its test); Windows
  logs inherit the directory ACL.
- audio_probe: gate this PipeWire diagnostic to unix with a stub main.
- app: open URLs via rundll32 on windows, xdg-open on unix (shell-free).
- ci: add .gitea/workflows/windows-build.yml (M1) — build + lib tests for
  x86_64-pc-windows-msvc, with CMAKE_POLICY_VERSION_MINIMUM=3.5 for the
  vendored libopus build. Needs a windows act_runner to actually run.

Linux build/clippy/tests green (316/316). The Windows path is verified by
inspection only (no local Windows toolchain); CI is the real gate.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 16:47:13 -04:00
molluskandClaude Opus 4.8 e4767be210 ci: enforce cargo-deny supply-chain policy on push and PRs
Adds a Gitea Actions workflow that runs `cargo deny --locked check` on
every push to main and every PR, so the deny.toml policy (advisories,
bans, licenses, sources) is enforced automatically rather than by hand.

Runs on a locked tree so the pinned versions in Cargo.lock are what get
audited; a poisoned dependency release can't reach CI until Cargo.lock is
deliberately updated. cargo-deny is pinned to 0.19.9 via a prebuilt binary.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 16:00:36 -04:00
mollusk 10ee765ffd Merge pull request 'Security scan' (#1) from security-scan into main
Reviewed-on: #1
2026-06-18 19:57:56 +00:00
molluskandClaude Opus 4.8 3034c42f71 Add security review report for security-scan branch
Documents the focused security review of the protocol-versioning migration
and the cargo-deny policy addition. Result: no high-confidence vulnerabilities
— the versioned_topic XOR transform is entropy-preserving, signature binding
uses the raw topic_id consistently, and the ALPN/domain changes are
handshake-level compatibility only.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 15:54:35 -04:00
molluskandClaude Opus 4.8 465c7ba2b0 Add cargo-deny supply-chain policy (deny.toml)
Supersede bare cargo-audit with an enforceable four-part policy, validated
against the current tree with cargo-deny 0.19.9 (advisories/bans/licenses/
sources all pass):

- advisories: deny vulnerabilities + yanked; ignore the two *unmaintained*
  warnings (paste RUSTSEC-2024-0436, audiopus_sys RUSTSEC-2026-0150) with
  rationale. Both are transitive and pinned via Cargo.lock, so a future
  malicious release can't reach us until a deliberate cargo update.
- sources: trust only crates.io; deny unknown registries and git sources
  (core anti-hijack control).
- bans: deny wildcard version reqs; warn on duplicate versions.
- licenses: permissive allow-list covering the current graph.

Mark peerspeak publish = false (it's an application, not a published
library): blocks accidental cargo publish and lets [licenses.private]
skip the missing-license check.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 15:46:52 -04:00
molluskandClaude Opus 4.8 3ec09de87e Adopt versioning standard + migrate to versioned protocol planes (0.2.0)
Establishes VERSIONING.md: SemVer 0.x (MINOR = breaking wire change) for the
release version, and per-plane protocol versions enforced on the wire so
incompatible peers fail fast and legibly instead of via silent decode/signature
errors.

⚠️ BREAKING WIRE CHANGE — all peers must run >= 0.2.0 to interoperate (ALPNs and
gossip subscription topics changed). A pre-0.2.0 peer (e.g. an un-resynced
dopedart) can no longer connect, by design, and now fails at the handshake.

- New src/protocol.rs: single source of truth for AUDIO/FRIENDS/GOSSIP_PROTO,
  the derived ALPNs (peerspeak/audio/1, peerspeak/friends/1), GOSSIP_SIG_DOMAIN,
  and versioned_topic(). Unit tests assert ALPN/domain strings match their
  integer versions (no silent drift) + that topic namespacing is deterministic.
- Unified ALPNs: audio was b"peerspeak-audio" (unversioned, and duplicated in
  iroh_impl.rs + core/mod.rs) -> peerspeak/audio/1 from protocol.rs; friends
  re-exports protocol::FRIENDS_ALPN (was peerspeak/friends/0 -> /1).
- Gossip: subscribe to versioned_topic(ticket.topic_id) so different gossip
  versions never share a swarm; the raw topic_id stays the room identity and
  what signatures bind. GOSSIP_SIG_DOMAIN centralized into protocol.rs.
- Cargo.toml 0.1.0 -> 0.2.0.

316 lib tests / clippy --all-targets clean. VERSIONING.md documents the bump
rules, the "I changed X -> what do I bump" table, and a release checklist.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 05:20:27 -04:00
mollusk 4b8fb92dc5 Merge codex-security-s8-audio-membership: gate inbound audio by live room membership (S8) 2026-06-18 04:41:41 -04:00
molluskandClaude Opus 4.8 1adf8a97bb S8 (Pass 2): wire grace-aware audio-membership admission
Closes S8 (High): inbound audio was authenticated by identity (remote_id) but
NOT by room membership, so a former member who knew a current member's endpoint
could reconnect on the audio ALPN and inject into / eavesdrop on the mix while
invisible in the roster. Now audio is admitted only for live gossip-roster
members (the senior+user-resolved GRACE-AWARE policy).

Transport (src/network/iroh_impl.rs):
- New per-session admitted_audio: HashSet<EndpointId> on Shared (internal state,
  no wire/serialization change). Cleared on disconnect_all.
- AudioRouter::accept consults audio_sender_admitted BEFORE ensure_supervisor —
  a non-member never gets a supervisor, sender handle, datagram reader, or
  outbound mix. Brief StdMutex check, released before the await (no RT lock).
- Pure apply_audio_admission_event(roster, peer, event) with AudioAdmissionEvent
  {RosterPresent insert, TransientDropGrace no-op, Remove}. Grace deliberately
  cannot ADD membership — it only preserves an already-admitted peer — so an
  unknown peer can't sneak in via a grace event. +3 lifecycle tests (on top of
  Pass-1's 4 predicate tests).
- admit/keep_for_reconnect_grace/remove/query methods for core to drive.

Core (src/core/mod.rs) — authority is core's VERIFIED gossip-roster events, not
transport connect/disconnect:
- PeerJoined / PeerUpdated: admit_audio_sender before connect_peer.
- PeerConnectionLost: keep_audio_sender_for_reconnect_grace (preserve through the
  existing RECONNECT_GRACE window — no audio cut on transient blips).
- gossip PeerLeft, transport ConnEvent::Left, grace-timer expiry: remove_audio_sender
  before disconnect_peer + jitter removal (removal-before-teardown bounds the
  in-flight-datagram race).
- datagram receiver: audio_sender_admitted gate before any jitter buffer (defense
  in depth against a datagram racing a removal). Mixer stays off the hot path.

Mid-join: a peer who dials audio before we've verified their signed Announce is
dropped (no "pending" admission, which would reintroduce the eavesdrop); their
reconnect loop recovers once the Announce admits them.

tests/transport_loopback.rs: admit both ends before connecting, mirroring the
production room-event order.

313 lib / clippy --all-targets / transport_loopback 4 / reconnect_eviction 6 /
release — all re-run green by the senior. Former-member-rejection + mid-join
recovery are verifiable only in a 2-machine call (senior's to run).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 04:41:37 -04:00
molluskandClaude Opus 4.8 10707152a3 S8 (design-first): pure audio_sender_admitted membership seam (unwired)
Design-first checkpoint for S8 (authorize inbound audio against live room
membership). Codex's design note (in the handoff task-report.md) establishes
the authoritative roster = gossip IrohGossipState.peers, NOT the audio
transport connection list, and recommends mirroring it into an audio-admission
snapshot consulted at AudioRouter::accept + datagram ingest.

This commit lands ONLY the pure decision seam + tests; wiring is deliberately
paused for a senior decision on the reconnect-grace policy (gossip drops a peer
from the roster on transient NeighborDown, but core keeps the audio supervisor
alive for RECONNECT_GRACE — a strict roster-only gate would cut audio on blips).

- audio_sender_admitted(remote, roster) -> bool (pub(crate), #[allow(dead_code)]).
- 4 tests: member admitted, stranger rejected, former member rejected after
  roster removal, mid-join peer rejected until authenticated Announce inserts it.
- No behavior change: accept/datagram/mixer paths untouched. S8 remains OPEN.

310 lib tests / clippy --all-targets / release all green (re-run by senior).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 04:24:01 -04:00
mollusk f2e72624f7 Merge codex-security-s11-presence-discovery: honest presence/discovery state on apply failure (S11) 2026-06-18 03:32:42 -04:00
18 changed files with 1745 additions and 155 deletions
+34
View File
@@ -0,0 +1,34 @@
name: cargo-deny
# Enforce the supply-chain policy in deny.toml (advisories / bans / licenses /
# sources) on every push to main and every PR. Runs on a *locked* tree so the
# pinned, vetted versions in Cargo.lock are exactly what get audited — see the
# deny.toml header and VERSIONING.md. A new poisoned release of a dependency
# cannot reach CI until Cargo.lock is deliberately updated.
on:
push:
branches: [main]
pull_request:
jobs:
cargo-deny:
runs-on: ubuntu-latest
# rust:1 provides the cargo toolchain that cargo-deny shells out to for
# `cargo metadata`. Adjust the runner label if your act_runner uses a
# different one.
container: rust:1
steps:
- uses: actions/checkout@v4
- name: Install cargo-deny (pinned prebuilt)
run: |
set -euo pipefail
version=0.19.9
curl -sSfL \
"https://github.com/EmbarkStudios/cargo-deny/releases/download/${version}/cargo-deny-${version}-x86_64-unknown-linux-musl.tar.gz" \
| tar -xz -C /usr/local/bin --strip-components=1 --wildcards '*/cargo-deny'
cargo-deny --version
- name: cargo deny check
run: cargo deny --locked check
+85
View File
@@ -0,0 +1,85 @@
name: windows-build
# Milestone M1 of the Windows port (see docs/handoff windows-migration-plan):
# prove the tree compiles for `x86_64-pc-windows-msvc` and the unit tests pass.
# The audio backend is the Phase 0 `CpalBackend` stub for now — this job guards
# the *compile* boundary (cfg gating, platform deps, the PlatformAudioBackend
# alias) so a Unix-only assumption can't sneak back in and break Windows.
#
# RUNNER REQUIREMENT: this needs a Windows act_runner registered with the
# `windows-latest` label (the Linux `cargo-deny` job's container approach does
# NOT apply here — Windows jobs run on the host, not a Linux container). If your
# runner advertises a different label, change `runs-on` below. Until a Windows
# runner exists this workflow is simply skipped/queued, not a failure of the
# Linux CI.
#
# BUILD-HOST REQUIREMENTS (validated by the opus spike, see
# peerspeak-windows-opus-spike.md):
# - MSVC C toolchain (Visual Studio Build Tools) — to compile vendored libopus.
# - CMake on PATH — `audiopus_sys` builds libopus from source via cmake.
# - CMAKE_POLICY_VERSION_MINIMUM=3.5 (set below) — the vendored libopus declares
# an ancient `cmake_minimum_required` that CMake >= 4.0 refuses without it.
# GitHub-hosted `windows-latest` images ship MSVC + CMake; a self-hosted runner
# must provide both.
on:
push:
# `main` plus the in-progress port branches, so the Windows path is exercised
# before merge rather than only after.
branches: [main, "windows-port-**"]
pull_request:
# Allow manual runs from the Gitea Actions UI.
workflow_dispatch:
permissions:
contents: read
env:
CARGO_TERM_COLOR: always
# The vendored libopus (audiopus_sys -> cmake) uses cmake_minimum_required < 3.5,
# which CMake 4.x rejects unless this is set. See the opus spike report.
CMAKE_POLICY_VERSION_MINIMUM: "3.5"
jobs:
windows-build:
runs-on: windows-latest
steps:
- uses: actions/checkout@v4
- name: Install Rust (MSVC, pinned to repo toolchain if present)
uses: dtolnay/rust-toolchain@stable
with:
targets: x86_64-pc-windows-msvc
components: clippy
- name: Show toolchain + build prerequisites
shell: bash
run: |
set -euo pipefail
rustc --version
cargo --version
# libopus is built from source via cmake; fail early with a clear
# message if the runner lacks it rather than deep in the opus build.
if ! command -v cmake >/dev/null 2>&1; then
echo "::error::cmake not found on PATH. The opus crate builds libopus from source via cmake; install CMake on this runner."
exit 1
fi
cmake --version
# Build on a *locked* tree so the pinned, vetted Cargo.lock versions are what
# get compiled — same supply-chain stance as the cargo-deny job.
- name: Build (all targets, msvc)
run: cargo build --all-targets --locked --target x86_64-pc-windows-msvc
# Unit (lib) tests only: the `transport_loopback` integration tests stand up
# real iroh/QUIC endpoints and need working loopback networking, which isn't
# guaranteed on a CI runner. Add `--tests` here once a networked Windows
# runner is confirmed.
- name: Unit tests (lib, msvc)
run: cargo test --lib --locked --target x86_64-pc-windows-msvc
# Informational for now (not `-D warnings`): the Windows tree may surface
# platform-specific lints we haven't triaged. Tighten to deny-warnings once
# it's clean.
- name: Clippy (msvc)
run: cargo clippy --all-targets --locked --target x86_64-pc-windows-msvc
Generated
+269 -25
View File
@@ -105,6 +105,28 @@ version = "0.2.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
[[package]]
name = "alsa"
version = "0.9.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed7572b7ba83a31e20d1b48970ee402d2e3e0537dcfe0a3ff4d6eb7508617d43"
dependencies = [
"alsa-sys",
"bitflags 2.11.1",
"cfg-if",
"libc",
]
[[package]]
name = "alsa-sys"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "db8fee663d06c4e303404ef5f40488a53e062f89ba8bfed81f42325aafad1527"
dependencies = [
"libc",
"pkg-config",
]
[[package]]
name = "android-activity"
version = "0.6.1"
@@ -114,12 +136,12 @@ dependencies = [
"android-properties",
"bitflags 2.11.1",
"cc",
"jni",
"jni 0.22.4",
"libc",
"log",
"ndk",
"ndk 0.9.0",
"ndk-context",
"ndk-sys",
"ndk-sys 0.6.0+11769913",
"num_enum",
"thiserror 2.0.18",
]
@@ -712,6 +734,12 @@ dependencies = [
"shlex",
]
[[package]]
name = "cesu8"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6d43a04d8753f35258c91f8ec639f792891f748a1edbd759cf1dcea3382ad83c"
[[package]]
name = "cexpr"
version = "0.6.0"
@@ -1002,6 +1030,26 @@ dependencies = [
"libm",
]
[[package]]
name = "coreaudio-rs"
version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "321077172d79c662f64f5071a03120748d5bb652f5231570141be24cfcd2bace"
dependencies = [
"bitflags 1.3.2",
"core-foundation-sys",
"coreaudio-sys",
]
[[package]]
name = "coreaudio-sys"
version = "0.2.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b9b4739a805a62757a83e5654fa3faabec0442666b263bb2287d5a8185bfd953"
dependencies = [
"bindgen",
]
[[package]]
name = "cosmic-text"
version = "0.15.0"
@@ -1026,6 +1074,29 @@ dependencies = [
"unicode-segmentation",
]
[[package]]
name = "cpal"
version = "0.15.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "873dab07c8f743075e57f524c583985fbaf745602acbe916a01539364369a779"
dependencies = [
"alsa",
"core-foundation-sys",
"coreaudio-rs",
"dasp_sample",
"jni 0.21.1",
"js-sys",
"libc",
"mach2",
"ndk 0.8.0",
"ndk-context",
"oboe",
"wasm-bindgen",
"wasm-bindgen-futures",
"web-sys",
"windows 0.54.0",
]
[[package]]
name = "cpufeatures"
version = "0.2.17"
@@ -1228,6 +1299,12 @@ dependencies = [
"syn",
]
[[package]]
name = "dasp_sample"
version = "0.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0c87e182de0887fd5361989c677c4e8f5000cd9491d6d563161a8f3a5519fc7f"
[[package]]
name = "data-encoding"
version = "2.11.0"
@@ -2227,7 +2304,7 @@ dependencies = [
"http",
"idna",
"ipnet",
"jni",
"jni 0.22.4",
"rand 0.10.1",
"rustls",
"thiserror 2.0.18",
@@ -2247,7 +2324,7 @@ dependencies = [
"data-encoding",
"idna",
"ipnet",
"jni",
"jni 0.22.4",
"once_cell",
"prefix-trie",
"rand 0.10.1",
@@ -2270,7 +2347,7 @@ dependencies = [
"hickory-proto",
"ipconfig",
"ipnet",
"jni",
"jni 0.22.4",
"moka",
"ndk-context",
"once_cell",
@@ -3102,6 +3179,22 @@ version = "1.0.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
[[package]]
name = "jni"
version = "0.21.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1a87aa2bb7d2af34197c04845522473242e1aa17c12f4935d5856491a7fb8c97"
dependencies = [
"cesu8",
"cfg-if",
"combine",
"jni-sys 0.3.1",
"log",
"thiserror 1.0.69",
"walkdir",
"windows-sys 0.45.0",
]
[[package]]
name = "jni"
version = "0.22.4"
@@ -3463,6 +3556,15 @@ version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d3d25b0e0b648a86960ac23b7ad4abb9717601dec6f66c165f5b037f3f03065f"
[[package]]
name = "mach2"
version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d640282b302c0bb0a2a8e0233ead9035e3bed871f0b7e81fe4a1ec829765db44"
dependencies = [
"libc",
]
[[package]]
name = "malloc_buf"
version = "0.0.6"
@@ -3596,7 +3698,7 @@ dependencies = [
"dispatch",
"futures-channel",
"futures-lite",
"jni",
"jni 0.22.4",
"ndk-context",
"objc2 0.6.4",
"objc2-app-kit 0.3.2",
@@ -3695,6 +3797,20 @@ dependencies = [
"unicode-ident",
]
[[package]]
name = "ndk"
version = "0.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2076a31b7010b17a38c01907c45b945e8f11495ee4dd588309718901b1f7a5b7"
dependencies = [
"bitflags 2.11.1",
"jni-sys 0.3.1",
"log",
"ndk-sys 0.5.0+25.2.9519653",
"num_enum",
"thiserror 1.0.69",
]
[[package]]
name = "ndk"
version = "0.9.0"
@@ -3704,7 +3820,7 @@ dependencies = [
"bitflags 2.11.1",
"jni-sys 0.3.1",
"log",
"ndk-sys",
"ndk-sys 0.6.0+11769913",
"num_enum",
"raw-window-handle",
"thiserror 1.0.69",
@@ -3716,6 +3832,15 @@ version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "27b02d87554356db9e9a873add8782d4ea6e3e58ea071a9adb9a2e8ddb884a8b"
[[package]]
name = "ndk-sys"
version = "0.5.0+25.2.9519653"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8c196769dd60fd4f363e11d948139556a344e79d451aeb2fa2fd040738ef7691"
dependencies = [
"jni-sys 0.3.1",
]
[[package]]
name = "ndk-sys"
version = "0.6.0+11769913"
@@ -4466,6 +4591,29 @@ dependencies = [
"objc2-foundation 0.2.2",
]
[[package]]
name = "oboe"
version = "0.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e8b61bebd49e5d43f5f8cc7ee2891c16e0f41ec7954d36bcb6c14c5e0de867fb"
dependencies = [
"jni 0.21.1",
"ndk 0.8.0",
"ndk-context",
"num-derive",
"num-traits",
"oboe-sys",
]
[[package]]
name = "oboe-sys"
version = "0.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6c8bb09a4a2b1d668170cfe0a7d5bc103f8999fb316c98099b6a9939c9f2e79d"
dependencies = [
"cc",
]
[[package]]
name = "once_cell"
version = "1.21.4"
@@ -4594,12 +4742,13 @@ checksum = "35fb2e5f958ec131621fdd531e9fc186ed768cbe395337403ae56c17a74c68ec"
[[package]]
name = "peerspeak"
version = "0.1.0"
version = "0.2.0"
dependencies = [
"anyhow",
"async-trait",
"base64",
"bytes",
"cpal",
"dirs",
"iced",
"image",
@@ -5436,7 +5585,7 @@ checksum = "26d1e2536ce4f35f4846aa13bff16bd0ff40157cdb14cc056c7b14ba41233ba0"
dependencies = [
"core-foundation 0.10.1",
"core-foundation-sys",
"jni",
"jni 0.22.4",
"log",
"once_cell",
"rustls",
@@ -5877,7 +6026,7 @@ dependencies = [
"fastrand",
"js-sys",
"memmap2",
"ndk",
"ndk 0.9.0",
"objc2 0.6.4",
"objc2-core-foundation",
"objc2-core-graphics",
@@ -7162,7 +7311,7 @@ dependencies = [
"log",
"metal",
"naga",
"ndk-sys",
"ndk-sys 0.6.0+11769913",
"objc",
"once_cell",
"ordered-float",
@@ -7247,6 +7396,16 @@ dependencies = [
"thiserror 2.0.18",
]
[[package]]
name = "windows"
version = "0.54.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9252e5725dbed82865af151df558e754e4a3c2c30818359eb17465f1346a1b49"
dependencies = [
"windows-core 0.54.0",
"windows-targets 0.52.6",
]
[[package]]
name = "windows"
version = "0.58.0"
@@ -7254,7 +7413,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dd04d41d93c4992d421894c18c8b43496aa748dd4c081bac0dc93eb0489272b6"
dependencies = [
"windows-core 0.58.0",
"windows-targets",
"windows-targets 0.52.6",
]
[[package]]
@@ -7278,6 +7437,16 @@ dependencies = [
"windows-core 0.62.2",
]
[[package]]
name = "windows-core"
version = "0.54.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "12661b9c89351d684a50a8a643ce5f608e20243b9fb84687800163429f161d65"
dependencies = [
"windows-result 0.1.2",
"windows-targets 0.52.6",
]
[[package]]
name = "windows-core"
version = "0.58.0"
@@ -7288,7 +7457,7 @@ dependencies = [
"windows-interface 0.58.0",
"windows-result 0.2.0",
"windows-strings 0.1.0",
"windows-targets",
"windows-targets 0.52.6",
]
[[package]]
@@ -7386,13 +7555,22 @@ dependencies = [
"windows-strings 0.5.1",
]
[[package]]
name = "windows-result"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5e383302e8ec8515204254685643de10811af0ed97ea37210dc26fb0032647f8"
dependencies = [
"windows-targets 0.52.6",
]
[[package]]
name = "windows-result"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d1043d8214f791817bab27572aaa8af63732e11bf84aa21a45a78d6c317ae0e"
dependencies = [
"windows-targets",
"windows-targets 0.52.6",
]
[[package]]
@@ -7411,7 +7589,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4cd9b125c486025df0eabcb585e62173c6c9eddcec5d117d3b6e8c30e2ee4d10"
dependencies = [
"windows-result 0.2.0",
"windows-targets",
"windows-targets 0.52.6",
]
[[package]]
@@ -7423,13 +7601,22 @@ dependencies = [
"windows-link",
]
[[package]]
name = "windows-sys"
version = "0.45.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "75283be5efb2831d37ea142365f009c02ec203cd29a3ebecbc093d52315b66d0"
dependencies = [
"windows-targets 0.42.2",
]
[[package]]
name = "windows-sys"
version = "0.52.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d"
dependencies = [
"windows-targets",
"windows-targets 0.52.6",
]
[[package]]
@@ -7441,20 +7628,35 @@ dependencies = [
"windows-link",
]
[[package]]
name = "windows-targets"
version = "0.42.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8e5180c00cd44c9b1c88adb3693291f1cd93605ded80c250a75d472756b4d071"
dependencies = [
"windows_aarch64_gnullvm 0.42.2",
"windows_aarch64_msvc 0.42.2",
"windows_i686_gnu 0.42.2",
"windows_i686_msvc 0.42.2",
"windows_x86_64_gnu 0.42.2",
"windows_x86_64_gnullvm 0.42.2",
"windows_x86_64_msvc 0.42.2",
]
[[package]]
name = "windows-targets"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973"
dependencies = [
"windows_aarch64_gnullvm",
"windows_aarch64_msvc",
"windows_i686_gnu",
"windows_aarch64_gnullvm 0.52.6",
"windows_aarch64_msvc 0.52.6",
"windows_i686_gnu 0.52.6",
"windows_i686_gnullvm",
"windows_i686_msvc",
"windows_x86_64_gnu",
"windows_x86_64_gnullvm",
"windows_x86_64_msvc",
"windows_i686_msvc 0.52.6",
"windows_x86_64_gnu 0.52.6",
"windows_x86_64_gnullvm 0.52.6",
"windows_x86_64_msvc 0.52.6",
]
[[package]]
@@ -7466,18 +7668,36 @@ dependencies = [
"windows-link",
]
[[package]]
name = "windows_aarch64_gnullvm"
version = "0.42.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "597a5118570b68bc08d8d59125332c54f1ba9d9adeedeef5b99b02ba2b0698f8"
[[package]]
name = "windows_aarch64_gnullvm"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3"
[[package]]
name = "windows_aarch64_msvc"
version = "0.42.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e08e8864a60f06ef0d0ff4ba04124db8b0fb3be5776a5cd47641e942e58c4d43"
[[package]]
name = "windows_aarch64_msvc"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469"
[[package]]
name = "windows_i686_gnu"
version = "0.42.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c61d927d8da41da96a81f029489353e68739737d3beca43145c8afec9a31a84f"
[[package]]
name = "windows_i686_gnu"
version = "0.52.6"
@@ -7490,24 +7710,48 @@ version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66"
[[package]]
name = "windows_i686_msvc"
version = "0.42.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "44d840b6ec649f480a41c8d80f9c65108b92d89345dd94027bfe06ac444d1060"
[[package]]
name = "windows_i686_msvc"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66"
[[package]]
name = "windows_x86_64_gnu"
version = "0.42.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8de912b8b8feb55c064867cf047dda097f92d51efad5b491dfb98f6bbb70cb36"
[[package]]
name = "windows_x86_64_gnu"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78"
[[package]]
name = "windows_x86_64_gnullvm"
version = "0.42.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "26d41b46a36d453748aedef1486d5c7a85db22e56aff34643984ea85514e94a3"
[[package]]
name = "windows_x86_64_gnullvm"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d"
[[package]]
name = "windows_x86_64_msvc"
version = "0.42.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9aec5da331524158c6d1a4ac0ab1541149c0b9505fde06423b02f5ef0106b9f0"
[[package]]
name = "windows_x86_64_msvc"
version = "0.52.6"
@@ -7536,7 +7780,7 @@ dependencies = [
"js-sys",
"libc",
"memmap2",
"ndk",
"ndk 0.9.0",
"objc2 0.5.2",
"objc2-app-kit 0.2.2",
"objc2-foundation 0.2.2",
+27 -8
View File
@@ -1,7 +1,10 @@
[package]
name = "peerspeak"
version = "0.1.0"
version = "0.2.0"
edition = "2024"
# Application crate, not a crates.io library — refuse `cargo publish` and let
# cargo-deny's [licenses.private] skip the missing-license check.
publish = false
[lib]
name = "peerspeak"
@@ -27,17 +30,12 @@ bytes = "1.11.1"
dirs = "6.0.0"
iced = { version = "0.14.0", features = ["canvas", "image"] }
# W4 custom avatars: decode/resize an arbitrary user image (png/jpeg only to keep
# the codec surface small) and a native file picker (xdg-portal backend, no GTK).
# the codec surface small). The matching native file picker (`rfd`) is platform-
# gated below — its backend differs per OS (xdg-portal on Linux, Win32 on Windows).
image = { version = "0.25", default-features = false, features = ["png", "jpeg"] }
rfd = { version = "0.17", default-features = false, features = ["xdg-portal"] }
iroh = "1.0.0-rc.0"
iroh-gossip = "0.99.0"
opus = "0.3.1"
# v0_3_49 exposes `Buffer::requested()` (the graph's per-cycle quantum), used by
# the playback RT callback to fill exactly what the device asks for instead of
# pinning the buffer to a hard-coded 1024-frame quantum (crackle on non-1024
# hardware). The field has existed in libpipewire since 0.3.49 (2022).
pipewire = { version = "0.9", features = ["v0_3_49"] }
rand = "0.10.1"
ringbuf = "0.5.0"
serde = { version = "1.0.228", features = ["derive"] }
@@ -45,3 +43,24 @@ serde_json = "1.0.150"
thiserror = "2.0.18"
tokio = { version = "1.52.3", features = ["full"] }
tokio-stream = "0.1.18"
# --- Platform-specific dependencies -----------------------------------------
# Audio and the native file-picker backends differ per OS. Everything else in the
# app talks to the `AudioBackend` trait and the `PlatformAudioBackend` alias (see
# `src/audio/mod.rs`), so platform selection is confined to these few lines.
[target.'cfg(unix)'.dependencies]
# Linux audio backend. v0_3_49 exposes `Buffer::requested()` (the graph's per-cycle
# quantum), used by the playback RT callback to fill exactly what the device asks
# for instead of a hard-coded 1024-frame quantum (crackle on non-1024 hardware).
# The field has existed in libpipewire since 0.3.49 (2022).
pipewire = { version = "0.9", features = ["v0_3_49"] }
# Native file picker via the XDG desktop portal (no GTK) on Linux.
rfd = { version = "0.17", default-features = false, features = ["xdg-portal"] }
[target.'cfg(windows)'.dependencies]
# Native file picker using the built-in Win32 dialog backend on Windows.
rfd = { version = "0.17", default-features = false }
# Windows audio backend: cpal drives WASAPI for capture/playback behind the
# AudioBackend trait (src/audio/cpal_impl.rs). The Linux counterpart is pipewire.
cpal = "0.15"
+51
View File
@@ -0,0 +1,51 @@
# Security Review: `security-scan` branch (PeerSpeak)
_Date: 2026-06-18_
**Scope:** Protocol-versioning migration (`src/protocol.rs`, `versioned_topic`,
ALPN/domain centralization, gossip topic namespacing) and the `deny.toml`
supply-chain policy addition.
## Result: No high-confidence security vulnerabilities found.
Each plausible attack surface introduced by this branch was investigated and
confirmed safe:
### 1. `versioned_topic` XOR transform — topic secrecy preserved
`src/protocol.rs:46`, used at `src/network/gossip.rs:255`
The room `topic_id` is a uniformly random 32-byte secret (`rand::random()`,
`src/core/mod.rs:1012`) acting as the room capability. XOR-ing it with the public
constant `GOSSIP_PROTO.to_le_bytes()` cyclically is **bijective and
entropy-preserving** — the result is still uniformly random; no byte becomes
predictable and no entropy is lost. The room secret is no more recoverable by an
observer than before the change (previously the raw `topic_id` was the on-wire
topic; now it's a trivial public XOR of it). Bijectivity also preserves room
distinctness, so isolation is not weakened. **Not a vulnerability.**
### 2. Signature topic-binding — no raw/versioned confusion
`src/network/gossip.rs`
`active_topic_bytes` stores the **raw** `ticket.topic_id` (line 293), and both
`sign_gossip` and `verify_gossip` bind against that raw value. Only the
*subscribed* swarm topic (line 255) uses the versioned value. There is one swarm
per join and every peer signs/verifies against the same raw topic, so no second
topic exists to enable a raw↔versioned replay/confusion attack. Code matches
VERSIONING.md's claim. **Not a vulnerability.**
### 3. `GOSSIP_SIG_DOMAIN` — moved verbatim
Value identical (`"peerspeak-gossip-v1"`, `src/protocol.rs:34`); cross-version
cryptographic domain separation preserved. **Not a vulnerability.**
### 4. ALPN changes — handshake compatibility only
Audio `peerspeak-audio``peerspeak/audio/1`, friends `/0``/1`. No security
check keys off the old ALPN strings (audio admission is gated by live room
membership per S8, not the ALPN literal); no residual references to old strings
in non-test code. **Not a vulnerability.**
### 5. `deny.toml`
Ignores only two *unmaintained* advisories (`RUSTSEC-2024-0436`,
`RUSTSEC-2026-0150`) on compile-time/FFI-only crates — documented, and dependency
advisories are out of scope. **Not a vulnerability.**
The versioning migration is a clean, security-preserving change.
+139
View File
@@ -0,0 +1,139 @@
# PeerSpeak Versioning Standard
PeerSpeak is a full-mesh P2P voice app. Its "API contract" is not a library
surface — it is the **wire protocol** two nodes use to talk. So versioning here
tracks one question above all others:
> **Can a node on build X talk to a node on build Y?**
There are two distinct version layers. Keep them straight.
---
## Layer 1 — Release version (`Cargo.toml`)
The human-facing label you put on a build ("install this one").
**Scheme: SemVer, pre-1.0 (`0.MINOR.PATCH`).**
While we are pre-1.0 (friends-only, no stability promise yet):
| Change | Bump | Example |
| --- | --- | --- |
| **Breaking wire/protocol change** — peers on the old build can no longer interoperate; *everyone must update* | **MINOR** | `0.4.2 → 0.5.0` |
| Compatible change — bug fix, internal refactor, or a feature that does **not** change the wire (UI, local-only behavior, additive logic that old peers ignore safely) | **PATCH** | `0.4.2 → 0.4.3` |
- **Reaching `1.0.0`:** when PeerSpeak is first shared beyond the trusted-friends
circle (a "public" release), and we are willing to commit to wire stability.
After 1.0, MAJOR = wire break, MINOR = compatible feature, PATCH = fix (normal
SemVer).
- Bump `version` in `Cargo.toml` as part of the change that warrants it, in the
same commit. The number in `Cargo.toml` is the source of truth; surface it in
the UI (e.g. an About/Settings line) so a user can read their build.
**Rule of thumb:** if you find yourself writing "all peers must rebuild" or
"breaking gossip wire change" in a commit message (as S2 and W4 did), that is a
**MINOR** bump, and it must also bump the relevant protocol version in Layer 2.
---
## Layer 2 — Protocol compatibility (the one that actually breaks calls)
Wire incompatibility must **fail fast and legibly** — never as a silent
signature/decode error that looks like a bug or an attack. We achieve this by
embedding a protocol version into each transport plane, so incompatible peers
are rejected at connect/subscribe time instead of mid-conversation.
PeerSpeak has **three independent planes**, each versioned **separately** — bump
only the plane whose wire format actually changed (audio rarely changes; gossip
changes often; they must not be forced to bump together).
### ALPN naming convention
All peerspeak ALPNs use the form **`peerspeak/<plane>/<N>`** where `<N>` is that
plane's protocol version (an integer, starts at `1`). iroh refuses a connection
whose ALPN does not match exactly, so two peers on different `<N>` for a plane
simply cannot open that connection → we map that to a clean "peer is running an
incompatible version" instead of garbage.
| Plane | ALPN / mechanism | Bump when… |
| --- | --- | --- |
| **Audio** | ALPN `peerspeak/audio/<N>` | the Opus/datagram framing, sequencing, or audio-handshake changes |
| **Friends/presence** | ALPN `peerspeak/friends/<N>` | the `ControlMsg` / presence ping-pong shape changes |
| **Gossip** | *(see below — cannot use a custom ALPN)* | `GossipPayload` / `GossipMessage` / `PeerState` shape, signing, or freshness rules change |
### Gossip is special
The gossip plane runs over **iroh-gossip's own `GOSSIP_ALPN`**, which we do not
control, so we cannot version it via the ALPN. Instead, the gossip protocol
version is bound in **two** places:
1. **Topic namespacing (primary, fail-fast):** the room's `topic_id` is a random
32 bytes carried in the ticket, but the topic we actually *subscribe* to is
`protocol::versioned_topic(topic_id)` — a deterministic, dependency-free
transform that folds `GOSSIP_PROTO` into the bytes. Peers on different gossip
versions therefore derive **different subscription topics from the same ticket**
and never share a swarm — the same isolation a versioned ALPN gives the other
planes. The ticket format and the room identity (`topic_id`) are unchanged; only
the subscribed topic is namespaced. (The transform is for *isolation*, not
security — cryptographic separation is the signature domain below.)
2. **Signature domain (cryptographic separation):** the signing domain string
(`peerspeak-gossip-v<N>`, bound into every signed payload) carries the version,
so two versions that somehow met on a topic would fail each other's verification
rather than misread it.
Bumping the gossip version = bump `protocol::GOSSIP_PROTO` (drives
`versioned_topic`) **and** `protocol::GOSSIP_SIG_DOMAIN` together (a unit test in
`protocol.rs` asserts the domain string matches `GOSSIP_PROTO`, so they can't drift).
### Single source of truth for protocol versions
All protocol versions, ALPNs, the gossip signature domain, and `versioned_topic`
live in **`src/protocol.rs`**. Every call site derives from there (e.g.
`crate::protocol::AUDIO_ALPN`); **never hand-write an ALPN literal inline.** A
unit test asserts each ALPN/domain string matches its integer version so a bump
can't half-apply.
---
## "I changed X — what do I bump?" (quick reference)
| You changed… | Layer 2 (plane version) | Layer 1 (`Cargo.toml`) |
| --- | --- | --- |
| Opus framing / audio datagram layout | `peerspeak/audio/N``N+1` | MINOR |
| `ControlMsg` / presence shape | `peerspeak/friends/N``N+1` | MINOR |
| `GossipPayload`/`PeerState`/signing | `GOSSIP_PROTO_VERSION` + sig domain → next | MINOR |
| UI, local config, recording, a fix that doesn't touch any wire | nothing | PATCH |
| An *additive* gossip field that old peers safely ignore | judgement call — if old peers misbehave without it, treat as breaking (MINOR + gossip bump); if truly ignorable, PATCH | PATCH or MINOR |
When in doubt about "is this additive-safe?", assume **breaking** and bump. A
false MINOR bump costs a coordinated rebuild; a false PATCH costs silent broken
calls in the field.
---
## Release checklist (per build handed to anyone)
1. Decide MINOR vs PATCH from the table above; bump `Cargo.toml`.
2. If MINOR for a wire reason, confirm the matching Layer-2 plane version(s) were
bumped in the same change.
3. Note the version + "breaking?" in the commit / handoff.
4. Tag the commit (`v0.x.y`) so a given binary maps to a known commit.
5. Rebuild **every** peer that must interoperate (e.g. dopedart, staged friend
releases) when the bump was a MINOR/wire break.
---
## Current baseline (standard adopted + migrated, 2026-06-18, `0.2.0`)
- `Cargo.toml`: **`0.2.0`** — the MINOR bump for the (deliberately breaking)
migration to this standard. **All peers must run ≥ `0.2.0` to interoperate**
(the ALPNs and gossip topics changed); the pre-standard `0.1.0`-era build
(e.g. an un-resynced dopedart) cannot talk to a `0.2.0` peer — by design, and it
now fails cleanly at the handshake instead of silently.
- Protocol versions (all at `1`): `peerspeak/audio/1`, `peerspeak/friends/1`,
gossip `peerspeak-gossip-v1` + `versioned_topic`. All sourced from
`src/protocol.rs`.
- **Remaining nicety (not blocking):** surface `env!("CARGO_PKG_VERSION")` in the
UI (an About/Settings line) and/or log it at startup, so a running build is
self-identifying in the field. Small follow-up.
+88
View File
@@ -0,0 +1,88 @@
# cargo-deny policy for peerspeak
#
# Supersedes a bare `cargo audit` run. Enforce with:
# cargo install cargo-deny --locked
# cargo deny check
#
# In CI, run `cargo deny check` on a locked tree so the pinned, vetted
# versions in Cargo.lock are what actually get audited.
# ---------------------------------------------------------------------------
# Advisories: RustSec database. Vulnerabilities and yanked crates are denied
# by default. The two `ignore` entries below are *unmaintained* warnings only
# (no known exploit); they are deep transitive deps we cannot remove. Pinning
# them via Cargo.lock is our real protection — a future malicious release does
# not reach us until we deliberately `cargo update`, so each update is a review
# checkpoint. Revisit these if either advisory is upgraded to a vulnerability.
# ---------------------------------------------------------------------------
[advisories]
ignore = [
# paste: unmaintained, compile-time proc-macro only (zero runtime surface),
# transitive via iroh/netdev/netlink and rav1e/image/iced. Maintained fork
# `pastey` is already in the tree; stragglers will follow upstream.
"RUSTSEC-2024-0436",
# audiopus_sys: unmaintained FFI bindings to the stable libopus C library,
# pulled in via our direct `opus 0.3.1` dep. No drop-in replacement.
"RUSTSEC-2026-0150",
]
# ---------------------------------------------------------------------------
# Bans: shape of the dependency graph.
# ---------------------------------------------------------------------------
[bans]
# Multiple versions of the same crate bloat the build; warn rather than fail
# since transitive graphs (iroh, iced) routinely carry duplicates we can't fix.
multiple-versions = "warn"
# Wildcard ("*") version requirements are a supply-chain footgun: they accept
# any future release, defeating the lockfile-as-review-checkpoint model.
wildcards = "deny"
# ...but our own intra-repo path deps may use "*"; don't penalize those.
allow-wildcard-paths = true
# Crates that may never appear in the graph. Add a maintained replacement's
# predecessor here once you've migrated off it, to prevent regressions.
deny = []
# ---------------------------------------------------------------------------
# Sources: where crates are allowed to come from. This is the core anti-hijack
# control — only the official crates.io registry is trusted; arbitrary git
# sources (a common vector for slipping in unaudited code) are rejected.
# ---------------------------------------------------------------------------
[sources]
unknown-registry = "deny"
unknown-git = "deny"
allow-registry = ["https://github.com/rust-lang/crates.io-index"]
# allow-git = [] # add a specific, pinned git repo here only if ever needed
# ---------------------------------------------------------------------------
# Licenses: permissive set covering the current graph. If `cargo deny check`
# reports an unmatched license, vet it and add the SPDX id here (or add a
# per-crate entry under [licenses.exceptions]) rather than widening blindly.
# ---------------------------------------------------------------------------
[licenses]
allow = [
"MIT",
"Apache-2.0",
"Apache-2.0 WITH LLVM-exception",
"BSD-2-Clause",
"BSD-3-Clause",
"ISC",
"Zlib",
"MPL-2.0",
"Unicode-3.0",
"Unicode-DFS-2016",
"CC0-1.0",
"0BSD",
"Unlicense",
"BSL-1.0",
"NCSA", # University of Illinois/NCSA — BSD-like permissive
"CDLA-Permissive-2.0", # Community Data License Agreement, permissive
]
confidence-threshold = 0.8
exceptions = []
# peerspeak itself has no `license` field and is not published, so skip the
# "unlicensed" check for our own (private) crate. Add a license to Cargo.toml
# if/when this is ever published.
[licenses.private]
ignore = true
+22 -6
View File
@@ -1350,12 +1350,28 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
// Defence in depth: only ever hand http(s) URLs to the opener. The
// link span's href came from `linkify`, which only emits http/https,
// but re-check here so this can't be widened into launching arbitrary
// schemes/args. `xdg-open` receives the URL as a single argv entry
// (no shell), so there's no injection surface.
if (url.starts_with("http://") || url.starts_with("https://"))
&& let Err(e) = std::process::Command::new("xdg-open").arg(&url).spawn()
{
crate::log_msg(&format!("Failed to open URL {url:?}: {e}"));
// schemes/args. Each opener receives the URL as a single argv entry
// (no shell), so there's no injection surface:
// - Unix: `xdg-open <url>`.
// - Windows: `rundll32 url.dll,FileProtocolHandler <url>` — opens the
// default browser without going through `cmd`/`start`, which would
// otherwise re-parse `&` in query strings.
if url.starts_with("http://") || url.starts_with("https://") {
let spawned = {
#[cfg(unix)]
{
std::process::Command::new("xdg-open").arg(&url).spawn()
}
#[cfg(windows)]
{
std::process::Command::new("rundll32")
.args(["url.dll,FileProtocolHandler", &url])
.spawn()
}
};
if let Err(e) = spawned {
crate::log_msg(&format!("Failed to open URL {url:?}: {e}"));
}
}
}
AppMessage::ToggleMicTest(enabled) => {
+602
View File
@@ -0,0 +1,602 @@
//! Windows audio backend — cpal / WASAPI (Phase 1).
//!
//! Implements [`AudioBackend`] on top of [`cpal`], which wraps WASAPI on Windows.
//! It is the Windows counterpart to `pipewire_impl.rs` and deliberately preserves
//! the exact same contract so the rest of the app (mixer, encoder, jitter buffer)
//! is unchanged:
//!
//! - **Capture**: mono, 48 kHz, S16 PCM, emitted as `Vec<i16>` frames of
//! [`CAPTURE_FRAME`] (960 = 20 ms) samples — matching the encoder/jitter frame.
//! - **Playback**: stereo interleaved ([`PLAYBACK_CHANNELS`]) S16 PCM at 48 kHz,
//! drained from a ring buffer that is paced to the device's hardware clock via
//! `ring_fill` exactly as the PipeWire backend does.
//!
//! ## Threading and the `!Send` stream
//!
//! `cpal::Stream` is `!Send` (some backends require it to be created and dropped
//! on the same thread), but [`AudioBackend`] is `Send + Sync` and the backend is
//! shared through an `Arc`. So the stream never lives in the struct: each of
//! `start_capture`/`start_playback` spawns one owning thread that builds the
//! stream, plays it, and keeps it alive until the per-worker `running` flag flips
//! (set by `stop`). The struct holds only `Send` handles (the flag + the join
//! handle). The stream's RT callback does the actual audio work; the owning
//! thread additionally feeds the playback ring from the network mixer.
//!
//! ## Sample rate
//!
//! The whole pipeline assumes 48 kHz (Opus + the 960-sample frame). Phase 1 only
//! selects a native-48 kHz device config; if the device can't do 48 kHz we return
//! a clear error rather than silently producing pitch-shifted audio. Arbitrary
//! sample-rate support (resampling) is a Phase 1.1 follow-up.
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::Duration;
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
use cpal::{Device, FromSample, Sample, SampleFormat, SampleRate, SizedSample, Stream, StreamConfig};
use ringbuf::{
traits::{Consumer, Producer, Split},
HeapRb,
};
use super::{AudioBackend, AudioError, PLAYBACK_CHANNELS, PLAYBACK_TARGET_SAMPLES};
/// The one sample rate the pipeline supports (Opus + the 20 ms frame).
const SAMPLE_RATE: u32 = 48_000;
/// Mono capture frame: 960 samples = 20 ms @ 48 kHz. Matches the PipeWire backend
/// and `core::jitter::FRAME_SAMPLES`.
const CAPTURE_FRAME: usize = 960;
/// Playback ring capacity in interleaved samples: 200 ms of stereo @ 48 kHz.
/// Comfortably above [`PLAYBACK_TARGET_SAMPLES`] so the clock-paced producer has
/// headroom and never has to drop frames in steady state.
const RING_CAPACITY: usize = 9600 * PLAYBACK_CHANNELS;
/// How often a blocked playback worker re-checks its `running` flag, bounding how
/// long `stop()` can take to join it (mirrors the PipeWire backend's `WORKER_POLL`).
const WORKER_POLL: Duration = Duration::from_millis(100);
/// Windows audio backend. See module docs.
pub struct CpalBackend {
capture: Mutex<Option<StreamWorker>>,
playback: Mutex<Option<StreamWorker>>,
}
/// A spawned owning thread plus the flag that tells it to drop its stream and exit.
struct StreamWorker {
running: Arc<AtomicBool>,
thread: JoinHandle<()>,
}
impl CpalBackend {
pub fn new() -> Self {
Self {
capture: Mutex::new(None),
playback: Mutex::new(None),
}
}
}
impl Default for CpalBackend {
fn default() -> Self {
Self::new()
}
}
impl AudioBackend for CpalBackend {
fn start_capture(
&self,
tx: Sender<Vec<i16>>,
target_node: Option<String>,
) -> Result<(), AudioError> {
let mut guard = self.capture.lock().unwrap();
if guard.is_some() {
return Err(AudioError::Stream("Capture already started".to_string()));
}
let running = Arc::new(AtomicBool::new(true));
let running_thread = running.clone();
let thread = thread::Builder::new()
.name("peerspeak-cpal-capture".to_string())
.spawn(move || {
if let Err(e) = run_capture(tx, target_node, running_thread) {
crate::log_msg(&format!("cpal capture error: {e}"));
}
})
.map_err(|e| AudioError::Init(e.to_string()))?;
*guard = Some(StreamWorker { running, thread });
Ok(())
}
fn start_playback(
&self,
rx: Receiver<Vec<i16>>,
target_node: Option<String>,
ring_fill: Arc<AtomicUsize>,
) -> Result<(), AudioError> {
let mut guard = self.playback.lock().unwrap();
if guard.is_some() {
return Err(AudioError::Stream("Playback already started".to_string()));
}
let running = Arc::new(AtomicBool::new(true));
let running_thread = running.clone();
let thread = thread::Builder::new()
.name("peerspeak-cpal-playback".to_string())
.spawn(move || {
if let Err(e) = run_playback(rx, target_node, ring_fill, running_thread) {
crate::log_msg(&format!("cpal playback error: {e}"));
}
})
.map_err(|e| AudioError::Init(e.to_string()))?;
*guard = Some(StreamWorker { running, thread });
Ok(())
}
fn stop(&self) -> Result<(), AudioError> {
for slot in [&self.capture, &self.playback] {
if let Some(worker) = slot.lock().unwrap().take() {
worker.running.store(false, Ordering::Relaxed);
let _ = worker.thread.join();
}
}
Ok(())
}
}
// ---------------------------------------------------------------------------
// Device / config selection
// ---------------------------------------------------------------------------
/// Resolve a device (by `target` name, else the system default) and a stream
/// config running natively at [`SAMPLE_RATE`].
///
/// For output we require [`PLAYBACK_CHANNELS`] (stereo) so the interleaved ring
/// maps 1:1 to the device buffer; for input we prefer mono but accept any channel
/// count and downmix. A device with no 48 kHz config is a hard error (no
/// resampling yet — see module docs).
fn resolve(
output: bool,
target: Option<String>,
) -> Result<(Device, StreamConfig, SampleFormat), AudioError> {
let host = cpal::default_host();
let default = || {
if output {
host.default_output_device()
} else {
host.default_input_device()
}
};
let device = match target {
Some(name) => find_device_by_name(&host, output, &name).or_else(default),
None => default(),
}
.ok_or_else(|| AudioError::Device("no audio device available".to_string()))?;
let supported = choose_config(&device, output)?;
let sample_format = supported.sample_format();
let config = supported.config();
Ok((device, config, sample_format))
}
fn find_device_by_name(host: &cpal::Host, output: bool, name: &str) -> Option<Device> {
let devices = if output {
host.output_devices().ok()?
} else {
host.input_devices().ok()?
};
devices.into_iter().find(|d| d.name().is_ok_and(|n| n == name))
}
/// Pick a supported config at exactly [`SAMPLE_RATE`]. Output must be stereo;
/// input prefers mono, then any channel count (downmixed later).
fn choose_config(
device: &Device,
output: bool,
) -> Result<cpal::SupportedStreamConfig, AudioError> {
let ranges: Vec<cpal::SupportedStreamConfigRange> = if output {
device
.supported_output_configs()
.map_err(|e| AudioError::Device(e.to_string()))?
.collect()
} else {
device
.supported_input_configs()
.map_err(|e| AudioError::Device(e.to_string()))?
.collect()
};
// A range covers a sample-rate span and a fixed channel count.
let supports_48k = |r: &cpal::SupportedStreamConfigRange| {
r.min_sample_rate().0 <= SAMPLE_RATE && SAMPLE_RATE <= r.max_sample_rate().0
};
let pick = |channels: Option<u16>| {
ranges
.iter()
.find(|r| supports_48k(r) && channels.is_none_or(|c| r.channels() == c))
.cloned()
};
let chosen = if output {
pick(Some(PLAYBACK_CHANNELS as u16))
} else {
pick(Some(1)).or_else(|| pick(None))
};
chosen
.map(|r| r.with_sample_rate(SampleRate(SAMPLE_RATE)))
.ok_or_else(|| {
AudioError::Device(format!(
"device '{}' has no {SAMPLE_RATE} Hz {} config; resampling not yet implemented (Phase 1.1)",
device.name().unwrap_or_else(|_| "<unknown>".to_string()),
if output { "stereo output" } else { "input" },
))
})
}
// ---------------------------------------------------------------------------
// Capture
// ---------------------------------------------------------------------------
fn run_capture(
tx: Sender<Vec<i16>>,
target: Option<String>,
running: Arc<AtomicBool>,
) -> Result<(), AudioError> {
let (device, config, sample_format) = resolve(false, target)?;
let channels = config.channels as usize;
let stream = match sample_format {
SampleFormat::F32 => build_input::<f32>(&device, &config, tx, channels),
SampleFormat::I16 => build_input::<i16>(&device, &config, tx, channels),
SampleFormat::U16 => build_input::<u16>(&device, &config, tx, channels),
other => Err(AudioError::Stream(format!(
"unsupported capture sample format: {other:?}"
))),
}?;
stream.play().map_err(|e| AudioError::Stream(e.to_string()))?;
// The RT callback does the work; this thread just keeps `stream` alive until
// `stop()` flips the flag, at which point the stream is dropped (= stopped).
while running.load(Ordering::Relaxed) {
thread::sleep(WORKER_POLL);
}
Ok(())
}
fn build_input<T>(
device: &Device,
config: &StreamConfig,
tx: Sender<Vec<i16>>,
channels: usize,
) -> Result<Stream, AudioError>
where
T: SizedSample + Send + 'static,
i16: FromSample<T>,
{
let mut acc = FrameAccumulator::new(CAPTURE_FRAME);
let err_fn = |e| crate::log_msg(&format!("cpal capture stream error: {e}"));
device
.build_input_stream::<T, _, _>(
config,
move |data: &[T], _| {
for frame in data.chunks_exact(channels) {
let mono = downmix_to_mono(frame);
if let Some(full) = acc.push(mono) {
// Consumer gone (call ended) → stop feeding; the owning
// thread will drop the stream on `stop()`.
if tx.send(full).is_err() {
return;
}
}
}
},
err_fn,
None,
)
.map_err(|e| AudioError::Stream(e.to_string()))
}
/// Average a device frame's channels down to a single mono i16. For a 1-channel
/// device this is just the converted sample.
fn downmix_to_mono<T>(frame: &[T]) -> i16
where
T: Copy,
i16: FromSample<T>,
{
if frame.is_empty() {
return 0;
}
let sum: i32 = frame.iter().map(|&s| i16::from_sample(s) as i32).sum();
(sum / frame.len() as i32) as i16
}
/// Accumulates mono samples into fixed-size [`CAPTURE_FRAME`] frames. Pulled out
/// of the RT callback so the framing is unit-testable.
struct FrameAccumulator {
buf: Vec<i16>,
frame_len: usize,
}
impl FrameAccumulator {
fn new(frame_len: usize) -> Self {
Self {
buf: Vec::with_capacity(frame_len),
frame_len,
}
}
/// Push one sample; returns a completed frame when the buffer fills.
fn push(&mut self, sample: i16) -> Option<Vec<i16>> {
self.buf.push(sample);
if self.buf.len() == self.frame_len {
Some(std::mem::replace(
&mut self.buf,
Vec::with_capacity(self.frame_len),
))
} else {
None
}
}
}
// ---------------------------------------------------------------------------
// Playback
// ---------------------------------------------------------------------------
fn run_playback(
rx: Receiver<Vec<i16>>,
target: Option<String>,
ring_fill: Arc<AtomicUsize>,
running: Arc<AtomicBool>,
) -> Result<(), AudioError> {
let (device, config, sample_format) = resolve(true, target)?;
let rb = HeapRb::<i16>::new(RING_CAPACITY);
let (mut producer, consumer) = rb.split();
// Prefill to the steady-state depth so playout starts at target. `ring_fill`
// is an EXACT occupancy counter maintained by deltas (worker fetch_add on
// push, RT callback fetch_sub on pop) — not ringbuf's cached `occupied_len`,
// which is stale across the split halves and would lie high and starve the
// ring. See pipewire_impl.rs for the full rationale.
for _ in 0..PLAYBACK_TARGET_SAMPLES {
let _ = producer.try_push(0);
}
ring_fill.store(PLAYBACK_TARGET_SAMPLES, Ordering::Relaxed);
// Diagnostics (mirrors the PipeWire backend's playout-health line).
let underrun = Arc::new(AtomicU64::new(0));
let dropped = Arc::new(AtomicU64::new(0));
let stream = match sample_format {
SampleFormat::F32 => {
build_output::<f32, _>(&device, &config, consumer, ring_fill.clone(), underrun.clone())
}
SampleFormat::I16 => {
build_output::<i16, _>(&device, &config, consumer, ring_fill.clone(), underrun.clone())
}
SampleFormat::U16 => {
build_output::<u16, _>(&device, &config, consumer, ring_fill.clone(), underrun.clone())
}
other => Err(AudioError::Stream(format!(
"unsupported playback sample format: {other:?}"
))),
}?;
stream.play().map_err(|e| AudioError::Stream(e.to_string()))?;
let logger = spawn_health_logger(
running.clone(),
ring_fill.clone(),
underrun.clone(),
dropped.clone(),
);
// Feed the ring from the network mixer until `stop()` flips `running` or the
// sender disconnects (call ended). Clock-paced production keeps the ring near
// target, so the drop path below should never fire in steady state.
drain_loop(&rx, &running, |frame| {
if ring_fill.load(Ordering::Relaxed) + frame.len() > RING_CAPACITY {
dropped.fetch_add(1, Ordering::Relaxed);
return;
}
for &sample in &frame {
let _ = producer.try_push(sample);
}
ring_fill.fetch_add(frame.len(), Ordering::Relaxed);
});
// We're shutting down (either stop() or disconnect). Ensure the logger sees it
// even on the disconnect path, then drop the stream.
running.store(false, Ordering::Relaxed);
let _ = logger.join();
drop(stream);
Ok(())
}
fn build_output<T, C>(
device: &Device,
config: &StreamConfig,
mut consumer: C,
ring_fill: Arc<AtomicUsize>,
underrun: Arc<AtomicU64>,
) -> Result<Stream, AudioError>
where
T: SizedSample + FromSample<i16> + Send + 'static,
C: Consumer<Item = i16> + Send + 'static,
{
let err_fn = |e| crate::log_msg(&format!("cpal playback stream error: {e}"));
device
.build_output_stream::<T, _, _>(
config,
move |data: &mut [T], _| {
let (popped, starved) = fill_output(&mut consumer, data);
if starved > 0 {
underrun.fetch_add(starved, Ordering::Relaxed);
}
if popped > 0 {
// Decrement the exact occupancy by what we actually pulled
// (underruns removed nothing) so the mixer paces against the
// true ring depth.
ring_fill.fetch_sub(popped, Ordering::Relaxed);
}
},
err_fn,
None,
)
.map_err(|e| AudioError::Stream(e.to_string()))
}
/// Drain the ring into the device buffer, substituting silence on underrun.
/// Returns `(samples_popped, samples_starved)`. RT-safe (wait-free `try_pop`).
fn fill_output<T, C>(consumer: &mut C, out: &mut [T]) -> (usize, u64)
where
T: Sample + FromSample<i16>,
C: Consumer<Item = i16>,
{
let mut popped = 0usize;
let mut starved = 0u64;
for slot in out.iter_mut() {
match consumer.try_pop() {
Some(v) => {
*slot = T::from_sample(v);
popped += 1;
}
None => {
*slot = T::from_sample(0i16);
starved += 1;
}
}
}
(popped, starved)
}
/// Once-per-second playout-health line (mirrors the PipeWire backend). Quiet
/// unless a second actually glitched, or `PEERSPEAK_AUDIO_VERBOSE` is set.
fn spawn_health_logger(
running: Arc<AtomicBool>,
ring_fill: Arc<AtomicUsize>,
underrun: Arc<AtomicU64>,
dropped: Arc<AtomicU64>,
) -> JoinHandle<()> {
let verbose = std::env::var_os("PEERSPEAK_AUDIO_VERBOSE").is_some();
thread::spawn(move || {
let (mut last_u, mut last_d) = (0u64, 0u64);
while running.load(Ordering::Relaxed) {
thread::sleep(Duration::from_secs(1));
let u = underrun.load(Ordering::Relaxed);
let d = dropped.load(Ordering::Relaxed);
let fill = ring_fill.load(Ordering::Relaxed);
let (du, dd) = (u - last_u, d - last_d);
last_u = u;
last_d = d;
if verbose || du > 0 || dd > 0 {
crate::log_msg(&format!(
"playout-health: fill={fill} samples (~{}ms) | underrun +{du} samples/s (total {u}) | dropped +{dd} frames/s (total {d})",
fill / (48 * PLAYBACK_CHANNELS),
));
}
}
})
}
/// Pump frames from `rx` to `on_frame` until `running` goes false or the sender
/// disconnects. The timed receive re-checks `running` at least every
/// [`WORKER_POLL`], so `stop()` can join the worker promptly instead of hanging
/// on a parked blocking `recv()` (same A7 fix as the PipeWire backend). Pure
/// w.r.t. its inputs, so it's unit-testable.
fn drain_loop(rx: &Receiver<Vec<i16>>, running: &AtomicBool, mut on_frame: impl FnMut(Vec<i16>)) {
while running.load(Ordering::Relaxed) {
match rx.recv_timeout(WORKER_POLL) {
Ok(frame) => on_frame(frame),
Err(RecvTimeoutError::Timeout) => continue,
Err(RecvTimeoutError::Disconnected) => return,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::mpsc;
#[test]
fn downmix_averages_channels() {
assert_eq!(downmix_to_mono::<i16>(&[100, 100]), 100);
assert_eq!(downmix_to_mono::<i16>(&[100, -100]), 0);
assert_eq!(downmix_to_mono::<i16>(&[50]), 50);
assert_eq!(downmix_to_mono::<i16>(&[]), 0);
// 4-channel average rounds toward zero (integer division).
assert_eq!(downmix_to_mono::<i16>(&[10, 20, 30, 41]), 25);
}
#[test]
fn frame_accumulator_emits_full_frames() {
let mut acc = FrameAccumulator::new(3);
assert_eq!(acc.push(1), None);
assert_eq!(acc.push(2), None);
assert_eq!(acc.push(3), Some(vec![1, 2, 3]));
// Resets for the next frame.
assert_eq!(acc.push(4), None);
assert_eq!(acc.push(5), None);
assert_eq!(acc.push(6), Some(vec![4, 5, 6]));
}
#[test]
fn fill_output_pops_then_substitutes_silence() {
let rb = HeapRb::<i16>::new(8);
let (mut prod, mut cons) = rb.split();
for v in [1, 2, 3] {
prod.try_push(v).unwrap();
}
let mut out = [0i16; 5];
let (popped, starved) = fill_output(&mut cons, &mut out);
assert_eq!(popped, 3);
assert_eq!(starved, 2);
assert_eq!(out, [1, 2, 3, 0, 0]);
}
#[test]
fn drain_loop_exits_when_running_flips_even_with_sender_alive() {
let (tx, rx) = mpsc::channel::<Vec<i16>>();
let running = Arc::new(AtomicBool::new(true));
let r2 = running.clone();
let h = thread::spawn(move || drain_loop(&rx, &r2, |_| {}));
thread::sleep(Duration::from_millis(50));
running.store(false, Ordering::Relaxed);
thread::sleep(WORKER_POLL + Duration::from_millis(150));
assert!(
h.is_finished(),
"drain_loop must exit after running=false even while the sender is alive"
);
drop(tx);
h.join().unwrap();
}
#[test]
fn drain_loop_returns_on_disconnect() {
let (tx, rx) = mpsc::channel::<Vec<i16>>();
let running = Arc::new(AtomicBool::new(true));
drop(tx);
drain_loop(&rx, &running, |_| panic!("no frame should arrive"));
}
#[test]
fn drain_loop_delivers_frames() {
let (tx, rx) = mpsc::channel::<Vec<i16>>();
let running = Arc::new(AtomicBool::new(true));
let r2 = running.clone();
let got = Arc::new(Mutex::new(Vec::new()));
let g2 = got.clone();
let h = thread::spawn(move || drain_loop(&rx, &r2, |f| g2.lock().unwrap().push(f)));
tx.send(vec![1, 2, 3]).unwrap();
tx.send(vec![4, 5]).unwrap();
thread::sleep(Duration::from_millis(50));
running.store(false, Ordering::Relaxed);
drop(tx);
h.join().unwrap();
assert_eq!(*got.lock().unwrap(), vec![vec![1, 2, 3], vec![4, 5]]);
}
}
+18
View File
@@ -62,6 +62,24 @@ pub mod gate;
pub mod limiter;
pub mod multitrack;
pub mod pan;
#[cfg(unix)]
pub mod pipewire_impl;
#[cfg(windows)]
pub mod cpal_impl;
pub mod pw_cli;
pub mod recorder;
/// The audio backend implementation for the current platform.
///
/// The whole app constructs and threads this alias (via
/// `PlatformAudioBackend::new()`) rather than any concrete backend type, so
/// platform selection lives entirely here. Both implementations satisfy the
/// [`AudioBackend`] trait, which is the only interface the core talks to.
///
/// - Linux/Unix → PipeWire ([`pipewire_impl::PipeWireBackend`]).
/// - Windows → cpal/WASAPI ([`cpal_impl::CpalBackend`]); a no-op stub until the
/// Phase 1 capture/playback implementation lands.
#[cfg(unix)]
pub type PlatformAudioBackend = pipewire_impl::PipeWireBackend;
#[cfg(windows)]
pub type PlatformAudioBackend = cpal_impl::CpalBackend;
+111 -95
View File
@@ -17,105 +17,121 @@
//!
//! Run: cargo run --bin audio_probe -- [freq_hz] [seconds] [target_node]
//! e.g. cargo run --release --bin audio_probe -- 440 30
//!
//! This probe exercises the PipeWire backend directly, so it is a Unix-only tool.
//! On non-Unix targets `main` is a stub that explains the limitation.
use std::io::{BufRead, BufReader, Seek, SeekFrom};
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use std::sync::mpsc;
use std::time::Duration;
use peerspeak::audio::AudioBackend;
use peerspeak::audio::pipewire_impl::PipeWireBackend;
use peerspeak::core::jitter::FRAME_SAMPLES; // 960 mono frames = 20ms @ 48kHz
const SAMPLE_RATE: f32 = 48_000.0;
#[tokio::main]
async fn main() {
let mut args = std::env::args().skip(1);
let freq: f32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(440.0);
let secs: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(30);
let target_node: Option<String> = args.next();
// The playout-health logger is quiet in normal operation (it only logs
// glitches); ask it for the full once-per-second heartbeat so the probe can
// show the steady-state numbers.
// SAFETY: set before any playback thread starts, so no concurrent env read.
unsafe { std::env::set_var("PEERSPEAK_AUDIO_VERBOSE", "1") };
println!("audio_probe: {freq} Hz tone for {secs}s through the real playback path.");
println!("Listen for clicks/pops; watch the playout-health lines below.\n");
// Tail the app log (where playout-health lines land) to stdout in the
// background so it's all in one terminal.
spawn_log_tailer();
let backend = PipeWireBackend::new();
let (tx, rx) = mpsc::channel::<Vec<i16>>();
let ring_fill = Arc::new(AtomicUsize::new(0));
if let Err(e) = backend.start_playback(rx, target_node, ring_fill.clone()) {
eprintln!("failed to start playback: {e}");
return;
}
// Phase-continuous sine, generated one 20ms frame at a time, fill-paced
// exactly like the production mixer: only produce while the ring is below
// target, so production tracks the PipeWire hardware clock.
use std::sync::atomic::Ordering;
let deadline = tokio::time::Instant::now() + Duration::from_secs(secs);
let mut n: u64 = 0; // running sample index keeps phase continuous across frames
while tokio::time::Instant::now() < deadline {
if ring_fill.load(Ordering::Relaxed) >= peerspeak::audio::PLAYBACK_TARGET_SAMPLES {
tokio::time::sleep(Duration::from_millis(2)).await;
continue;
}
let mut frame = Vec::with_capacity(FRAME_SAMPLES * peerspeak::audio::PLAYBACK_CHANNELS);
for _ in 0..FRAME_SAMPLES {
let t = n as f32 / SAMPLE_RATE;
// 0.25 amplitude: clearly audible but not harsh.
let sample = (0.25 * i16::MAX as f32 * (2.0 * std::f32::consts::PI * freq * t).sin()) as i16;
// Stereo playback bus: duplicate the probe tone to L/R.
frame.push(sample);
frame.push(sample);
n += 1;
}
if tx.send(frame).is_err() {
eprintln!("playback channel closed early");
break;
}
}
// Let the ring drain, then stop.
tokio::time::sleep(Duration::from_millis(300)).await;
let _ = backend.stop();
println!("\naudio_probe: done.");
#[cfg(unix)]
fn main() {
unix_probe::run();
}
/// Open the app log, seek to the end, and echo new lines (the `playout-health:`
/// reports) to stdout once they appear.
fn spawn_log_tailer() {
let path = peerspeak::log_file_path();
std::thread::spawn(move || {
// Wait for the file to exist (first log_msg creates it).
let file = loop {
if let Ok(f) = std::fs::File::open(&path) {
break f;
#[cfg(not(unix))]
fn main() {
eprintln!("audio_probe is only supported on Unix builds (it drives the PipeWire backend directly).");
}
#[cfg(unix)]
mod unix_probe {
use std::io::{BufRead, BufReader, Seek, SeekFrom};
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use std::sync::mpsc;
use std::time::Duration;
use peerspeak::audio::AudioBackend;
use peerspeak::audio::pipewire_impl::PipeWireBackend;
use peerspeak::core::jitter::FRAME_SAMPLES; // 960 mono frames = 20ms @ 48kHz
const SAMPLE_RATE: f32 = 48_000.0;
#[tokio::main]
pub async fn run() {
let mut args = std::env::args().skip(1);
let freq: f32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(440.0);
let secs: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(30);
let target_node: Option<String> = args.next();
// The playout-health logger is quiet in normal operation (it only logs
// glitches); ask it for the full once-per-second heartbeat so the probe can
// show the steady-state numbers.
// SAFETY: set before any playback thread starts, so no concurrent env read.
unsafe { std::env::set_var("PEERSPEAK_AUDIO_VERBOSE", "1") };
println!("audio_probe: {freq} Hz tone for {secs}s through the real playback path.");
println!("Listen for clicks/pops; watch the playout-health lines below.\n");
// Tail the app log (where playout-health lines land) to stdout in the
// background so it's all in one terminal.
spawn_log_tailer();
let backend = PipeWireBackend::new();
let (tx, rx) = mpsc::channel::<Vec<i16>>();
let ring_fill = Arc::new(AtomicUsize::new(0));
if let Err(e) = backend.start_playback(rx, target_node, ring_fill.clone()) {
eprintln!("failed to start playback: {e}");
return;
}
// Phase-continuous sine, generated one 20ms frame at a time, fill-paced
// exactly like the production mixer: only produce while the ring is below
// target, so production tracks the PipeWire hardware clock.
use std::sync::atomic::Ordering;
let deadline = tokio::time::Instant::now() + Duration::from_secs(secs);
let mut n: u64 = 0; // running sample index keeps phase continuous across frames
while tokio::time::Instant::now() < deadline {
if ring_fill.load(Ordering::Relaxed) >= peerspeak::audio::PLAYBACK_TARGET_SAMPLES {
tokio::time::sleep(Duration::from_millis(2)).await;
continue;
}
std::thread::sleep(Duration::from_millis(100));
};
let mut reader = BufReader::new(file);
let _ = reader.seek(SeekFrom::End(0));
loop {
let mut line = String::new();
match reader.read_line(&mut line) {
Ok(0) => std::thread::sleep(Duration::from_millis(150)),
Ok(_) => {
if line.contains("playout-health:") {
print!("{line}");
}
let mut frame = Vec::with_capacity(FRAME_SAMPLES * peerspeak::audio::PLAYBACK_CHANNELS);
for _ in 0..FRAME_SAMPLES {
let t = n as f32 / SAMPLE_RATE;
// 0.25 amplitude: clearly audible but not harsh.
let sample = (0.25 * i16::MAX as f32 * (2.0 * std::f32::consts::PI * freq * t).sin()) as i16;
// Stereo playback bus: duplicate the probe tone to L/R.
frame.push(sample);
frame.push(sample);
n += 1;
}
if tx.send(frame).is_err() {
eprintln!("playback channel closed early");
break;
}
}
// Let the ring drain, then stop.
tokio::time::sleep(Duration::from_millis(300)).await;
let _ = backend.stop();
println!("\naudio_probe: done.");
}
/// Open the app log, seek to the end, and echo new lines (the `playout-health:`
/// reports) to stdout once they appear.
fn spawn_log_tailer() {
let path = peerspeak::log_file_path();
std::thread::spawn(move || {
// Wait for the file to exist (first log_msg creates it).
let file = loop {
if let Ok(f) = std::fs::File::open(&path) {
break f;
}
std::thread::sleep(Duration::from_millis(100));
};
let mut reader = BufReader::new(file);
let _ = reader.seek(SeekFrom::End(0));
loop {
let mut line = String::new();
match reader.read_line(&mut line) {
Ok(0) => std::thread::sleep(Duration::from_millis(150)),
Ok(_) => {
if line.contains("playout-health:") {
print!("{line}");
}
}
Err(_) => std::thread::sleep(Duration::from_millis(150)),
}
Err(_) => std::thread::sleep(Duration::from_millis(150)),
}
}
});
});
}
}
+14 -5
View File
@@ -1,7 +1,7 @@
pub mod messages;
pub mod jitter;
use crate::audio::{AudioBackend, pipewire_impl::PipeWireBackend};
use crate::audio::{AudioBackend, PlatformAudioBackend};
use crate::audio::eq::{Eq, EqSettings};
use crate::codec::{AudioEncoder, opus_impl::OpusEncoder};
use crate::core::jitter::{JitterBuffer, FRAME_SAMPLES};
@@ -137,6 +137,7 @@ fn arm_grace_timer(
let handle = tokio::spawn(async move {
tokio::time::sleep(grace).await;
crate::log_msg(&format!("Reconnect grace expired; evicting peer {:?}", peer_id));
transport_evict.remove_audio_sender(peer_id);
transport_evict.disconnect_peer(peer_id).await;
jitter_evict.lock().await.remove(&peer_id);
// Scrub our internal state *before* announcing the eviction, so anything
@@ -236,7 +237,7 @@ fn run_mic_monitor(
/// Stops a standalone mic monitor if one is running. MUST NOT be called while a
/// room session is active — `backend.stop()` would also tear down the call's
/// capture/playback. Monitor and session are mutually exclusive by construction.
fn stop_mic_monitor(backend: &PipeWireBackend, monitor: Option<MicMonitor>) {
fn stop_mic_monitor(backend: &PlatformAudioBackend, monitor: Option<MicMonitor>) {
if let Some(m) = monitor {
let _ = backend.stop();
let _ = m.thread.join();
@@ -370,6 +371,7 @@ impl ConnEventHandler {
// until the grace timer or the slow gossip Leave.
cancel_grace_timer(&self.grace_timers, &id);
self.seen_connected.lock().unwrap().remove(&id);
self.transport.remove_audio_sender(id);
self.transport.disconnect_peer(id).await;
self.jitter.lock().await.remove(&id);
let _ = self.ui_tx.send(UiEvent::PeerLeft { id }).await;
@@ -398,7 +400,7 @@ struct ActiveSession {
}
impl ActiveSession {
async fn shutdown(mut self, audio_backend: Arc<PipeWireBackend>) {
async fn shutdown(mut self, audio_backend: Arc<PlatformAudioBackend>) {
crate::log_msg("ActiveSession::shutdown started");
// Tear down any screen-share children first so the host stops streaming
// promptly (kill_on_drop is the backstop, but kill explicitly so viewers
@@ -582,7 +584,7 @@ async fn build_net_stack(
// report) is injected via `friends_handler`.
let router = Router::builder(endpoint.clone())
.accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone())
.accept(b"peerspeak-audio", audio_router.clone())
.accept(crate::protocol::AUDIO_ALPN, audio_router.clone())
.accept(
crate::presence_net::FRIENDS_ALPN,
crate::presence_net::FriendsProtocol::new(friends_handler),
@@ -729,7 +731,7 @@ async fn run_core_loop(
let known_peers: Arc<std::sync::Mutex<HashMap<String, HashMap<EndpointId, EndpointAddr>>>> =
Arc::new(std::sync::Mutex::new(HashMap::new()));
let audio_backend = Arc::new(PipeWireBackend::new());
let audio_backend = Arc::new(PlatformAudioBackend::new());
let is_muted = Arc::new(AtomicBool::new(false));
let is_deafened = Arc::new(AtomicBool::new(false));
@@ -1239,6 +1241,9 @@ async fn run_core_loop(
};
while let Some((from_peer, bytes)) = datagram_rx.recv().await {
if !transport_recv.audio_sender_admitted(from_peer) {
continue;
}
if !audio_datagram_len_ok(bytes.len()) {
// Malformed (< sequence header) or oversized Opus payload.
continue;
@@ -1473,6 +1478,7 @@ async fn run_core_loop(
// A (re)join means the peer is back — cancel any
// pending reconnect grace timer before re-adding it.
cancel_grace_timer(&grace_timers_events, &peer_id);
transport_events.admit_audio_sender(peer_id);
// Establish the audio connection as soon as the peer
// is known (the transport dedupes the full-mesh race).
// Hand over the full address so reconnects can dial
@@ -1524,6 +1530,7 @@ async fn run_core_loop(
{
peers.remove(&peer_id);
}
transport_events.remove_audio_sender(peer_id);
transport_events.disconnect_peer(peer_id).await;
jitter_events.lock().await.remove(&peer_id);
let _ = ui_tx_events.send(UiEvent::PeerLeft { id: peer_id }).await;
@@ -1536,6 +1543,7 @@ async fn run_core_loop(
// it. Idempotent: an ordinary mute/unmute update just
// re-records the same address.
cancel_grace_timer(&grace_timers_events, &peer_id);
transport_events.admit_audio_sender(peer_id);
transport_events.connect_peer(state.addr.clone()).await;
// Auto-heal a friend's saved address (W7) on the
// re-announce too — this is the path that catches a
@@ -1577,6 +1585,7 @@ async fn run_core_loop(
// hasn't recovered within RECONNECT_GRACE. A gossip
// rejoin (PeerJoined/PeerUpdated) or a transport
// reconnect (ConnEvent::Connected) cancels it first.
transport_events.keep_audio_sender_for_reconnect_grace(peer_id);
let _ = ui_tx_events.send(UiEvent::PeerConnecting { id: peer_id }).await;
arm_grace_timer(
&grace_timers_events,
+25 -8
View File
@@ -2,6 +2,7 @@ pub mod audio;
pub mod codec;
pub mod dsp;
pub mod network;
pub mod protocol;
pub mod core;
pub mod app;
pub mod config;
@@ -23,6 +24,9 @@ use std::path::{Path, PathBuf};
use std::sync::OnceLock;
const LOG_MAX_BYTES: u64 = 5 * 1024 * 1024;
// Owner-only log permissions are a Unix concept (mode bits); on Windows the log
// inherits the directory's default ACL. Only referenced under `cfg(unix)`.
#[cfg(unix)]
const LOG_MODE: u32 = 0o600;
/// Resolves the log file path once: `$XDG_STATE_HOME/peerspeak/peerspeak.log`
@@ -83,8 +87,6 @@ fn prepare_log_file(path: &Path) -> std::io::Result<File> {
}
fn prepare_log_file_with_limit(path: &Path, max_bytes: u64) -> std::io::Result<File> {
use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
if let Some(parent) = path.parent() {
let _ = std::fs::create_dir_all(parent);
}
@@ -97,12 +99,23 @@ fn prepare_log_file_with_limit(path: &Path, max_bytes: u64) -> std::io::Result<F
}
}
let file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.mode(LOG_MODE)
.open(path)?;
let _ = std::fs::set_permissions(path, std::fs::Permissions::from_mode(LOG_MODE));
let mut opts = std::fs::OpenOptions::new();
opts.create(true).append(true);
// The log can carry capability-bearing values (redacted, but still): keep it
// owner-only on Unix via the open mode. Windows has no mode bits; it inherits
// the directory ACL, so this hardening is Unix-only.
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
opts.mode(LOG_MODE);
}
let file = opts.open(path)?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
// Re-assert the mode in case the file pre-existed with looser perms.
let _ = std::fs::set_permissions(path, std::fs::Permissions::from_mode(LOG_MODE));
}
Ok(file)
}
@@ -125,6 +138,7 @@ pub fn log_msg(msg: &str) {
mod tests {
use super::*;
use std::io::Write;
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
fn temp_log_dir() -> PathBuf {
@@ -144,6 +158,9 @@ mod tests {
assert_eq!(redact_for_log(" "), "<redacted:empty>");
}
// Owner-only log perms are a Unix concept; on Windows the file inherits the
// directory ACL and there's no mode to assert.
#[cfg(unix)]
#[test]
fn log_file_is_created_private() {
let dir = temp_log_dir();
+6 -2
View File
@@ -12,7 +12,7 @@ use serde::{Serialize, Deserialize};
/// Domain-separation tag mixed into every signed gossip payload so a signature
/// can never be lifted out of this protocol/version into another context.
const GOSSIP_SIG_DOMAIN: &str = "peerspeak-gossip-v1";
use crate::protocol::GOSSIP_SIG_DOMAIN;
/// How far a payload's sender-stamped timestamp may differ from local time
/// before it's rejected as stale (replayed) or implausibly future. Bounds the
@@ -248,7 +248,11 @@ impl RoomState for IrohGossipState {
crate::redact_for_log(ticket_str)
));
let ticket = ticket_str.parse::<PeerSpeakTicket>()?;
let topic_id = TopicId::from_bytes(ticket.topic_id);
// Version-namespace the subscribed topic (VERSIONING.md): peers on a
// different gossip protocol version derive a different topic from the same
// ticket and never share a swarm. The raw ticket.topic_id stays the room
// identity (and what signatures bind, below).
let topic_id = TopicId::from_bytes(crate::protocol::versioned_topic(ticket.topic_id));
crate::log_msg(&format!(
"Parsed ticket. host_id={}, host_addrs={}, topic={}",
+168 -5
View File
@@ -5,11 +5,11 @@ use bytes::Bytes;
use tokio::sync::mpsc;
use tokio::sync::mpsc::Receiver;
use std::sync::{Arc, Mutex as StdMutex};
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::time::Duration;
use async_trait::async_trait;
const AUDIO_ALPN: &[u8] = b"peerspeak-audio";
use crate::protocol::AUDIO_ALPN;
/// Per-peer datagram send queue depth. Audio is real-time, so a backlog is
/// useless latency — keep it shallow and drop the oldest frame when full.
@@ -56,6 +56,10 @@ struct Shared {
/// supervisor inserts its connection when the link comes up and removes it
/// when the link dies.
live_conns: StdMutex<HashMap<EndpointId, Connection>>,
/// Core-owned audio admission snapshot for this room session. It mirrors the
/// verified gossip roster plus peers still inside reconnect grace; transport
/// connections alone never mutate this set.
admitted_audio: StdMutex<HashSet<EndpointId>>,
incoming_tx: mpsc::Sender<(EndpointId, Bytes)>,
/// Best-effort link-state notifications for the UI (connecting / connected).
conn_events_tx: mpsc::Sender<ConnEvent>,
@@ -112,6 +116,16 @@ impl Shared {
crate::log_msg(&format!("Transport: stopped supervising peer {:?}", peer_id));
}
}
fn audio_sender_admitted(&self, peer_id: EndpointId) -> bool {
let roster = self.admitted_audio.lock().unwrap();
audio_sender_admitted(peer_id, &roster)
}
fn apply_audio_admission(&self, peer_id: EndpointId, event: AudioAdmissionEvent) {
let mut roster = self.admitted_audio.lock().unwrap();
apply_audio_admission_event(&mut roster, peer_id, event);
}
}
/// Why a peer's live-link wait woke up.
@@ -135,6 +149,41 @@ fn is_graceful_leave(err: &ConnectionError) -> bool {
matches!(err, ConnectionError::ApplicationClosed(frame) if frame.error_code == VarInt::from_u32(GOODBYE_CODE))
}
/// Pure S8 membership decision: iroh already authenticated `remote` as the
/// connection's endpoint id, so audio admission is exactly live roster membership.
pub(crate) fn audio_sender_admitted(remote: EndpointId, roster: &HashSet<EndpointId>) -> bool {
roster.contains(&remote)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum AudioAdmissionEvent {
/// A signed gossip Announce/Update says the peer is in the live room roster.
RosterPresent,
/// Gossip reported a transient drop; keep admission during reconnect grace.
TransientDropGrace,
/// Graceful leave, transport Left eviction, or reconnect-grace expiry.
Remove,
}
pub(crate) fn apply_audio_admission_event(
roster: &mut HashSet<EndpointId>,
peer_id: EndpointId,
event: AudioAdmissionEvent,
) {
match event {
AudioAdmissionEvent::RosterPresent => {
roster.insert(peer_id);
}
AudioAdmissionEvent::TransientDropGrace => {
// Grace is not an authority to add membership; it only preserves an
// already-admitted peer until either rejoin or grace expiry.
}
AudioAdmissionEvent::Remove => {
roster.remove(&peer_id);
}
}
}
/// Owns a single peer's connection lifecycle for as long as the peer is in the
/// room: obtain a link, run the send/read loops, and on loss obtain a new one —
/// with capped backoff on the dialing side. The deterministic-initiator rule
@@ -333,10 +382,17 @@ impl iroh::protocol::ProtocolHandler for AudioRouter {
if shared.self_id.to_string() < peer_id.to_string() {
return Ok(());
}
if !shared.audio_sender_admitted(peer_id) {
crate::log_msg(&format!(
"Transport: rejected inbound audio from non-member {}",
crate::short_id(&peer_id.to_string())
));
return Ok(());
}
// Route the connection to this peer's supervisor (creating it if the
// inbound link beat the gossip join event). try_send keeps the
// protocol handler from ever blocking; a full queue only happens if
// links are churning, and the supervisor will get the next one.
// inbound link arrives after the signed gossip Announce admitted it).
// try_send keeps the protocol handler from ever blocking; a full queue
// only happens if links are churning, and the supervisor gets the next one.
let inbound_tx = shared.ensure_supervisor(peer_id).await;
if inbound_tx.try_send(connection).is_err() {
crate::log_msg(&format!("Transport: dropped inbound link from {:?} (queue full)", peer_id));
@@ -369,6 +425,7 @@ impl IrohTransport {
addrs: StdMutex::new(HashMap::new()),
peers: tokio::sync::Mutex::new(HashMap::new()),
live_conns: StdMutex::new(HashMap::new()),
admitted_audio: StdMutex::new(HashSet::new()),
incoming_tx,
conn_events_tx,
});
@@ -397,11 +454,35 @@ impl IrohTransport {
}
self.shared.senders.lock().unwrap().clear();
self.shared.addrs.lock().unwrap().clear();
self.shared.admitted_audio.lock().unwrap().clear();
// Give the CONNECTION_CLOSE frames a moment to flush before the caller
// shuts the endpoint/router down (the `conns` clones are still alive
// here, so the endpoint can still transmit them).
tokio::time::sleep(Duration::from_millis(150)).await;
}
/// Admit a peer to this session's audio plane. Core calls this from verified
/// gossip roster events; the transport never derives membership on its own.
pub fn admit_audio_sender(&self, peer_id: EndpointId) {
self.shared
.apply_audio_admission(peer_id, AudioAdmissionEvent::RosterPresent);
}
/// Preserve an already-admitted peer through the reconnect grace window.
pub fn keep_audio_sender_for_reconnect_grace(&self, peer_id: EndpointId) {
self.shared
.apply_audio_admission(peer_id, AudioAdmissionEvent::TransientDropGrace);
}
/// Remove a peer from audio admission before tearing down transport/jitter state.
pub fn remove_audio_sender(&self, peer_id: EndpointId) {
self.shared
.apply_audio_admission(peer_id, AudioAdmissionEvent::Remove);
}
pub fn audio_sender_admitted(&self, peer_id: EndpointId) -> bool {
self.shared.audio_sender_admitted(peer_id)
}
}
#[async_trait]
@@ -443,3 +524,85 @@ impl NetworkTransport for IrohTransport {
.ok_or_else(|| NetError::Other("Connection events already subscribed".to_string()))
}
}
#[cfg(test)]
mod tests {
use super::*;
use iroh::SecretKey;
fn endpoint_id() -> EndpointId {
SecretKey::generate().public()
}
#[test]
fn audio_sender_admission_accepts_roster_member() {
let member = endpoint_id();
let roster = HashSet::from([member]);
assert!(audio_sender_admitted(member, &roster));
}
#[test]
fn audio_sender_admission_rejects_unknown_sender() {
let member = endpoint_id();
let stranger = endpoint_id();
let roster = HashSet::from([member]);
assert!(!audio_sender_admitted(stranger, &roster));
}
#[test]
fn audio_sender_admission_rejects_former_member_after_roster_removal() {
let former = endpoint_id();
let mut roster = HashSet::from([former]);
assert!(audio_sender_admitted(former, &roster));
roster.remove(&former);
assert!(!audio_sender_admitted(former, &roster));
}
#[test]
fn audio_sender_admission_waits_for_mid_join_announce() {
let joining_peer = endpoint_id();
let mut roster = HashSet::new();
assert!(!audio_sender_admitted(joining_peer, &roster));
roster.insert(joining_peer);
assert!(audio_sender_admitted(joining_peer, &roster));
}
#[test]
fn audio_admission_lifecycle_keeps_peer_through_transient_grace() {
let peer = endpoint_id();
let mut roster = HashSet::new();
apply_audio_admission_event(&mut roster, peer, AudioAdmissionEvent::RosterPresent);
assert!(audio_sender_admitted(peer, &roster));
apply_audio_admission_event(&mut roster, peer, AudioAdmissionEvent::TransientDropGrace);
assert!(audio_sender_admitted(peer, &roster));
}
#[test]
fn audio_admission_lifecycle_does_not_add_unknown_peer_on_grace_event() {
let peer = endpoint_id();
let mut roster = HashSet::new();
apply_audio_admission_event(&mut roster, peer, AudioAdmissionEvent::TransientDropGrace);
assert!(!audio_sender_admitted(peer, &roster));
}
#[test]
fn audio_admission_lifecycle_removes_peer_on_leave_or_grace_expiry() {
let peer = endpoint_id();
let mut roster = HashSet::from([peer]);
apply_audio_admission_event(&mut roster, peer, AudioAdmissionEvent::Remove);
assert!(!audio_sender_admitted(peer, &roster));
}
}
+1 -1
View File
@@ -31,7 +31,7 @@ use std::time::Duration;
/// ALPN for the friends presence/control plane. Separate from the audio/gossip
/// ALPNs so a control dial never lands on a bare room endpoint and vice versa.
pub const FRIENDS_ALPN: &[u8] = b"peerspeak/friends/0";
pub const FRIENDS_ALPN: &[u8] = crate::protocol::FRIENDS_ALPN;
/// Upper bound on a single control message — generous for a Pong carrying a
/// member ticket (~300 chars), but rejects a peer trying to make us buffer a
+82
View File
@@ -0,0 +1,82 @@
//! Single source of truth for PeerSpeak's on-wire protocol versions and the
//! per-plane ALPNs / gossip constants derived from them.
//!
//! See `VERSIONING.md`. The rule: each transport plane is versioned independently
//! (audio rarely changes, gossip changes often), and incompatible peers must fail
//! fast — never as a silent decode/signature error. iroh refuses a mismatched
//! ALPN at the QUIC handshake, so the audio/friends planes are self-isolating;
//! gossip can't use a custom ALPN (it rides iroh-gossip's `GOSSIP_ALPN`), so its
//! version is bound into the subscribed topic ([`versioned_topic`]) and the
//! signature domain ([`GOSSIP_SIG_DOMAIN`]).
//!
//! **Never hand-write an ALPN literal elsewhere — derive it here.** Bumping a
//! plane's protocol version is a breaking wire change → also bump `Cargo.toml`
//! MINOR (see `VERSIONING.md`).
/// Audio datagram plane version (Opus framing / sequencing). Bump on any audio
/// wire change. Mirrored in [`AUDIO_ALPN`].
pub const AUDIO_PROTO: u32 = 1;
/// Friends/presence plane version (`ControlMsg` ping-pong shape). Bump on any
/// change. Mirrored in [`FRIENDS_ALPN`].
pub const FRIENDS_PROTO: u32 = 1;
/// Gossip plane version (`GossipPayload`/`GossipMessage`/`PeerState`, signing,
/// freshness). Bump on any change. Mirrored in [`GOSSIP_SIG_DOMAIN`] and folded
/// into [`versioned_topic`].
pub const GOSSIP_PROTO: u32 = 1;
/// ALPN for the audio datagram plane: `peerspeak/audio/<AUDIO_PROTO>`.
pub const AUDIO_ALPN: &[u8] = b"peerspeak/audio/1";
/// ALPN for the friends/presence plane: `peerspeak/friends/<FRIENDS_PROTO>`.
pub const FRIENDS_ALPN: &[u8] = b"peerspeak/friends/1";
/// ed25519 gossip signature domain: `peerspeak-gossip-v<GOSSIP_PROTO>`. Carries
/// the gossip protocol version into every signed payload — a version mismatch
/// fails verification (cryptographic separation between gossip versions).
pub const GOSSIP_SIG_DOMAIN: &str = "peerspeak-gossip-v1";
/// Version-namespace a room topic so peers on different gossip protocol versions
/// derive **different subscription topics from the same ticket** and therefore
/// never share a swarm — the gossip analog of a versioned ALPN. The room's raw
/// `topic_id` (random 32 bytes, carried in the ticket) is the room identity and
/// is unchanged; only the *subscribed* topic is namespaced.
///
/// Deterministic and dependency-free; bijective for a fixed version, so distinct
/// rooms stay distinct after namespacing. This transform is for *isolation*, not
/// security — cryptographic separation between versions comes from
/// [`GOSSIP_SIG_DOMAIN`].
pub fn versioned_topic(topic_id: [u8; 32]) -> [u8; 32] {
let v = GOSSIP_PROTO.to_le_bytes();
let mut out = topic_id;
for (i, b) in out.iter_mut().enumerate() {
*b ^= v[i % v.len()];
}
out
}
#[cfg(test)]
mod tests {
use super::*;
/// The ALPN/domain strings must stay in lock-step with the integer versions
/// so a version bump can't silently forget to update the wire string.
#[test]
fn alpns_match_their_proto_versions() {
assert_eq!(AUDIO_ALPN, format!("peerspeak/audio/{AUDIO_PROTO}").as_bytes());
assert_eq!(FRIENDS_ALPN, format!("peerspeak/friends/{FRIENDS_PROTO}").as_bytes());
assert_eq!(GOSSIP_SIG_DOMAIN, format!("peerspeak-gossip-v{GOSSIP_PROTO}"));
}
#[test]
fn versioned_topic_is_deterministic_and_room_distinct() {
let a = [9u8; 32];
let mut b = a;
b[5] = 10;
assert_eq!(versioned_topic(a), versioned_topic(a), "deterministic");
assert_ne!(versioned_topic(a), versioned_topic(b), "distinct rooms stay distinct");
}
#[test]
fn versioned_topic_actually_namespaces_for_current_version() {
// Guards against a no-op transform: GOSSIP_PROTO=1 must change the topic.
assert_ne!(versioned_topic([0u8; 32]), [0u8; 32]);
}
}
+3
View File
@@ -169,6 +169,9 @@ async fn loopback_sequenced_audio_reaches_peer_and_decodes() {
b.lookup.add_endpoint_info(a.endpoint.addr());
let a_id = a.endpoint.id();
let b_id = b.endpoint.id();
a.transport.admit_audio_sender(b_id);
b.transport.admit_audio_sender(a_id);
// Subscribe to incoming datagrams on B before any are sent.
let mut b_rx = b.transport.receive_datagrams().await.expect("subscribe B");