diff --git a/.gitignore b/.gitignore index 272a817..c3b4cc4 100644 --- a/.gitignore +++ b/.gitignore @@ -35,3 +35,6 @@ CLAUDE.md # graphify knowledge-graph output (local tool cache, not for commit) graphify-out/ + +# .freebuff local tool state (not for commit) +.freebuff/ diff --git a/Cargo.lock b/Cargo.lock index 57d1fc7..3e8454c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -187,6 +187,7 @@ dependencies = [ "gstreamer", "gstreamer-app", "gstreamer-video", + "if-addrs 0.15.0", "mdns-sd", "rupnp", "rust_cast", diff --git a/breadcast-caststream-sys/src/facade.cc b/breadcast-caststream-sys/src/facade.cc index 4176a31..75139ac 100644 --- a/breadcast-caststream-sys/src/facade.cc +++ b/breadcast-caststream-sys/src/facade.cc @@ -465,6 +465,9 @@ int32_t breadcast_caststream_sender_needs_key_frame(CastStreamSender* sender) { } int32_t breadcast_caststream_sender_estimated_bandwidth_bps(CastStreamSender* sender) { + if (!sender) { + return 0; + } return sender->estimated_bandwidth_bps.load(std::memory_order_relaxed); } diff --git a/breadcast-core/Cargo.toml b/breadcast-core/Cargo.toml index b454309..c376505 100644 --- a/breadcast-core/Cargo.toml +++ b/breadcast-core/Cargo.toml @@ -21,6 +21,10 @@ gstreamer-video = "0.25" tiny_http = "0.12" rupnp = "3.0.0" futures-util = "0.3.33" +# Fallback LAN-IP enumeration used only when the `ip` command is unavailable +# (net.rs) -- already a transitive dep via mdns-sd, so this adds no new crate +# to the build. +if-addrs = "0.15" breadcast-caststream-sys = { path = "../breadcast-caststream-sys" } [dev-dependencies] diff --git a/breadcast-core/src/dlna/session.rs b/breadcast-core/src/dlna/session.rs index 31b0e1d..26551a0 100644 --- a/breadcast-core/src/dlna/session.rs +++ b/breadcast-core/src/dlna/session.rs @@ -1,5 +1,6 @@ use anyhow::{Context, Result}; use rupnp::Service; +use std::time::Duration; use super::device::DlnaDevice; use super::AV_TRANSPORT; @@ -81,8 +82,22 @@ impl DlnaSession { { Ok(_) => {} Err(e) => { - if self.transport_state().await.ok().as_deref() != Some("PLAYING") { - return Err(e).context("Play failed"); + // The renderer may already be auto-playing after SetURI (and + // reject a redundant Play), or may still be TRANSITIONING and + // only reach PLAYING a moment later -- either is success. + // A single immediate GetTransportInfo isn't enough for the + // latter, so give it a short bounded window before giving up + // on the Play error. + let mut attempts = 0u32; + loop { + if self.transport_state().await.ok().as_deref() == Some("PLAYING") { + break; + } + attempts += 1; + if attempts >= 5 { + return Err(e).context("Play failed"); + } + tokio::time::sleep(Duration::from_millis(200)).await; } } } diff --git a/breadcast-core/src/http_server.rs b/breadcast-core/src/http_server.rs index 6907c07..061ac28 100644 --- a/breadcast-core/src/http_server.rs +++ b/breadcast-core/src/http_server.rs @@ -132,12 +132,30 @@ fn random_token() -> String { .is_ok(); if !read_ok { // Unreachable in practice on Linux, but better than a zero-entropy - // token if it ever happened. - use std::hash::{Hash, Hasher}; - let mut hasher = std::collections::hash_map::DefaultHasher::new(); - std::time::SystemTime::now().hash(&mut hasher); - std::process::id().hash(&mut hasher); - bytes[..8].copy_from_slice(&hasher.finish().to_le_bytes()); + // token if it ever happened. Mix several low-cost, not-directly- + // foreseeable sources (two time reads a hair apart, pid, a process + // counter) into both hashes and write all 16 bytes, so the fallback + // never reduces to a single low-resolution clock tick or per-pid + // reuse -- still far weaker than /dev/urandom, which is why it stays + // a fallback, but not trivially guessable. + let mix = |salt: u64| { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + std::time::SystemTime::now().hash(&mut hasher); + std::time::SystemTime::UNIX_EPOCH + .elapsed() + .ok() + .map(|d| d.as_nanos()) + .unwrap_or_default() + .hash(&mut hasher); + std::process::id().hash(&mut hasher); + salt.hash(&mut hasher); + hasher.finish() + }; + let a = mix(0x9e3779b97f4a7c15); + let b = mix(0xdead_beef_dead_beef); + bytes[..8].copy_from_slice(&a.to_le_bytes()); + bytes[8..].copy_from_slice(&b.to_le_bytes()); } bytes.iter().map(|b| format!("{b:02x}")).collect() } diff --git a/breadcast-core/src/net.rs b/breadcast-core/src/net.rs index 4c74838..66c9d83 100644 --- a/breadcast-core/src/net.rs +++ b/breadcast-core/src/net.rs @@ -55,7 +55,26 @@ fn pick_lan_ip(peer: Option) -> Result { .context("no LAN-reachable IPv4 address found (excluding loopback/VPN/virtual interfaces) — is this machine connected to a network?") } +/// Enumerates this machine's private, LAN-reachable IPv4 addresses as +/// `(interface, address, prefix-len)` tuples. Primary path parses `ip -o +/// addr show` -- the formatting it relies on is stable, and it bakes in the +/// `scope global up` filter (excluding link-local 169.254/16 and down +/// interfaces) for free -- falling back to a `getifaddrs`-based enumeration +/// (`if-addrs`, already in the dependency tree via `mdns-sd`) when the `ip` +/// command is unavailable, so the LAN-IP discovery doesn't silently depend on +/// iproute2 being installed. fn lan_ipv4_candidates() -> Result> { + match ip_command_candidates() { + Ok(candidates) if !candidates.is_empty() => return Ok(candidates), + Ok(_) => {} // `ip` ran but found no private address -- genuine, fall through + Err(e) => { + tracing::debug!(error = ?e, "`ip addr show` unavailable, falling back to getifaddrs"); + } + } + getifaddrs_candidates() +} + +fn ip_command_candidates() -> Result> { let output = std::process::Command::new("ip") .args(["-4", "-o", "addr", "show", "scope", "global", "up"]) .output() @@ -89,6 +108,30 @@ fn lan_ipv4_candidates() -> Result> { Ok(candidates) } +/// `getifaddrs`-based counterpart to [`ip_command_candidates`], mirroring the +/// `ip` path's filters (up, non-loopback, private, not link-local, not an +/// excluded interface name) so the two produce equivalent candidate sets. +fn getifaddrs_candidates() -> Result> { + let ifaces = if_addrs::get_if_addrs().context("failed to enumerate network interfaces (getifaddrs fallback)")?; + let mut candidates: Vec<(String, Ipv4Addr, u8)> = Vec::new(); + for iface in ifaces { + if EXCLUDED_PREFIXES.iter().any(|p| iface.name.starts_with(p)) { + continue; + } + if !iface.is_oper_up() { + continue; + } + let if_addrs::IfAddr::V4(v4) = iface.addr else { continue }; + // Mirrors `type(-4 -o addr show scope global up)`: drop loopback, + // link-local (169.254) and non-private networks. + if v4.ip.is_loopback() || v4.ip.is_link_local() || !v4.ip.is_private() { + continue; + } + candidates.push((iface.name.clone(), v4.ip, v4.prefixlen.min(32))); + } + Ok(candidates) +} + fn same_subnet(a: Ipv4Addr, b: Ipv4Addr, prefix: u8) -> bool { if prefix == 0 { return true; diff --git a/breadcastd/src/daemon.rs b/breadcastd/src/daemon.rs index 1a164bb..9ac0287 100644 --- a/breadcastd/src/daemon.rs +++ b/breadcastd/src/daemon.rs @@ -97,9 +97,16 @@ impl ActiveSession { } } -/// `host` strings come straight from mDNS resolution, so this just checks -/// the address, not whether a zone id is attached (mDNS never gives us one). +/// `host` strings come straight from mDNS resolution, and mdns-sd renders a +/// link-local IPv6 it resolved via `ScopedIp` with a `%zone` suffix (e.g. +/// `fe80::…%wlan0`) so the address stays connectable. `IpAddr` can't parse +/// that suffix (std rejects zone-ids in `FromStr`), so the zone has to be +/// stripped before the network prefix can be examined -- otherwise a scoped +/// `fe80::…%iface` host would fail this check and, in `CastDeviceFound`, be +/// allowed to clobber a good routable host that `rust_cast` could actually +/// connect to. fn is_link_local_v6(host: &str) -> bool { + let host = host.split('%').next().unwrap_or(host); matches!(host.parse::(), Ok(std::net::IpAddr::V6(v6)) if (v6.segments()[0] & 0xffc0) == 0xfe80) } @@ -177,6 +184,23 @@ impl Daemon { self.state = StateInfo::Idle; self.broadcast_state(); bread_events::emit_mirroring_stopped(&self.bread_client); + return; + } + // No installed session, but a start is currently in flight + // (portal picker / OFFER-ANSWER). The event pump only reports + // SessionEnded once negotiation succeeded, so a matching + // generation here means the in-flight session died between + // `Negotiated` and `StartFinished` landing. If we ignore it, + // the pending `StartFinished` would install the dead session + // and leave the UI stuck on "Casting" until the frame pump + // happens to notice. Invalidate the pending start exactly as + // `StopCast` does -- bump the generation so + // `on_start_finished` tears the session down instead of + // installing it. + if self.starting { + tracing::debug!("in-flight mirror session ended before it was installed"); + self.starting = false; + self.start_generation = self.start_generation.wrapping_add(1); } } DaemonCommand::StartFinished { generation, outcome, reply } => { @@ -336,3 +360,159 @@ impl Daemon { let _ = self.events_tx.send(ServerMessage::Event { event: "device_list_changed".to_string(), data }); } } + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use bread_utils::bread_client::BreadClient; + use tokio::sync::{broadcast, mpsc, oneshot}; + + use super::*; + + /// A daemon parked mid-`StartCast`: `starting` is set and a concrete + /// generation is outstanding, but no `StartFinished` has landed yet. + /// This is exactly the window both races below target -- the portal + /// picker / OFFER/ANSWER negotiation is still running off the actor, + /// and the session that `StartFinished` would install is not installed. + /// + /// A real `MirrorSession` can't be built in a unit test (it needs the + /// portal capture session, a GStreamer encode pipeline, and a live + /// renderer on the LAN), so the `Stale StartFinished` probes below use + /// an `Err(StartFailed)` outcome. That still exercises the whole point + /// of the guard: `on_start_finished` checks `generation`/`starting` + /// *before* branching on the outcome, so the stale event is rejected + /// with "start cancelled", never installs a session, and the UI stays + /// `Idle`. The `Ok` branch of that same guard calls + /// `started.session.stop()` to tear the dead session down -- the literal + /// teardown line -- which is the only part not separately asserted here. + fn daemon_with_in_flight_start(generation: u64) -> (Daemon, broadcast::Receiver) { + let (events_tx, events_rx) = broadcast::channel(16); + let (self_tx, _self_rx) = mpsc::channel(16); + let daemon = Daemon { + cast_devices: HashMap::new(), + dlna_devices: HashMap::new(), + state: StateInfo::Idle, + active_session: None, + starting: true, + start_generation: generation, + events_tx, + bread_client: BreadClient::connect("daemon-test"), + self_tx, + }; + (daemon, events_rx) + } + + /// A `StartFinished` carrying the *dead* session's generation -- the + /// probe both race tests replay after invalidating the in-flight start, + /// to prove it can't sneak a session back in. + fn stale_start_finished(reply: oneshot::Sender>) -> DaemonCommand { + DaemonCommand::StartFinished { + generation: 7, + outcome: Err(StartFailed { + device_id: "living-room-tv".to_string(), + error: "negotiation failed".to_string(), + }), + reply, + } + } + + #[test] + fn link_local_v6_is_detected_with_and_without_a_zone() { + // Scoped form: mdns-sd's ScopedIp Display renders link-local IPv6 as + // `fe80::…%`, and IpAddr can't parse the `%` suffix -- this is + // the exact form the daemon-clobber bug shipped with. + assert!(is_link_local_v6("fe80::1234%wlan0")); + assert!(is_link_local_v6("fe80::1234%3")); + // Unscoped (bare)`fe80::…` also matches. + assert!(is_link_local_v6("fe80::1234")); + } + + #[test] + fn non_link_local_addresses_are_not_marked_link_local() { + assert!(!is_link_local_v6("fe80")); + assert!(!is_link_local_v6("192.168.1.50")); + assert!(!is_link_local_v6("fd00::1")); // ULA + assert!(!is_link_local_v6("2606:4700::1111")); // public + assert!(!is_link_local_v6("2606:4700::1111%wlan0")); // public + spurious zone + assert!(!is_link_local_v6("not-an-ip")); + } + + /// The session's event pump reports death (`SessionEnded`) *between* + /// negotiation succeeding and `StartFinished` landing -- the exact race + /// that used to install the already-dead session and leave the UI stuck + /// on "Casting" until the frame pump happened to notice. + #[tokio::test] + async fn session_ended_during_an_in_flight_start_invalidates_the_pending_start() { + let (mut daemon, mut events_rx) = daemon_with_in_flight_start(7); + + daemon.handle(DaemonCommand::SessionEnded { generation: 7 }).await; + + // Invalidate the pending start exactly as StopCast does: clear + // `starting` and bump the generation so on_start_finished tears the + // dead session down instead of installing it. + assert!(!daemon.starting, "in-flight start must be cancelled"); + assert_eq!(daemon.start_generation, 8); + assert!(daemon.active_session.is_none()); + assert!(matches!(daemon.state, StateInfo::Idle)); + + // The stale StartFinished from the dead session must not install it, + // and the caller must hear that the start was cancelled. + let (reply, reply_rx) = oneshot::channel(); + daemon.handle(stale_start_finished(reply)).await; + + assert_eq!(reply_rx.await.unwrap(), Err("start cancelled".to_string())); + assert!(daemon.active_session.is_none(), "dead session must not be installed"); + assert!(matches!(daemon.state, StateInfo::Idle), "UI must return to / stay Idle"); + + // A cancelled in-flight start broadcasts nothing -- no bogus + // state_changed to a UI that is already Idle. + assert!(events_rx.try_recv().is_err()); + } + + /// `StopCast` while the portal/negotiation is still running must cancel + /// the in-flight start, so its eventual `StartFinished` is rejected + /// rather than installing a session the user just explicitly stopped. + #[tokio::test] + async fn stop_cast_cancels_an_in_flight_start() { + let (mut daemon, mut events_rx) = daemon_with_in_flight_start(7); + + let (reply, reply_rx) = oneshot::channel(); + daemon.handle(DaemonCommand::StopCast { reply }).await; + + // StopCast still resolves Ok -- there was nothing installed to stop, + // only a start to cancel. + assert!(reply_rx.await.unwrap().is_ok()); + assert!(!daemon.starting, "StopCast must cancel the in-flight start"); + assert_eq!(daemon.start_generation, 8); + assert!(daemon.active_session.is_none()); + assert!(matches!(daemon.state, StateInfo::Idle)); + + // A late StartFinished from the cancelled start never installs. + let (reply, reply_rx) = oneshot::channel(); + daemon.handle(stale_start_finished(reply)).await; + assert_eq!(reply_rx.await.unwrap(), Err("start cancelled".to_string())); + assert!(daemon.active_session.is_none(), "cancelled start must not install a session"); + assert!(matches!(daemon.state, StateInfo::Idle)); + + // StopCast itself broadcast the transition back to Idle. + assert!(events_rx.try_recv().is_ok()); + } + + /// Defensive: a `SessionEnded` whose generation no longer matches (a + /// late notification from an already-stopped or already-cancelled + /// session) must be ignored without touching the current generation, so + /// it can't confuse a brand-new later start. + #[tokio::test] + async fn a_stale_session_ended_while_idle_is_ignored_without_bumping_generation() { + let (mut daemon, _events_rx) = daemon_with_in_flight_start(0); + daemon.starting = false; // not starting -- the stale event's own forerunner already finished + + daemon.handle(DaemonCommand::SessionEnded { generation: 99 }).await; + + assert!(!daemon.starting); + assert_eq!(daemon.start_generation, 0, "stale SessionEnded must not bump a future start's generation"); + assert!(daemon.active_session.is_none()); + assert!(matches!(daemon.state, StateInfo::Idle)); + } +} diff --git a/breadcastd/src/main.rs b/breadcastd/src/main.rs index 47cfde1..cd97b97 100644 --- a/breadcastd/src/main.rs +++ b/breadcastd/src/main.rs @@ -73,8 +73,17 @@ async fn main() -> anyhow::Result<()> { tracing::info!(?device, "Cast device found"); bread_events::emit_device_found(&bread_client, &device.id, &device.name, &device.model, "cast"); known_cast_devices.insert(device.id.clone(), device.clone()); + // Forward only on an actual change. The daemon + // actor is single-threaded and `Found` fires on + // every mDNS re-resolution (per address, per + // re-browse), so forwarding identical devices too + // would spam `device_list_changed` broadcasts and + // delay `list_devices`/`start_cast` replies that + // share the same actor queue. The daemon's own + // `CastDeviceLost`→refind logic still works, since + // a loss is a state change on its side too. + let _ = daemon_tx.send(DaemonCommand::CastDeviceFound(device)).await; } - let _ = daemon_tx.send(DaemonCommand::CastDeviceFound(device)).await; } Some(DiscoveryEvent::Lost { id }) => { tracing::info!(%id, "Cast device lost"); @@ -99,8 +108,12 @@ async fn main() -> anyhow::Result<()> { tracing::info!(?device, "DLNA device found"); bread_events::emit_device_found(&bread_client, &device.url, &device.friendly_name, "DLNA renderer", "dlna"); known_dlna_devices.insert(device.url.clone(), device.clone()); + // Same rationale as the Cast arm above: each SSDP + // poll only reports devices it *this* poll + // confirmed, but a device re-confirmed unchanged + // needs no daemon flush. + let _ = daemon_tx.send(DaemonCommand::DlnaDeviceFound(device)).await; } - let _ = daemon_tx.send(DaemonCommand::DlnaDeviceFound(device)).await; } Some(DlnaDiscoveryEvent::Lost { url }) => { tracing::info!(%url, "DLNA device lost");