Breadway/breadcast
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: Breadway:main
Branches Tags
Breadway:main
Breadway:v0.1.1-rc.1
Breadway:v0.1.0
..
compare: Breadway:v0.1.1-rc.1
Branches Tags
Breadway:main
Breadway:v0.1.1-rc.1
Breadway:v0.1.0

No commits in common. "main" and "v0.1.1-rc.1" have entirely different histories.
main ... v0.1.1-rc.

9 changed files with 12 additions and 292 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -35,6 +35,3 @@ CLAUDE.md
# graphify knowledge-graph output (local tool cache, not for commit)
graphify-out/
# .freebuff local tool state (not for commit)
.freebuff/

1
Cargo.lock generated
View file

@ -187,7 +187,6 @@ dependencies = [
"gstreamer",
"gstreamer-app",
"gstreamer-video",
"if-addrs 0.15.0",
"mdns-sd",
"rupnp",
"rust_cast",

3
breadcast-caststream-sys/src/facade.cc
View file

@ -465,9 +465,6 @@ 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);
}

4
breadcast-core/Cargo.toml
View file

@ -21,10 +21,6 @@ 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]

17
breadcast-core/src/dlna/session.rs
View file

@ -1,6 +1,5 @@
use anyhow::{Context, Result};
use rupnp::Service;
use std::time::Duration;
use super::device::DlnaDevice;
use super::AV_TRANSPORT;
@ -82,23 +81,9 @@ impl DlnaSession {
{
Ok(_) => {}
Err(e) => {
// 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 {
if self.transport_state().await.ok().as_deref() != Some("PLAYING") {
return Err(e).context("Play failed");
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
}
}

22
breadcast-core/src/http_server.rs
View file

@ -132,30 +132,12 @@ 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. 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| {
// 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::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[..8].copy_from_slice(&hasher.finish().to_le_bytes());
}
bytes.iter().map(|b| format!("{b:02x}")).collect()
}

43
breadcast-core/src/net.rs
View file

@ -55,26 +55,7 @@ fn pick_lan_ip(peer: Option<IpAddr>) -> Result<IpAddr> {
.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<Vec<(String, Ipv4Addr, u8)>> {
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<Vec<(String, Ipv4Addr, u8)>> {
let output = std::process::Command::new("ip")
.args(["-4", "-o", "addr", "show", "scope", "global", "up"])
.output()
@ -108,30 +89,6 @@ fn ip_command_candidates() -> Result<Vec<(String, Ipv4Addr, u8)>> {
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<Vec<(String, Ipv4Addr, u8)>> {
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;

184
breadcastd/src/daemon.rs
View file

@ -97,16 +97,9 @@ impl ActiveSession {
}
}
/// `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.
/// `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).
fn is_link_local_v6(host: &str) -> bool {
let host = host.split('%').next().unwrap_or(host);
matches!(host.parse::<std::net::IpAddr>(), Ok(std::net::IpAddr::V6(v6)) if (v6.segments()[0] & 0xffc0) == 0xfe80)
}
@ -184,23 +177,6 @@ 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 } => {
@ -360,159 +336,3 @@ 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<ServerMessage>) {
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<Result<(), String>>) -> 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::…%<iface>`, 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));
}
}

17
breadcastd/src/main.rs
View file

@ -73,17 +73,8 @@ 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");
@ -108,12 +99,8 @@ 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");