diff --git a/AGENTS.md b/AGENTS.md index b8d174e..394ebb4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -50,6 +50,12 @@ Keep these separate. Crossing them is the main thing to review for. volatile: each authenticated member advertises its own protocol-specific links. Build routing tables on topology changes, never per packet. Transit must not acquire a routing mutex or wait for a protocol/TUN reader. +- **Broadcast fanout belongs at local IP ingress.** Participation is local to + each network, enabled by default, persisted and advertised to authenticated + peers. Resolve one source domain and target only its willing members. Remote + delivery never triggers another fanout. Physical LAN exports must extend + explicit ingress and source/destination admission policies; never bypass + ownership checks or add broadcast flooding to encrypted transit. - **Plugins never learn reachability.** An `IpPlugin` is handed a `PacketLink` per peer and moves datagrams over it. Addresses, hole punching and relays belong to `crates/tsunagi/src/dataplane/transport/`. A plugin announcement says *who*, never @@ -168,7 +174,26 @@ kind. - Running several library instances in one process is not a test of several system processes; do not describe it as one. -## Before you open a change +## Commit every completed change + +Agents must create a Git commit before finishing any task that changes this +repository. Do not leave the user to make the commit. Inspect recent commit +messages and follow their style: a short English subject stating the change, +then a useful body explaining the problem, final behavior, architectural +decisions, compatibility/migration effects, and checks actually performed. +Write enough context for a future maintainer or LLM to understand why the code +exists. Record failed or skipped checks honestly; never claim unrun tests. + +Before committing, inspect the diff and stage only this task's changes. Preserve +unrelated user changes and never commit secrets or generated build artifacts. +If Git has no effective author name or email, set only the missing repository- +local values to `AB` and `ab@hexor.cy`. Do not override an existing identity or +change global Git configuration. Report the resulting commit hash. + +Version bumps, release tags, and pushing commits/tags belong to the user unless +they explicitly ask the agent to do them. Read-only tasks need no empty commit. + +## Required validation ```bash cargo fmt --all -- --check diff --git a/README.md b/README.md index 9c39887..ccf7efa 100644 --- a/README.md +++ b/README.md @@ -586,9 +586,28 @@ transport handle. It takes no routing mutex, walks no graph and does not parse the encrypted payload. See [routing.md](docs/routing.md) for the architecture, limits and reproducible forwarding microbenchmark. -This wire format requires all members to upgrade together (control ALPN 2, +This wire format requires all members to upgrade together (control ALPN 3, data ALPN 4); saved identities, network names, secrets and addresses survive. +### LAN game discovery + +IPv4 UDP broadcast relay is enabled by default for each network. The IP router +sends one encrypted copy to each participating peer, including through multihop +paths. It supports `255.255.255.255` and the overlay subnet's broadcast address; +received broadcasts never trigger another fanout. + +```sh +tsunagi join -n games --no-broadcast +tsunagi join -n games --broadcast +tsunagi network broadcast off +tsunagi network broadcast on +``` + +The choice persists across restarts and a plain `join`. It can be changed while +the agent runs; `status` shows it for each network. The game must send through +the Tsunagi interface. Physical LAN capture/subnet sharing is not implemented. +See [broadcast.md](docs/broadcast.md) for domain isolation and future LAN gateways. + ## How peers find each other Two different lookups are involved, and only one of them is this project's: diff --git a/crates/tsunagi-cli/src/main.rs b/crates/tsunagi-cli/src/main.rs index 4b4b2ee..455f3fa 100644 --- a/crates/tsunagi-cli/src/main.rs +++ b/crates/tsunagi-cli/src/main.rs @@ -116,6 +116,13 @@ struct NetworkArgs { /// so it can be passed on. #[derive(Debug, Args)] struct JoinArgs { + /// Enable LAN UDP broadcast relay for this network (default for new networks). + #[arg(long, conflicts_with = "no_broadcast")] + broadcast: bool, + /// Disable LAN UDP broadcast relay for this network, including after restart. + #[arg(long, conflicts_with = "broadcast")] + no_broadcast: bool, + #[command(flatten)] paths: PathArgs, @@ -138,6 +145,14 @@ struct JoinArgs { #[derive(Debug, Subcommand)] enum NetworkAction { + /// Shows or changes this network's local LAN broadcast participation. + Broadcast { + /// Network id or unique id prefix. + network: String, + /// New choice; omit to show the stored choice. + #[arg(value_enum)] + choice: Option, + }, /// Makes a network, or joins one, in the agent that is already running. /// /// The state directory belongs to one live agent, so this is how a @@ -196,6 +211,12 @@ enum NetworkAction { }, } +#[derive(Debug, Clone, Copy, clap::ValueEnum)] +enum BroadcastChoice { + On, + Off, +} + #[derive(Debug, Subcommand)] enum SecretAction { /// Prints a fresh random secret, for a network that does not exist yet. @@ -1326,6 +1347,7 @@ impl tsunagi::ipc::ReportSource for AgentControl { &self, name: String, secret: String, + broadcast: Option, ) -> tsunagi::BoxFuture<'_, Result> { Box::pin(async move { let name = NetworkName::new(&name).map_err(|err| err.to_string())?; @@ -1349,7 +1371,7 @@ impl tsunagi::ipc::ReportSource for AgentControl { let network_id = self .agent - .join_network(&name, &secret) + .join_network_with_broadcast(&name, &secret, broadcast) .await .map_err(|err| err.to_string())?; Ok(tsunagi::ipc::JoinedReport { @@ -1361,6 +1383,23 @@ impl tsunagi::ipc::ReportSource for AgentControl { }) } + fn set_broadcast( + &self, + network_id: String, + enabled: bool, + ) -> tsunagi::BoxFuture<'_, Result> { + Box::pin(async move { + let id = network_id + .parse() + .map_err(|err| format!("invalid network id: {err}"))?; + self.agent + .set_broadcast(id, enabled) + .await + .map_err(|err| err.to_string())?; + Ok(enabled) + }) + } + fn set_active( &self, network_id: String, @@ -1506,6 +1545,9 @@ async fn network_command(args: NetworkArgs) -> Result<(), Box show_networks(&paths, &socket).await, + Some(NetworkAction::Broadcast { network, choice }) => { + broadcast_command(&paths, &socket, &network, choice).await + } Some(NetworkAction::Join(args)) => join_command(args).await, Some(NetworkAction::Stop { network }) => set_active(&paths, &socket, &network, false).await, Some(NetworkAction::Start { network }) => set_active(&paths, &socket, &network, true).await, @@ -1549,7 +1591,14 @@ async fn join_command(args: JoinArgs) -> Result<(), Box> // configured and the difference is what the user needs to see. let network_id = tsunagi::identity::NetworkKeys::derive(&name, &secret).network_id(); let (standing, _) = network_context(&stored_networks(&paths), &name, network_id); - join_network(&paths, &socket, &name, secret, origin, standing).await + let broadcast = if args.no_broadcast { + Some(false) + } else if args.broadcast { + Some(true) + } else { + None + }; + join_network(&paths, &socket, &name, secret, origin, standing, broadcast).await } /// Joins a network: into the running agent if there is one. @@ -1560,13 +1609,19 @@ async fn join_network( secret: NetworkSecret, origin: SecretOrigin, standing: NetworkStanding, + broadcast: Option, ) -> Result<(), Box> { // The running agent, because a second `up` cannot have the directory // and because this way the network starts at once instead of at the // next restart. if tsunagi::ipc::is_serving(socket).await { - let report = - tsunagi::ipc::join_network(socket, name.as_str(), secret.encode().as_str()).await?; + let report = tsunagi::ipc::join_network_with_broadcast( + socket, + name.as_str(), + secret.encode().as_str(), + broadcast, + ) + .await?; // The id in full either way: it is what every other command takes, // and the shortened form in a report is for reading, not copying. match standing { @@ -1614,6 +1669,9 @@ async fn join_network( storage .upsert_network(keys.network_id(), name.clone(), secret.clone(), true) .await?; + if let Some(enabled) = broadcast { + storage.set_broadcast(keys.network_id(), enabled).await?; + } storage.release_ownership_lock(); match standing { @@ -1747,6 +1805,33 @@ async fn show_networks( /// Deliberately not a signed anything: stopping is this device being away, /// which is an ordinary condition the others already handle, and the whole /// point is that everything is still here when it comes back. +async fn broadcast_command( + paths: &StoragePaths, + socket: &std::path::Path, + wanted: &str, + choice: Option, +) -> Result<(), Box> { + let configured = stored_networks(paths); + let network = resolve_network(&configured, wanted)?; + let mut enabled = network.broadcast; + if let Some(choice) = choice { + enabled = matches!(choice, BroadcastChoice::On); + if tsunagi::ipc::is_serving(socket).await { + tsunagi::ipc::set_broadcast(socket, &network.network_id.to_string(), enabled).await?; + } else { + let storage = tsunagi::storage::Storage::open(paths)?; + storage.set_broadcast(network.network_id, enabled).await?; + } + } + println!( + "broadcast {} for `{}` ({})", + if enabled { "on" } else { "off" }, + network.name, + network.network_id + ); + Ok(()) +} + async fn set_active( paths: &StoragePaths, socket: &std::path::Path, @@ -2555,6 +2640,11 @@ fn network_section( // Only when something has: a relay that has carried nothing is not // worth a line, and one that has is worth knowing about — it is // somebody else's traffic on this device's uplink. + section.push(Row::new( + Health::Info, + "broadcast", + if network.broadcast { "on" } else { "off" }, + )); let relayed = network.relay_forwarded + network.relay_sent_via + network.relay_received_via; if relayed > 0 { section.push(Row::new( @@ -3577,6 +3667,7 @@ async fn build_report( }); NetworkReport { + broadcast: network.broadcast, name: network.name.to_string(), network_id: network.network_id.to_string(), active: matches!(network.state, tsunagi::agent::NetworkState::Active), @@ -3884,6 +3975,7 @@ mod status_tests { /// The situation that prompted this: one peer left and came back. fn network_after_a_peer_returned() -> NetworkReport { NetworkReport { + broadcast: true, name: "LAB".into(), network_id: "xa7gyz".into(), active: true, @@ -4205,6 +4297,7 @@ mod network_tests { NetworkSecret::from_bytes(&[secret.as_bytes(), &[0u8; 32]].concat()[..32]).unwrap(); let keys = NetworkKeys::derive(&name, &secret); StoredNetwork { + broadcast: true, network_id: keys.network_id(), name, secret, @@ -4373,6 +4466,7 @@ mod network_context_tests { let secret = NetworkSecret::from_bytes([seed; 32]).unwrap(); let keys = NetworkKeys::derive(&name, &secret); StoredNetwork { + broadcast: true, network_id: keys.network_id(), name, secret, diff --git a/crates/tsunagi-cli/tests/network_cli.rs b/crates/tsunagi-cli/tests/network_cli.rs index 130de82..87e9897 100644 --- a/crates/tsunagi-cli/tests/network_cli.rs +++ b/crates/tsunagi-cli/tests/network_cli.rs @@ -250,3 +250,65 @@ fn up_is_the_agent_and_takes_no_network_at_all() { ); } } +#[test] +fn broadcast_choice_is_per_network_live_persistent_and_preserved_by_rejoin() { + let agent = start_bare(0); + for args in [ + vec!["join", "-n", "broadcast-default"], + vec!["join", "-n", "broadcast-off", "--no-broadcast"], + ] { + let result = agent.run(&args); + assert!( + result.status.success(), + "{}", + String::from_utf8_lossy(&result.stderr) + ); + } + let read = || { + tsunagi::storage::StateStore::open(agent.dir.path().join("state/state.sqlite")) + .unwrap() + .list_networks() + .unwrap() + }; + let networks = read(); + let enabled = networks + .iter() + .find(|n| n.name.as_str() == "broadcast-default") + .unwrap(); + let disabled = networks + .iter() + .find(|n| n.name.as_str() == "broadcast-off") + .unwrap(); + assert!(enabled.broadcast); + assert!(!disabled.broadcast); + assert!(agent.run(&["join", "-n", "broadcast-off"]).status.success()); + assert!( + !read() + .iter() + .find(|n| n.name.as_str() == "broadcast-off") + .unwrap() + .broadcast + ); + let id = disabled.network_id.to_string(); + let change = agent.run(&["network", "broadcast", &id, "on"]); + assert!( + change.status.success(), + "{}", + String::from_utf8_lossy(&change.stderr) + ); + assert!(read().iter().all(|n| n.broadcast)); + let change = agent.run(&["join", "-n", "broadcast-off", "--no-broadcast"]); + assert!( + change.status.success(), + "{}", + String::from_utf8_lossy(&change.stderr) + ); + let status = agent.run(&["network", "broadcast", &id]); + assert!(String::from_utf8_lossy(&status.stdout).contains("broadcast off")); + assert!( + !agent + .run(&["join", "-n", "conflict", "--broadcast", "--no-broadcast"]) + .status + .success() + ); +} diff --git a/crates/tsunagi-wg-quic/tests/local_control.rs b/crates/tsunagi-wg-quic/tests/local_control.rs index 5ca3df4..8b193e4 100644 --- a/crates/tsunagi-wg-quic/tests/local_control.rs +++ b/crates/tsunagi-wg-quic/tests/local_control.rs @@ -247,6 +247,7 @@ impl tsunagi::ipc::ReportSource for Control { &self, name: String, secret: String, + _broadcast: Option, ) -> BoxFuture<'_, Result> { Box::pin(async move { let name = tsunagi::identity::NetworkName::new(&name).map_err(|e| e.to_string())?; diff --git a/crates/tsunagi-wg-quic/tests/wireguard.rs b/crates/tsunagi-wg-quic/tests/wireguard.rs index 98010fd..d2598dc 100644 --- a/crates/tsunagi-wg-quic/tests/wireguard.rs +++ b/crates/tsunagi-wg-quic/tests/wireguard.rs @@ -1627,3 +1627,120 @@ async fn a_chain_routes_through_two_transit_peers_without_touching_their_tuns() agent.shutdown().await; } } +#[tokio::test] +async fn lan_broadcast_fanout_is_scoped_opt_in_and_does_not_reflood() { + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("lan-broadcast"); + let cuts = Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())); + let a = WgAgent::spawn_cut_off(&discovery, "bca", cuts.clone()).await; + let b = WgAgent::spawn(&discovery, "bcb").await; + let c = WgAgent::spawn(&discovery, "bcc").await; + cuts.lock().unwrap().insert(c.endpoint_id()); + let id = a.agent.join_network(&name, &secret).await.unwrap(); + b.agent.join_network(&name, &secret).await.unwrap(); + c.agent.join_network(&name, &secret).await.unwrap(); + for agent in [&a, &b, &c] { + agent.wait_for_tunnels(id, 2).await; + } + let addr = a.overlay(id).await; + let at = a.tun(id).await; + let bt = b.tun(id).await; + let ct = c.tun(id).await; + // Valid UDP discovery queries with game-like ports and an unchanged body. + fn query(source: std::net::Ipv4Addr, destination: std::net::Ipv4Addr) -> Bytes { + let mut bytes = vec![0u8; 36]; + bytes[0] = 0x45; + bytes[2..4].copy_from_slice(&36u16.to_be_bytes()); + bytes[8] = 1; + bytes[9] = 17; + bytes[12..16].copy_from_slice(&source.octets()); + bytes[16..20].copy_from_slice(&destination.octets()); + bytes[20..22].copy_from_slice(&27015u16.to_be_bytes()); + bytes[22..24].copy_from_slice(&6112u16.to_be_bytes()); + bytes[24..26].copy_from_slice(&16u16.to_be_bytes()); + bytes[28..].copy_from_slice(b"LAN GAME"); + // UDP checksum zero is valid for IPv4. + let sum: u32 = bytes[..20] + .chunks_exact(2) + .map(|w| u16::from_be_bytes([w[0], w[1]]) as u32) + .sum(); + let checksum = !((sum & 0xffff) + (sum >> 16)) as u16; + bytes[10..12].copy_from_slice(&checksum.to_be_bytes()); + Bytes::from(bytes) + } + let range = a.agent.network_status(id).await.unwrap().range.unwrap(); + let directed = std::net::Ipv4Addr::from(u32::from(range.base) | (u32::MAX >> range.prefix_len)); + for destination in [std::net::Ipv4Addr::BROADCAST, directed] { + let packet = query(addr, destination); + at.push_from_os(packet.clone()); + for tun in [&bt, &ct] { + assert_eq!( + tokio::time::timeout(tsunagi::testing::DEADLINE, tun.pop_to_os()) + .await + .unwrap() + .unwrap(), + packet + ); + assert!( + tokio::time::timeout(Duration::from_millis(100), tun.pop_to_os()) + .await + .is_err(), + "each recipient gets one copy" + ); + } + assert!( + tokio::time::timeout(Duration::from_millis(100), at.pop_to_os()) + .await + .is_err(), + "never reflect to the origin" + ); + } + c.agent.set_broadcast(id, false).await.unwrap(); + wait_until("broadcast opt-out reaches sender", || async { + a.agent + .network_status(id) + .await + .ok()? + .peers + .iter() + .find(|peer| peer.endpoint_id == c.endpoint_id()) + .filter(|peer| !peer.broadcast) + .map(|_| ()) + }) + .await; + let packet = query(addr, std::net::Ipv4Addr::BROADCAST); + at.push_from_os(packet.clone()); + assert_eq!( + tokio::time::timeout(tsunagi::testing::DEADLINE, bt.pop_to_os()) + .await + .unwrap() + .unwrap(), + packet + ); + assert!( + tokio::time::timeout(Duration::from_millis(150), ct.pop_to_os()) + .await + .is_err() + ); + // Disabling origin participation also stops outgoing copies. + a.agent.set_broadcast(id, false).await.unwrap(); + at.push_from_os(packet); + assert!( + tokio::time::timeout(Duration::from_millis(150), bt.pop_to_os()) + .await + .is_err() + ); + // Discovery's unicast reply still crosses the same multihop route. + let reply = tcp_packet(c.overlay(id).await, addr, 1280); + ct.push_from_os(reply.clone()); + assert_eq!( + tokio::time::timeout(tsunagi::testing::DEADLINE, at.pop_to_os()) + .await + .unwrap() + .unwrap(), + reply + ); + for agent in [a, b, c] { + agent.shutdown().await; + } +} diff --git a/crates/tsunagi/src/agent/mod.rs b/crates/tsunagi/src/agent/mod.rs index e2bfca6..a5473ea 100644 --- a/crates/tsunagi/src/agent/mod.rs +++ b/crates/tsunagi/src/agent/mod.rs @@ -58,6 +58,8 @@ use shutdown::Shutdown; /// Summary of a configured network, whether or not it is running. #[derive(Debug, Clone)] pub struct ConfiguredNetwork { + /// Saved local broadcast participation. + pub broadcast: bool, /// Public network identifier. pub network_id: NetworkId, /// Network name. @@ -361,6 +363,7 @@ impl Agent { }; let reserve = |wanted: Ipv4Range| { let reservation = crate::overlay::NetworkRoutes { + broadcast: Default::default(), range: Some(wanted), local: None, peers: Vec::new(), @@ -488,6 +491,16 @@ impl Agent { &self, name: &NetworkName, secret: &NetworkSecret, + ) -> Result { + self.join_network_with_broadcast(name, secret, None).await + } + + /// Joins with an explicit broadcast choice; absent preserves saved policy. + pub async fn join_network_with_broadcast( + &self, + name: &NetworkName, + secret: &NetworkSecret, + broadcast: Option, ) -> Result { let keys = NetworkKeys::derive(name, secret); let network_id = keys.network_id(); @@ -516,6 +529,9 @@ impl Agent { .storage .upsert_network(network_id, name.clone(), secret.clone(), true) .await?; + if let Some(enabled) = broadcast { + self.set_broadcast(network_id, enabled).await?; + } match self.activate_with_keys(keys).await { // Already a member of exactly this network space: nothing to do. Ok(()) | Err(Error::NetworkAlreadyActive(_)) => Ok(network_id), @@ -523,6 +539,21 @@ impl Agent { } } + /// Changes a network's local broadcast policy now and after restart. + pub async fn set_broadcast(&self, network_id: NetworkId, enabled: bool) -> Result<()> { + self.inner + .storage + .set_broadcast(network_id, enabled) + .await?; + if self.is_active(network_id).await { + let (reply, receive) = oneshot::channel(); + self.command(network_id, NetCommand::SetBroadcast { enabled, reply }) + .await?; + receive.await.map_err(|_| Error::Stopped)?; + } + Ok(()) + } + /// Activates a configured network that is currently inactive. /// /// Fails with [`Error::NetworkAlreadyActive`] if it is already running. @@ -567,7 +598,16 @@ impl Agent { as Arc, ) }; + let broadcast = self + .inner + .storage + .list_networks() + .await? + .into_iter() + .find(|stored| stored.network_id == network_id) + .is_none_or(|stored| stored.broadcast); let handle = network::spawn(RuntimeParams { + broadcast, keys, adapter: self.inner.adapter.clone(), storage: self.inner.storage.clone(), @@ -703,6 +743,7 @@ impl Agent { .await? .into_iter() .map(|stored| ConfiguredNetwork { + broadcast: stored.broadcast, network_id: stored.network_id, name: stored.name, auto_start: stored.auto_start, @@ -777,6 +818,7 @@ impl Agent { } let keys = NetworkKeys::derive(&stored.name, &stored.secret); networks.push(NetworkStatus { + broadcast: stored.broadcast, descriptor: keys.descriptor(), name: stored.name, network_id: stored.network_id, diff --git a/crates/tsunagi/src/agent/network.rs b/crates/tsunagi/src/agent/network.rs index e2222b9..1b3efdd 100644 --- a/crates/tsunagi/src/agent/network.rs +++ b/crates/tsunagi/src/agent/network.rs @@ -62,6 +62,10 @@ pub(crate) enum NetCommand { reply: oneshot::Sender>, }, Recheck, + SetBroadcast { + enabled: bool, + reply: oneshot::Sender<()>, + }, /// Resend this agent's announcement to every peer of this network. Reannounce, /// Answer to a different name from now on. @@ -95,6 +99,7 @@ impl std::fmt::Debug for NetCommand { NetCommand::Broadcast { message, .. } => write!(f, "Broadcast({})", kind(message)), NetCommand::Status { .. } => f.write_str("Status"), NetCommand::Recheck => f.write_str("Recheck"), + NetCommand::SetBroadcast { enabled, .. } => write!(f, "SetBroadcast({enabled})"), NetCommand::Reannounce => f.write_str("Reannounce"), NetCommand::SetHostname(_) => f.write_str("SetHostname"), NetCommand::Release { .. } => f.write_str("Release"), @@ -122,6 +127,7 @@ impl NetworkHandle { /// Everything a network runtime needs to run. pub(crate) struct RuntimeParams { + pub(crate) broadcast: bool, pub(crate) keys: NetworkKeys, pub(crate) adapter: EndpointAdapter, pub(crate) storage: Storage, @@ -455,6 +461,12 @@ impl Runtime { NetCommand::Status { reply } => { let _ = reply.send(Box::new(self.status())); } + NetCommand::SetBroadcast { enabled, reply } => { + self.params.broadcast = enabled; + self.update_broadcast(); + self.reannounce(); + let _ = reply.send(()); + } NetCommand::Recheck => self.discovery_round().await, NetCommand::Reannounce => self.reannounce(), NetCommand::Release { reply } => { @@ -1017,6 +1029,24 @@ impl Runtime { } } + fn broadcast_policy(&self) -> crate::overlay::broadcast::BroadcastPolicy { + crate::overlay::broadcast::BroadcastPolicy { + enabled: self.params.broadcast, + peers: self + .sessions + .values() + .filter(|session| session.broadcast) + .map(|session| session.peer) + .collect(), + } + } + + fn update_broadcast(&self) { + self.params + .routes + .set_broadcast(self.network_id, self.broadcast_policy()); + } + /// Tells the plugins who holds which overlay address. async fn publish_allocations(&mut self) { let Some(range) = self.effective_range() else { @@ -1039,6 +1069,7 @@ impl Runtime { // a packet for ourselves does not go over a tunnel. let local = self.state.address_of(&self.local_id); let routes = crate::overlay::NetworkRoutes { + broadcast: self.broadcast_policy(), range: Some(range), local, peers: allocations @@ -1628,6 +1659,7 @@ impl Runtime { } capabilities.truncate(self.params.limits.max_capabilities); Announcement { + broadcast: self.params.broadcast, hostname: self.params.hostname.clone(), capabilities, } @@ -1705,6 +1737,7 @@ impl Runtime { } self.metrics.disconnects += 1; self.drop_links_for(peer); + self.update_broadcast(); self.update_paths(); self.announce_reach(false); for plugin in &self.params.plugins { @@ -1733,8 +1766,10 @@ impl Runtime { let capabilities = announcement.capabilities.clone(); if let Some(session) = self.sessions.get_mut(&peer) { session.hostname = Some(announcement.hostname.clone()); + session.broadcast = announcement.broadcast; session.capabilities = capabilities.clone(); } + self.update_broadcast(); self.dispatch_capabilities(peer, &capabilities); self.ensure_links(); self.update_paths(); @@ -1831,6 +1866,7 @@ impl Runtime { .map(|session| { let snapshot = snapshot_connection(&session.conn); PeerStatus { + broadcast: session.broadcast, endpoint_id: session.peer, role: session.role, hostname: session.hostname.clone(), @@ -1898,6 +1934,7 @@ impl Runtime { members.dedup_by(|a, b| a.endpoint_id == b.endpoint_id); NetworkStatus { + broadcast: self.params.broadcast, descriptor: self.params.keys.descriptor(), name: self.params.keys.name().clone(), network_id: self.network_id, diff --git a/crates/tsunagi/src/agent/session.rs b/crates/tsunagi/src/agent/session.rs index 9d95f85..e274265 100644 --- a/crates/tsunagi/src/agent/session.rs +++ b/crates/tsunagi/src/agent/session.rs @@ -73,6 +73,7 @@ pub(crate) struct Session { pub(crate) outbound: mpsc::Sender>, pub(crate) hostname: Option, pub(crate) capabilities: Vec, + pub(crate) broadcast: bool, pub(crate) messages_sent: u64, pub(crate) messages_received: u64, pub(crate) bytes_sent: u64, @@ -174,6 +175,7 @@ pub(crate) fn spawn( outbound: outbound_tx, hostname: None, capabilities: Vec::new(), + broadcast: false, messages_sent: 0, messages_received: 0, bytes_sent: 0, diff --git a/crates/tsunagi/src/agent/status.rs b/crates/tsunagi/src/agent/status.rs index 98ce247..bdcf392 100644 --- a/crates/tsunagi/src/agent/status.rs +++ b/crates/tsunagi/src/agent/status.rs @@ -81,6 +81,8 @@ pub struct MemberStatus { /// Status of one authenticated session. #[derive(Debug, Clone)] pub struct PeerStatus { + /// This authenticated peer accepts network broadcasts. + pub broadcast: bool, /// Authenticated endpoint id. pub endpoint_id: EndpointId, /// Which side this agent played in the handshake. @@ -151,6 +153,8 @@ pub struct NetworkMetrics { /// Status of one network. #[derive(Debug, Clone)] pub struct NetworkStatus { + /// Local broadcast participation in this network. + pub broadcast: bool, /// Immutable deterministic description of the network space. pub descriptor: NetworkDescriptor, /// Network name, for convenience. diff --git a/crates/tsunagi/src/ipc/mod.rs b/crates/tsunagi/src/ipc/mod.rs index 344ff31..9706e7d 100644 --- a/crates/tsunagi/src/ipc/mod.rs +++ b/crates/tsunagi/src/ipc/mod.rs @@ -92,6 +92,13 @@ pub fn control_socket_path(state_dir: &Path) -> PathBuf { pub enum Request { /// Report what the agent is doing. Status, + /// Set local broadcast participation in one configured network. + SetBroadcast { + /// Public network identifier. + network_id: String, + /// Whether to originate and accept broadcasts. + enabled: bool, + }, /// Answer to a different name from now on. /// /// Applied by the running agent rather than written behind its back, so @@ -135,6 +142,9 @@ pub enum Request { /// It travels over a socket only its owner can open, to the agent /// that stores it anyway, and never appears in `Debug`. secret: String, + /// Explicit choice, or preserve the saved setting. + #[serde(default)] + broadcast: Option, }, } @@ -142,6 +152,10 @@ impl std::fmt::Debug for Request { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Request::Status => f.write_str("Status"), + Request::SetBroadcast { + network_id, + enabled, + } => write!(f, "SetBroadcast {{ {network_id}, enabled: {enabled} }}"), Request::SetHostname(name) => write!(f, "SetHostname({name})"), Request::Leave(network) => write!(f, "Leave({network})"), Request::SetActive { network_id, active } => { @@ -162,6 +176,8 @@ impl std::fmt::Debug for Request { pub enum Response { /// A status report. Status(Box), + /// Accepted local broadcast participation. + Broadcast(bool), /// The name the agent now answers to, after reducing it to canonical form. Hostname(String), /// A network was left. @@ -274,6 +290,9 @@ pub struct DnsReport { /// One network. #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct NetworkReport { + /// Whether this agent participates in LAN broadcasts in this network. + #[serde(default)] + pub broadcast: bool, /// Network name. pub name: String, /// Public network identifier. @@ -455,10 +474,20 @@ pub trait ReportSource: Send + Sync + 'static { &self, _name: String, _secret: String, + _broadcast: Option, ) -> BoxFuture<'_, std::result::Result> { Box::pin(async move { Err("this agent cannot join a network".to_string()) }) } + /// Persists and applies local broadcast participation. + fn set_broadcast( + &self, + _network_id: String, + _enabled: bool, + ) -> BoxFuture<'_, std::result::Result> { + Box::pin(async move { Err("this agent cannot change broadcast participation".into()) }) + } + /// Stops serving a network, or starts serving it again. /// /// Defaulted to a refusal, like the others. @@ -510,7 +539,7 @@ pub const EXCHANGE_TIMEOUT: Duration = Duration::from_secs(5); /// /// Bump it whenever [`Request`], [`Response`] or anything they contain /// changes shape. -pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 13]); +pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 14]); /// Reads one request off an accepted stream, answers it, writes the response. /// @@ -535,6 +564,13 @@ where }; let response = match request { Request::Status => Response::Status(Box::new(source.report().await)), + Request::SetBroadcast { + network_id, + enabled, + } => match source.set_broadcast(network_id, enabled).await { + Ok(enabled) => Response::Broadcast(enabled), + Err(error) => Response::Error(error), + }, Request::SetHostname(hostname) => match source.set_hostname(hostname).await { Ok(accepted) => Response::Hostname(accepted), Err(reason) => Response::Error(reason), @@ -543,7 +579,11 @@ where Ok(report) => Response::Left(report), Err(reason) => Response::Error(reason), }, - Request::Join { name, secret } => match source.join(name, secret).await { + Request::Join { + name, + secret, + broadcast, + } => match source.join(name, secret, broadcast).await { Ok(report) => Response::Joined(report), Err(reason) => Response::Error(reason), }, @@ -636,9 +676,21 @@ pub async fn join_network( path: impl AsRef, name: &str, secret: &str, +) -> Result { + let path = path.as_ref(); + join_network_with_broadcast(path, name, secret, None).await +} + +/// Joins with an explicit per-network broadcast choice. +pub async fn join_network_with_broadcast( + path: impl AsRef, + name: &str, + secret: &str, + broadcast: Option, ) -> Result { let path = path.as_ref(); let request = Request::Join { + broadcast, name: name.to_string(), secret: secret.to_string(), }; @@ -649,6 +701,24 @@ pub async fn join_network( } } +/// Changes one network's broadcast participation while the agent runs. +pub async fn set_broadcast( + path: impl AsRef, + network_id: &str, + enabled: bool, +) -> Result { + let path = path.as_ref(); + let request = Request::SetBroadcast { + network_id: network_id.to_owned(), + enabled, + }; + match exchange(path, &request, EXCHANGE_TIMEOUT).await? { + Response::Broadcast(enabled) => Ok(enabled), + Response::Error(message) => Err(Error::Storage(message)), + other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), + } +} + /// Asks a running agent to stop serving a network, or to serve it again. pub async fn set_active( path: impl AsRef, diff --git a/crates/tsunagi/src/overlay/broadcast.rs b/crates/tsunagi/src/overlay/broadcast.rs new file mode 100644 index 0000000..34f2ec1 --- /dev/null +++ b/crates/tsunagi/src/overlay/broadcast.rs @@ -0,0 +1,228 @@ +//! Scoped LAN discovery fanout at IP ingress. A domain is one logical network; +//! routed copies use ordinary encrypted peer links and are never reflooded. +//! Future subnet exporters must supply explicitly authorized ingress domains +//! here, not teach the encrypted transit router to inspect application bytes. + +use super::router::{NetworkRoutes, Route}; +use crate::NetworkId; +use iroh::EndpointId; +use std::collections::{HashMap, HashSet}; +use std::net::Ipv4Addr; +use std::sync::Arc; + +/// Local participation and the currently authenticated participants in a domain. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BroadcastPolicy { + /// Whether this agent originates and accepts LAN broadcasts in this network. + pub enabled: bool, + /// Live members explicitly advertising that they accept broadcasts. + pub peers: HashSet, +} +impl Default for BroadcastPolicy { + fn default() -> Self { + Self { + enabled: true, + peers: HashSet::new(), + } + } +} + +#[derive(Debug)] +struct Domain { + network: NetworkId, + directed: Ipv4Addr, + enabled: bool, + recipients: Arc<[Route]>, + members: HashSet, +} + +/// Immutable ingress-domain index rebuilt with address/policy changes. +#[derive(Debug, Default)] +pub(crate) struct BroadcastTable { + sources: HashMap>, + networks: HashMap>, + destinations: HashSet, +} +impl BroadcastTable { + pub fn build(networks: &HashMap) -> Self { + let mut table = Self::default(); + for (&network, routes) in networks { + let (Some(range), Some(local)) = (routes.range, routes.local) else { + continue; + }; + if range.prefix_len > 30 { + continue; + } + let directed = Ipv4Addr::from(u32::from(range.base) | (u32::MAX >> range.prefix_len)); + let mut recipients: Vec<_> = routes + .peers + .iter() + .filter(|(_, peer)| routes.broadcast.peers.contains(peer)) + .map(|(_, peer)| Route { + network, + peer: *peer, + }) + .collect(); + recipients.sort_by_key(|route| route.peer); + recipients.dedup(); + let domain = Arc::new(Domain { + network, + directed, + enabled: routes.broadcast.enabled, + members: recipients.iter().map(|route| route.peer).collect(), + recipients: recipients.into(), + }); + table.sources.insert(local, domain.clone()); + table.networks.insert(network, domain); + table.destinations.insert(directed); + } + table + } + pub fn is_destination(&self, destination: Ipv4Addr) -> bool { + destination.is_broadcast() || self.destinations.contains(&destination) + } + pub fn outgoing(&self, source: Ipv4Addr, destination: Ipv4Addr) -> Option> { + let domain = self.sources.get(&source)?; + (domain.enabled && (destination.is_broadcast() || destination == domain.directed)) + .then(|| domain.recipients.clone()) + } + pub fn accepts(&self, network: NetworkId, peer: EndpointId, destination: Ipv4Addr) -> bool { + self.networks.get(&network).is_some_and(|domain| { + domain.network == network + && domain.enabled + && domain.members.contains(&peer) + && (destination.is_broadcast() || destination == domain.directed) + }) + } +} + +/// Validates IPv4/UDP lengths before broadcast fanout. No game-specific ports, +/// payload rewriting or checksum changes. IP fragments retain their bytes and +/// are reassembled by the destination OS; even later fragments have protocol 17. +pub(crate) fn valid_udp(packet: &[u8]) -> bool { + if packet.len() < 20 || packet[0] >> 4 != 4 || packet[9] != 17 { + return false; + } + let header = usize::from(packet[0] & 15) * 4; + let total = usize::from(u16::from_be_bytes([packet[2], packet[3]])); + if header < 20 || total <= header || total > packet.len() { + return false; + } + let fragment = u16::from_be_bytes([packet[6], packet[7]]); + if fragment & 0x1fff != 0 { + return true; + } + if total < header + 8 { + return false; + } + let udp = usize::from(u16::from_be_bytes([packet[header + 4], packet[header + 5]])); + udp >= 8 + && if fragment & 0x2000 != 0 { + udp >= total - header + } else { + udp == total - header + } +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + use super::*; + use crate::identity::{NetworkKeys, NetworkName, NetworkSecret}; + fn id(name: &str) -> NetworkId { + NetworkKeys::derive( + &NetworkName::new(name).unwrap(), + &NetworkSecret::from_bytes([4; 32]).unwrap(), + ) + .network_id() + } + #[test] + fn limited_and_directed_broadcasts_never_cross_domains_or_target_disabled_peers() { + let peer = iroh::SecretKey::from_bytes(&[8; 32]).public(); + let a = NetworkRoutes { + range: Some("10.0.0.0/24".parse().unwrap()), + local: Some("10.0.0.1".parse().unwrap()), + peers: vec![("10.0.0.2".parse().unwrap(), peer)], + broadcast: BroadcastPolicy { + enabled: true, + peers: HashSet::from([peer]), + }, + }; + let b = NetworkRoutes { + range: Some("10.1.0.0/24".parse().unwrap()), + local: Some("10.1.0.1".parse().unwrap()), + peers: vec![("10.1.0.2".parse().unwrap(), peer)], + broadcast: BroadcastPolicy::default(), + }; + let table = BroadcastTable::build(&HashMap::from([(id("a"), a.clone()), (id("b"), b)])); + let routes = table + .outgoing(a.local.unwrap(), Ipv4Addr::BROADCAST) + .unwrap(); + assert_eq!( + &*routes, + &[Route { + network: id("a"), + peer + }] + ); + assert!( + table + .outgoing(a.local.unwrap(), "10.1.0.255".parse().unwrap()) + .is_none() + ); + assert!( + table + .outgoing("192.168.1.2".parse().unwrap(), Ipv4Addr::BROADCAST) + .is_none() + ); + assert!( + table + .outgoing("10.1.0.1".parse().unwrap(), Ipv4Addr::BROADCAST) + .unwrap() + .is_empty() + ); + assert!(!table.accepts(id("a"), peer, "10.1.0.255".parse().unwrap())); + assert!(!table.accepts(id("b"), peer, Ipv4Addr::BROADCAST)); + assert!(table.accepts(id("a"), peer, Ipv4Addr::BROADCAST)); + let disabled = NetworkRoutes { + broadcast: BroadcastPolicy { + enabled: false, + ..a.broadcast.clone() + }, + ..a.clone() + }; + let table = BroadcastTable::build(&HashMap::from([(id("a"), disabled)])); + assert!( + table + .outgoing(a.local.unwrap(), Ipv4Addr::BROADCAST) + .is_none() + ); + assert!(!table.accepts(id("a"), peer, Ipv4Addr::BROADCAST)); + } + #[test] + fn malformed_udp_is_not_amplified_but_ipv4_fragments_are_supported() { + let mut packet = vec![0u8; 36]; + packet[0] = 0x45; + packet[3] = 36; + packet[9] = 17; + packet[25] = 16; + assert!(valid_udp(&packet)); + for length in 0..36 { + assert!(!valid_udp(&packet[..length])); + } + packet[9] = 6; + assert!(!valid_udp(&packet)); + packet[9] = 17; + packet[0] = 0x44; + assert!(!valid_udp(&packet)); + packet[0] = 0x45; + packet[25] = 8; + assert!(!valid_udp(&packet)); + packet[6] = 0x20; + packet[25] = 64; + assert!(valid_udp(&packet)); + packet[6] = 0; + packet[7] = 2; + assert!(valid_udp(&packet)); + } +} diff --git a/crates/tsunagi/src/overlay/interface.rs b/crates/tsunagi/src/overlay/interface.rs index ac56e08..921868a 100644 --- a/crates/tsunagi/src/overlay/interface.rs +++ b/crates/tsunagi/src/overlay/interface.rs @@ -53,6 +53,10 @@ pub enum Rejected { WrongFamily, /// The sending peer does not hold the source address it used. WrongSource, + /// Broadcast disabled, outside this domain, or malformed/non-UDP. + BroadcastDenied, + /// The packet terminates outside this host's network address/domain. + WrongDestination, } /// Counters for one interface. @@ -60,6 +64,12 @@ pub enum Rejected { pub struct Counters { /// Packets handed to a protocol. pub sent: u64, + /// Broadcast copies handed to recipient tunnels. + pub broadcast_sent: u64, + /// Broadcast packets admitted to this host. + pub broadcast_received: u64, + /// Broadcasts refused by domain policy or packet validation. + pub broadcast_dropped: u64, /// Packets written to the operating system. pub received: u64, /// Packets for an address nobody in any network holds. @@ -68,10 +78,12 @@ pub struct Counters { pub unroutable_sample: Option, /// Packets nothing could carry, though their destination was known. pub undeliverable: u64, - /// Multicast and broadcast packets, which the overlay does not carry. + /// Unsupported multicast or unspecified-destination packets. pub multicast: u64, /// Packets from a peer that does not hold the source address used. pub wrong_source: u64, + /// Decrypted packets addressed outside the receiving network's local host. + pub wrong_destination: u64, /// Packets that could not be read at all. pub malformed: u64, } @@ -79,11 +91,15 @@ pub struct Counters { #[derive(Debug, Default)] struct Tally { sent: AtomicU64, + broadcast_sent: AtomicU64, + broadcast_received: AtomicU64, + broadcast_dropped: AtomicU64, received: AtomicU64, unroutable: AtomicU64, undeliverable: AtomicU64, multicast: AtomicU64, wrong_source: AtomicU64, + wrong_destination: AtomicU64, malformed: AtomicU64, sample: std::sync::Mutex>, } @@ -92,6 +108,9 @@ impl Tally { fn snapshot(&self) -> Counters { Counters { sent: self.sent.load(Ordering::Relaxed), + broadcast_sent: self.broadcast_sent.load(Ordering::Relaxed), + broadcast_received: self.broadcast_received.load(Ordering::Relaxed), + broadcast_dropped: self.broadcast_dropped.load(Ordering::Relaxed), received: self.received.load(Ordering::Relaxed), unroutable: self.unroutable.load(Ordering::Relaxed), unroutable_sample: match self.sample.lock() { @@ -101,6 +120,7 @@ impl Tally { undeliverable: self.undeliverable.load(Ordering::Relaxed), multicast: self.multicast.load(Ordering::Relaxed), wrong_source: self.wrong_source.load(Ordering::Relaxed), + wrong_destination: self.wrong_destination.load(Ordering::Relaxed), malformed: self.malformed.load(Ordering::Relaxed), } } @@ -158,6 +178,27 @@ impl Interface { continue; }; let destination = header.destination(); + if let (IpAddr::V4(source), IpAddr::V4(destination)) = + (header.source(), destination) + && routes.is_broadcast(destination) + { + let recipients = super::broadcast::valid_udp(&packet) + .then(|| routes.broadcast_recipients(source, destination)) + .flatten(); + if let Some(recipients) = recipients { + for &route in recipients.iter() { + if carrier.carry(route, packet.clone()) { + tally.sent.fetch_add(1, Ordering::Relaxed); + tally.broadcast_sent.fetch_add(1, Ordering::Relaxed); + } else { + tally.undeliverable.fetch_add(1, Ordering::Relaxed); + } + } + } else { + tally.broadcast_dropped.fetch_add(1, Ordering::Relaxed); + } + continue; + } if is_multicast_or_broadcast(destination) { // The operating system emits these on any interface. // The overlay is a set of point-to-point tunnels and @@ -286,7 +327,31 @@ impl Interface { self.tally.wrong_source.fetch_add(1, Ordering::Relaxed); return Err(Rejected::WrongSource); } + let broadcast = match header.destination() { + IpAddr::V4(destination) if self.routes.is_broadcast(destination) => { + if !super::broadcast::valid_udp(&packet) + || !self.routes.accepts_broadcast(network, peer, destination) + { + self.tally.broadcast_dropped.fetch_add(1, Ordering::Relaxed); + return Err(Rejected::BroadcastDenied); + } + true + } + IpAddr::V4(destination) if self.routes.is_local_destination(network, destination) => { + false + } + _ => { + self.tally.wrong_destination.fetch_add(1, Ordering::Relaxed); + return Err(Rejected::WrongDestination); + } + }; + // Remote broadcasts terminate here. Only local TUN ingress can fan out. if self.device.send(packet).await.is_ok() { + if broadcast { + self.tally + .broadcast_received + .fetch_add(1, Ordering::Relaxed); + } self.tally.received.fetch_add(1, Ordering::Relaxed); } Ok(()) @@ -393,6 +458,73 @@ mod tests { Bytes::from(packet) } + #[tokio::test] + async fn broadcast_ingress_is_validated_and_remote_delivery_never_refloods() { + use super::super::broadcast::BroadcastPolicy; + use std::collections::HashSet; + let (interface, device, routes, carrier, id) = interface(false).await; + let enabled = BroadcastPolicy { + enabled: true, + peers: HashSet::from([peer(2)]), + }; + routes.set_broadcast(id, enabled.clone()); + let packet = ipv4(addr(2), Ipv4Addr::BROADCAST, &[0, 1, 0, 2, 0, 8, 0, 0]); + interface + .deliver(id, peer(2), packet.clone()) + .await + .unwrap(); + assert_eq!(device.pop_to_os().await.unwrap(), packet); + assert!( + carrier.carried().is_empty(), + "remote ingress never originates a fanout" + ); + assert_eq!( + interface.deliver(id, peer(3), packet.clone()).await, + Err(Rejected::WrongSource) + ); + assert_eq!( + interface + .deliver(id, peer(2), ipv4(addr(2), Ipv4Addr::BROADCAST, b"bad")) + .await, + Err(Rejected::BroadcastDenied) + ); + routes.set_broadcast( + id, + BroadcastPolicy { + enabled: false, + ..enabled + }, + ); + assert_eq!( + interface.deliver(id, peer(2), packet).await, + Err(Rejected::BroadcastDenied) + ); + // An authenticated sender cannot use TUN as a gateway to a physical LAN. + assert_eq!( + interface + .deliver( + id, + peer(2), + ipv4( + addr(2), + "192.168.1.255".parse().unwrap(), + &[0, 1, 0, 2, 0, 8, 0, 0] + ) + ) + .await, + Err(Rejected::WrongDestination) + ); + device.push_from_os(ipv4( + addr(1), + Ipv4Addr::BROADCAST, + &[0, 1, 0, 2, 0, 8, 0, 0], + )); + wait_for(&interface, |c| c.broadcast_dropped.saturating_sub(2)).await; + assert!(carrier.carried().is_empty()); + assert_eq!(interface.counters().received, 1); + interface.remove().await; + } + /// Records what it was asked to carry, and can refuse. #[derive(Debug, Default)] struct Recorder { @@ -437,6 +569,7 @@ mod tests { .set_network( id, NetworkRoutes { + broadcast: Default::default(), range: Some("10.13.37.0/24".parse().unwrap()), local: Some(addr(1)), peers: vec![(addr(2), peer(2)), (addr(3), peer(3))], @@ -602,6 +735,7 @@ mod tests { .set_network( id, NetworkRoutes { + broadcast: Default::default(), range: Some("10.13.37.0/24".parse().unwrap()), local: Some(addr(9)), peers: vec![(addr(2), peer(2))], diff --git a/crates/tsunagi/src/overlay/mod.rs b/crates/tsunagi/src/overlay/mod.rs index 09abd15..e0f33f9 100644 --- a/crates/tsunagi/src/overlay/mod.rs +++ b/crates/tsunagi/src/overlay/mod.rs @@ -36,6 +36,7 @@ pub enum OverlayError { Other(String), } +pub mod broadcast; pub mod config; pub mod interface; pub mod packet; diff --git a/crates/tsunagi/src/overlay/router.rs b/crates/tsunagi/src/overlay/router.rs index f030316..302473c 100644 --- a/crates/tsunagi/src/overlay/router.rs +++ b/crates/tsunagi/src/overlay/router.rs @@ -20,9 +20,11 @@ //! carries traffic for the same addresses and the table is the same whichever //! one is in use. +use super::broadcast::{BroadcastPolicy, BroadcastTable}; +use arc_swap::ArcSwap; use std::collections::HashMap; use std::net::Ipv4Addr; -use std::sync::RwLock; +use std::sync::{Arc, RwLock}; use iroh::EndpointId; @@ -41,6 +43,8 @@ pub struct Route { /// What one network contributes to the table. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct NetworkRoutes { + /// Local broadcast policy and authenticated willing recipients. + pub broadcast: BroadcastPolicy, /// The range this network allocates from, once it has agreed one. pub range: Option, /// This agent's own address in the network. @@ -73,6 +77,7 @@ pub enum RouteError { /// Address ownership across every network this agent is in. #[derive(Debug, Default)] pub struct RoutingTable { + broadcast: ArcSwap, networks: RwLock>, } @@ -122,12 +127,54 @@ impl RoutingTable { } } networks.insert(network, routes); + self.broadcast + .store(Arc::new(BroadcastTable::build(&networks))); Ok(()) } /// Forgets a network. pub fn remove_network(&self, network: NetworkId) { - self.write().remove(&network); + let mut networks = self.write(); + networks.remove(&network); + self.broadcast + .store(Arc::new(BroadcastTable::build(&networks))); + } + + /// Replaces only a network's broadcast participation, without changing addresses. + pub fn set_broadcast(&self, network: NetworkId, policy: BroadcastPolicy) { + let mut networks = self.write(); + if let Some(routes) = networks.get_mut(&network) { + if routes.broadcast == policy { + return; + } + routes.broadcast = policy; + self.broadcast + .store(Arc::new(BroadcastTable::build(&networks))); + } + } + + /// Recognizes limited and configured subnet broadcasts before unicast lookup. + pub fn is_broadcast(&self, destination: Ipv4Addr) -> bool { + self.broadcast.load().is_destination(destination) + } + + /// A precomputed domain-scoped recipient list; no per-packet graph search. + pub fn broadcast_recipients( + &self, + source: Ipv4Addr, + destination: Ipv4Addr, + ) -> Option> { + self.broadcast.load().outgoing(source, destination) + } + + /// Local admission after the plugin authenticated and decrypted the sender. + pub fn accepts_broadcast( + &self, + network: NetworkId, + peer: EndpointId, + destination: Ipv4Addr, + ) -> bool { + self.broadcast.load().accepts(network, peer, destination) } /// The peer that holds a destination address, if anybody does. @@ -164,6 +211,14 @@ impl RoutingTable { }) } + /// Whether an ordinary packet terminates on this host in this network. + /// Exported LAN subnets must extend this admission policy explicitly later. + pub fn is_local_destination(&self, network: NetworkId, destination: Ipv4Addr) -> bool { + self.read() + .get(&network) + .is_some_and(|routes| routes.local == Some(destination)) + } + /// Every address this agent should answer to, with its prefix length. pub fn local_addresses(&self) -> Vec<(Ipv4Addr, u8)> { let mut addresses: Vec<(Ipv4Addr, u8)> = self @@ -236,6 +291,7 @@ mod tests { fn routes() -> NetworkRoutes { NetworkRoutes { + broadcast: Default::default(), range: Some("10.13.37.0/24".parse().unwrap()), local: Some(addr(1)), peers: vec![(addr(2), peer(2)), (addr(3), peer(3))], @@ -318,6 +374,7 @@ mod tests { .set_network( second, NetworkRoutes { + broadcast: Default::default(), range: Some("10.99.0.0/16".parse().unwrap()), local: Some(Ipv4Addr::new(10, 99, 0, 1)), peers: vec![(Ipv4Addr::new(10, 99, 0, 2), peer(4))], @@ -362,6 +419,7 @@ mod tests { .set_network( network("wide"), NetworkRoutes { + broadcast: Default::default(), range: Some("10.0.0.0/8".parse().unwrap()), ..Default::default() }, @@ -373,6 +431,7 @@ mod tests { .set_network( network("narrow"), NetworkRoutes { + broadcast: Default::default(), range: Some("10.13.37.0/24".parse().unwrap()), ..Default::default() } @@ -414,6 +473,7 @@ mod tests { .set_network( network("two"), NetworkRoutes { + broadcast: Default::default(), range: Some("10.99.0.0/16".parse().unwrap()), local: Some(Ipv4Addr::new(10, 99, 0, 1)), peers: Vec::new(), diff --git a/crates/tsunagi/src/proto/message.rs b/crates/tsunagi/src/proto/message.rs index 21eb122..6e5d8ed 100644 --- a/crates/tsunagi/src/proto/message.rs +++ b/crates/tsunagi/src/proto/message.rs @@ -20,7 +20,7 @@ use crate::error::ProtocolError; /// The version in the ALPN is the wire-compatibility version of the control /// protocol. It is independent of the network identity scheme version, so /// bumping it must not change any existing [`crate::NetworkId`]. -pub const ALPN: &[u8] = b"tsunagi/ctrl/2"; +pub const ALPN: &[u8] = b"tsunagi/ctrl/3"; /// ALPN of the tsunagi data plane. /// @@ -41,7 +41,7 @@ pub const MAX_DATA_PROTOCOL_LEN: usize = 32; pub const MAX_SIGNATURE_LEN: usize = 64; /// Control protocol version carried inside the handshake. -pub const PROTOCOL_VERSION: u16 = 2; +pub const PROTOCOL_VERSION: u16 = 3; /// First message of the handshake, sent by the initiator. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -73,6 +73,8 @@ pub struct AuthProof { /// What this agent tells a peer about itself. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct Announcement { + /// Local opt-in to receive LAN broadcasts in this network. + pub broadcast: bool, /// Human-readable hostname. A mutable binding, not an identity. pub hostname: String, /// Announced IP plugin capabilities. Opaque to the core. diff --git a/crates/tsunagi/src/storage/mod.rs b/crates/tsunagi/src/storage/mod.rs index 66d95b6..ed405ff 100644 --- a/crates/tsunagi/src/storage/mod.rs +++ b/crates/tsunagi/src/storage/mod.rs @@ -248,6 +248,12 @@ impl Storage { .await } + /// Persists local broadcast participation for one configured network. + pub async fn set_broadcast(&self, network_id: NetworkId, enabled: bool) -> Result<()> { + self.with_state(move |state| state.set_broadcast(network_id, enabled)) + .await + } + /// Updates the auto-start flag of a network. pub async fn set_auto_start(&self, network_id: NetworkId, auto_start: bool) -> Result<()> { self.with_state(move |state| state.set_auto_start(network_id, auto_start)) diff --git a/crates/tsunagi/src/storage/state.rs b/crates/tsunagi/src/storage/state.rs index 68045d3..cfc720f 100644 --- a/crates/tsunagi/src/storage/state.rs +++ b/crates/tsunagi/src/storage/state.rs @@ -21,7 +21,7 @@ use crate::identity::{DeviceIdentity, NetworkId, NetworkName, NetworkSecret}; use crate::state::{RecordBody, SignedRecord}; /// Schema version written by this build. -pub const SCHEMA_VERSION: i64 = 3; +pub const SCHEMA_VERSION: i64 = 4; /// Key of the stored hostname setting. const SETTING_HOSTNAME: &str = "hostname"; @@ -40,6 +40,8 @@ pub struct StoredNetwork { pub secret: NetworkSecret, /// Whether the network is activated automatically at agent startup. pub auto_start: bool, + /// Local broadcast participation, enabled unless explicitly disabled. + pub broadcast: bool, } /// The mandatory state store. @@ -183,6 +185,12 @@ impl StateStore { .execute_batch(SIGNED_RECORDS_SCHEMA) .map_err(|err| self.corrupt(format!("cannot create schema: {err}")))?; } + if found < 4 { + self.conn.execute_batch("BEGIN; + ALTER TABLE networks ADD COLUMN broadcast INTEGER NOT NULL DEFAULT 1 CHECK (broadcast IN (0,1)); + PRAGMA user_version = 4; + COMMIT;").map_err(|err| self.corrupt(format!("cannot migrate schema to 4: {err}")))?; + } Ok(()) } @@ -396,6 +404,21 @@ impl StateStore { Ok(()) } + /// Saves local participation without changing identity, membership or auto-start. + pub fn set_broadcast(&self, network_id: NetworkId, enabled: bool) -> Result<()> { + let changed = self + .conn + .execute( + "UPDATE networks SET broadcast = ?2 WHERE network_id = ?1", + params![network_id.as_bytes().as_slice(), enabled as i64], + ) + .map_err(|err| Error::Storage(format!("cannot update broadcast policy: {err}")))?; + if changed == 0 { + return Err(Error::NetworkUnknown(network_id)); + } + Ok(()) + } + /// Removes a network configuration entirely. pub fn remove_network(&self, network_id: NetworkId) -> Result<()> { self.conn @@ -411,7 +434,7 @@ impl StateStore { pub fn list_networks(&self) -> Result> { let mut stmt = self .conn - .prepare("SELECT network_id, name, secret, auto_start FROM networks ORDER BY name") + .prepare("SELECT network_id, name, secret, auto_start, broadcast FROM networks ORDER BY name") .map_err(|err| Error::Storage(format!("cannot list networks: {err}")))?; let rows = stmt .query_map([], |row| { @@ -419,13 +442,13 @@ impl StateStore { let name: String = row.get(1)?; let secret: Vec = row.get(2)?; let auto_start: i64 = row.get(3)?; - Ok((id, name, secret, auto_start != 0)) + Ok((id, name, secret, auto_start != 0, row.get::<_, bool>(4)?)) }) .map_err(|err| Error::Storage(format!("cannot list networks: {err}")))?; let mut out = Vec::new(); for row in rows { - let (id, name, secret, auto_start) = + let (id, name, secret, auto_start, broadcast) = row.map_err(|err| Error::Storage(format!("cannot read network row: {err}")))?; let id: [u8; 32] = id .as_slice() @@ -436,6 +459,7 @@ impl StateStore { name: NetworkName::new(name)?, secret: NetworkSecret::from_bytes(secret)?, auto_start, + broadcast, }); } Ok(out) @@ -668,3 +692,58 @@ pub(crate) fn now_unix() -> i64 { .map(|d| d.as_secs() as i64) .unwrap_or(0) } +#[cfg(test)] +mod broadcast_storage_tests { + #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + use super::*; + use crate::identity::NetworkKeys; + + #[test] + fn v3_migration_preserves_identity_and_networks_and_defaults_broadcast_on() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("state.sqlite"); + let name = NetworkName::new("migration-broadcast").unwrap(); + let secret = NetworkSecret::from_bytes([7; 32]).unwrap(); + let id = NetworkKeys::derive(&name, &secret).network_id(); + let store = StateStore::open(&path).unwrap(); + let identity = store + .load_or_create_device_identity() + .unwrap() + .endpoint_id(); + store.upsert_network(id, &name, &secret, false).unwrap(); + store.set_hostname("old-host").unwrap(); + store + .conn + .execute_batch("ALTER TABLE networks DROP COLUMN broadcast; PRAGMA user_version=3;") + .unwrap(); + drop(store); + let store = StateStore::open(&path).unwrap(); + let networks = store.list_networks().unwrap(); + assert_eq!(networks.len(), 1); + assert_eq!(networks[0].network_id, id); + assert!(!networks[0].auto_start); + assert!(networks[0].broadcast); + assert_eq!(store.hostname().unwrap().as_deref(), Some("old-host")); + assert_eq!( + store + .load_or_create_device_identity() + .unwrap() + .endpoint_id(), + identity + ); + store.set_broadcast(id, false).unwrap(); + store.upsert_network(id, &name, &secret, true).unwrap(); + drop(store); + let store = StateStore::open(&path).unwrap(); + assert!( + !store.list_networks().unwrap()[0].broadcast, + "rejoin and restart preserve opt-out" + ); + store.remove_network(id).unwrap(); + store.upsert_network(id, &name, &secret, true).unwrap(); + assert!( + store.list_networks().unwrap()[0].broadcast, + "forgotten network gets defaults" + ); + } +} diff --git a/docs/architecture.md b/docs/architecture.md index e079798..d899195 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -60,6 +60,7 @@ not tunnel control sessions. See [routing.md](routing.md). | `dataplane::routing` | transport-independent graph, shortest paths and opaque flow identifiers | | `dataplane::relay` | immutable forwarding snapshots and transport-to-transport transit | | `dataplane` | the contract an IP protocol implements | +| `overlay::broadcast` | domain-scoped IPv4 UDP discovery fanout and receive admission | | `overlay` | the one interface an agent owns: provisioning, the TUN, whose packet is whose | | `dns` | the DNS view of a network, and telling the system resolver about it | diff --git a/docs/broadcast.md b/docs/broadcast.md new file mode 100644 index 0000000..2cbb4b2 --- /dev/null +++ b/docs/broadcast.md @@ -0,0 +1,91 @@ +# LAN discovery broadcasts + +Broadcast participation is a **local setting of each network**, enabled by +default. Different networks on the same agent can have different settings; +each member advertises whether it accepts broadcasts over its authenticated +control session. It is not a network-wide vote or signed shared configuration. + +```sh +tsunagi join -n games --no-broadcast +tsunagi join -n games --broadcast +tsunagi network broadcast off +tsunagi network broadcast on +tsunagi network broadcast +``` + +Joining without either flag preserves the saved choice. The `network broadcast` +command can change a running network immediately or configure it while the +agent is stopped. `status` reports the local setting. There is no global `up` +override: starting the agent preserves every network's independent policy. + +## Delivery + +At local TUN ingress, the IP router recognizes IPv4 UDP addressed to the limited +broadcast `255.255.255.255` or the configured overlay subnet's directed broadcast +(for example `10.13.37.255` for `/24`). These address forms follow +[RFC 1122 section 3.3.6](https://www.rfc-editor.org/rfc/rfc1122#section-3.3.6). +There is no port allowlist: discovery requests and server announcements on +game-specific UDP ports use the same mechanism. IPv4/UDP lengths are checked +before fanout, including handling IP fragments. IPv6 multicast, mDNS, DHCP +bridging, Ethernet/IPX and physical LAN capture are outside this feature. + +The local source address selects exactly one broadcast domain. A limited +broadcast is never copied into all networks merely because they share a TUN. +A directed broadcast must belong to that source's network. Unspecified sources +and physical LAN sources have no authorized domain yet. Recipients are live, +authenticated, opted-in members with signed overlay address ownership. The +immutable recipient list is rebuilt with address/policy changes, not per packet. + +The origin creates **one encrypted unicast copy per eligible peer** and hands it +to the existing shortest-path router. Each can travel through several transport +links. Transit handles opaque ciphertext with the existing hop limit and does +not create another fanout. The original IP source, destination, TTL, UDP ports, +checksum and payload remain unchanged. Reply packets use ordinary unicast. +There is no broadcast retransmission timer or application-payload deduplication: +games may legitimately repeat the same discovery query. WireGuard replay checks +reject duplicated encrypted packets delivered by the transport. + +At the destination, the usual authenticated-source ownership check still runs. +Broadcast admission also checks the receiving network's setting, the sender's +announced participation, the destination domain and UDP framing. Disabling it +takes effect locally even while a remote sender still has an older announcement. +Received broadcasts terminate in the local TUN and never enter the fanout path. +An ordinary decrypted packet must target this host's address in that network; +the TUN is not a transit gateway to another overlay or physical subnet. + +Opt-out disables originating and accepting broadcasts on that host/network. +It does not disable opaque unicast transit for other members: intermediate +routers cannot inspect another pair's encrypted IP payload. + +## Host behavior and verification + +The game must send discovery through the Tsunagi interface. Traffic bound to a +physical adapter never reaches this TUN, and this feature does not capture it. +The suite tests real agents, authenticated iroh links and WireGuard encryption +with memory TUNs: limited/directed UDP discovery, multihop fanout, single-copy +delivery, opt-out, unicast replies, malformed input, source validation and +network isolation. It does not launch CS 1.6 or Warcraft III or establish that +every game/host chooses the virtual adapter automatically. + +## Future exported LANs + +`overlay::broadcast` owns domain selection and recipient policy separately from +IP parsing, encryption and encrypted transit. A future LAN adapter should feed +an explicitly authorized ingress domain into this layer, and export routes +through the corresponding source/destination admission policy. It must not +relax address checks globally or reinterpret all `.255` addresses as broadcasts. +Multiple gateways to the same LAN will need origin identifiers and bounded +duplicate suppression at LAN ingress/egress before physical rebroadcast is +enabled. No subnet export or host forwarding configuration is added here. + +## Compatibility + +Control ALPN is `tsunagi/ctrl/3` because `Announce` now carries participation. +The encrypted transit envelope and data ALPN `tsunagi/data/4` stay unchanged. +Local control protocol 14 rejects commands from mismatched running binaries +with a restart message instead of decoding a different request shape. +Upgrade all members and restart the agent before using the new CLI. + +SQLite schema 4 adds `networks.broadcast` with default `1`. Existing network +names, secrets, identity keys and signed records are retained. The old binary +does not understand schema 4; the migration is not a downgrade mechanism. diff --git a/docs/protocol.md b/docs/protocol.md index 4951c5f..3f09f68 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -9,7 +9,7 @@ Two versions exist and are independent: - **Identity scheme**, `tsunagi-network-id-v1`. Frozen. Changing it creates a different network space for the same name and secret. -- **Control protocol**, ALPN `tsunagi/ctrl/2`, `PROTOCOL_VERSION = 2`. +- **Control protocol**, ALPN `tsunagi/ctrl/3`, `PROTOCOL_VERSION = 3`. Upgrading the crate or bumping the control protocol must never change an existing `NetworkId`. @@ -191,7 +191,7 @@ transport connection, not supplied by the frame. Intermediate nodes cannot decrypt or authenticate the inner WireGuard payload; the destination does that. Flow ids are routing hints, not authorization proofs. -Control ALPN 2 carries `Reachable { links: [{ peer, protocol }] }`. Each row +Control ALPN 3 carries `Reachable { links: [{ peer, protocol }] }`. Each row belongs to the authenticated sender and is replaced atomically, expires after 90 seconds, and is withdrawn on session closure. Only compatible authenticated members enter a protocol's graph; local edges always come from actual links. @@ -215,7 +215,7 @@ not affect other networks. | message | meaning | |---|---| -| `Announce { hostname, capabilities }` | this agent's hostname and IP-plugin capabilities | +| `Announce { hostname, capabilities, broadcast }` | this agent's hostname, IP-plugin capabilities and local broadcast participation | | `Ping { seq, payload }` | small request used to verify the exchange | | `Pong { seq, payload }` | the echoed reply | | `State { records }` | a snapshot of signed records, merged into what the receiver holds | @@ -223,6 +223,11 @@ not affect other networks. | `Reachable { links }` | sender's current direct data links, scoped by protocol | | `Bye { reason }` | graceful goodbye; not a revocation of anything | +`broadcast` is a per-network local opt-in, enabled by default and treated as +false until an authenticated announcement arrives. It governs IP broadcast +fanout and local admission; encrypted transit still uses the existing envelope. +See [broadcast.md](broadcast.md) for scope and persistence. + A `State` snapshot is merged, never substituted: an author missing from it is left untouched. Each record carries its own signature, so a peer forwarding somebody else's record cannot alter it, and a record that fails verification diff --git a/docs/routing.md b/docs/routing.md index 4567540..678aeb9 100644 --- a/docs/routing.md +++ b/docs/routing.md @@ -86,5 +86,5 @@ counts sends without storing frames. Results are CPU forwarding cost, **not** end-to-end network latency or VPN throughput; encryption, fragmentation, sockets, congestion and scheduling contribute separately. -Wire compatibility: control ALPN `tsunagi/ctrl/2`, data ALPN `tsunagi/data/4`. +Wire compatibility: control ALPN `tsunagi/ctrl/3`, data ALPN `tsunagi/data/4`. Upgrade every participant together; saved identities and network state persist. diff --git a/docs/testing.md b/docs/testing.md index f0fec9f..5f0c73c 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -76,6 +76,15 @@ links. Real encrypted 1280-byte TCP packets travel in both directions while the middle TUNs remain empty. A direct A—D link is enabled, then removed; the route switches back to the chain without replacing end-to-end tunnels. +Broadcast tests use limited and directed UDP game discovery packets over real +WireGuard tunnels, including a missing direct link. Each willing peer gets one +copy; disabling reception/origination works at runtime and unicast still works. +IP-router tests cover source/destination domain checks, malformed UDP and the +absence of reflection into outgoing fanout. SQLite migration tests preserve v3 +identity/settings and check opt-out after reopen/rejoin. CLI tests cover default +on, per-network opt-out, runtime updates and conflicting flags. Actual games and +host adapter selection are not simulated by these tests. + The ignored `forwarding_benchmark` measures the synchronous transit routine in release mode, excluding crypto and socket I/O. Run it explicitly as described in [routing.md](routing.md); it has no timing threshold in the default suite.