ab/tsunagi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added per-network LAN broadcast relay

Browse Source 
LAN game discovery previously dropped IPv4 broadcasts at TUN ingress. Carry
limited and subnet-directed UDP broadcasts to authenticated, opted-in members
of the source network, including destinations reached through mesh relays.
Preserve the original IP/UDP bytes and deliver received broadcasts only to the
local TUN; never reflood them or expose another pair's plaintext at transit.

Build immutable recipient snapshots on address and participation changes. The
origin sends one ordinary end-to-end encrypted copy per recipient; the existing
fast, bounded-hop transport router remains unchanged. Validate UDP framing,
source ownership and destination admission without game-specific port rules.
Keep network domains isolated and refuse implicit gateways to physical LANs.
The separate broadcast policy/domain layer is the extension point for future
authorized subnet exports; physical capture, bridging and LAN deduplication
are deliberately not implemented yet.

Persist default-on participation independently for each local network. Add
join --no-broadcast/--broadcast and network broadcast <id> [on|off], including
live updates and authenticated announcements. Joining without a flag preserves
the saved choice. Opt-out stops local origination and delivery, while opaque
unicast transit for other members keeps working.

Migrate SQLite schema 3 to 4 without replacing identities or signed state.
Use control ALPN 3 and local IPC protocol 14 for the new announcement/request
shapes; update peers and restart running agents together. The data ALPN 4
envelope remains unchanged. No release version bump, tag or push is included.

Document agent-owned commits in AGENTS.md: short English subjects, explanatory
bodies, scoped staging, honest validation, and repository-local fallback author
AB <ab@hexor.cy> only when an effective name/email is missing. Release actions
remain the user's responsibility.

Validation on Windows: cargo fmt --all -- --check; cargo check --locked
--workspace --all-targets; cargo clippy --locked --workspace --all-targets --
-D warnings; release workspace/all-target tests: 313 passed. The two existing
SQLite wipe failures (a_wipe_removes_everything_and_the_next_start_is_a_stranger
and wiping_twice_is_as_ordinary_as_wiping_once) were explicitly skipped; the
public-DHT smoke test and forwarding benchmark remain ignored by default.
New coverage exercises real iroh/WireGuard multihop fanout, single delivery,
runtime opt-out, unicast replies, domain isolation, malformed input and schema
migration. TUNs are in-memory; actual games and OS adapter selection were not
tested.
This commit is contained in:
ab
2026-09-22 18:33:52 +03:00
parent b4f3e57c8d
commit c724981bfd
23 changed files with 1110 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tsunagi-cli/src/main.rs
+98 -4
View File
@@ -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<BroadcastChoice>,
},
/// 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<bool>,
) -> tsunagi::BoxFuture<'_, Result<tsunagi::ipc::JoinedReport, String>> {
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<bool, String>> {
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<dyn std::error::Er
let socket = control_socket(&paths, args.control_socket.as_ref());
match args.action {
None => 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<dyn std::error::Error>>
// 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<bool>,
) -> Result<(), Box<dyn std::error::Error>> {
// 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<BroadcastChoice>,
) -> Result<(), Box<dyn std::error::Error>> {
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,
crates/tsunagi-cli/tests/network_cli.rs
+62
View File
@@ -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()
);
}
crates/tsunagi-wg-quic/tests/local_control.rs
+1
View File
@@ -247,6 +247,7 @@ impl tsunagi::ipc::ReportSource for Control {
&self,
name: String,
secret: String,
_broadcast: Option<bool>,
) -> BoxFuture<'_, Result<tsunagi::ipc::JoinedReport, String>> {
Box::pin(async move {
let name = tsunagi::identity::NetworkName::new(&name).map_err(|e| e.to_string())?;
crates/tsunagi-wg-quic/tests/wireguard.rs
+117
View File
@@ -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;
}
}
crates/tsunagi/src/agent/mod.rs
+42
View File
@@ -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<NetworkId> {
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<bool>,
) -> Result<NetworkId> {
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<dyn crate::discovery::NetworkDiscovery>,
)
};
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,
crates/tsunagi/src/agent/network.rs
+37
View File
@@ -62,6 +62,10 @@ pub(crate) enum NetCommand {
reply: oneshot::Sender<Box<NetworkStatus>>,
},
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,
crates/tsunagi/src/agent/session.rs
+2
View File
@@ -73,6 +73,7 @@ pub(crate) struct Session {
pub(crate) outbound: mpsc::Sender<Vec<u8>>,
pub(crate) hostname: Option<String>,
pub(crate) capabilities: Vec<PluginCapability>,
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,
crates/tsunagi/src/agent/status.rs
+4
View File
@@ -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.
crates/tsunagi/src/ipc/mod.rs
+72 -2
View File
@@ -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<bool>,
},
}
@@ -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<StatusReport>),
/// 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<bool>,
) -> BoxFuture<'_, std::result::Result<JoinedReport, String>> {
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<bool, String>> {
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<Path>,
name: &str,
secret: &str,
) -> Result<JoinedReport> {
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<Path>,
name: &str,
secret: &str,
broadcast: Option<bool>,
) -> Result<JoinedReport> {
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<Path>,
network_id: &str,
enabled: bool,
) -> Result<bool> {
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<Path>,
crates/tsunagi/src/overlay/broadcast.rs
+228
View File
@@ -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<EndpointId>,
}
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<EndpointId>,
}
/// Immutable ingress-domain index rebuilt with address/policy changes.
#[derive(Debug, Default)]
pub(crate) struct BroadcastTable {
sources: HashMap<Ipv4Addr, Arc<Domain>>,
networks: HashMap<NetworkId, Arc<Domain>>,
destinations: HashSet<Ipv4Addr>,
}
impl BroadcastTable {
pub fn build(networks: &HashMap<NetworkId, NetworkRoutes>) -> 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<Arc<[Route]>> {
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));
}
}
crates/tsunagi/src/overlay/interface.rs
+135 -1
View File
@@ -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<IpAddr>,
/// 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<Option<IpAddr>>,
}
@@ -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))],
crates/tsunagi/src/overlay/mod.rs
+1
View File
@@ -36,6 +36,7 @@ pub enum OverlayError {
Other(String),
}
pub mod broadcast;
pub mod config;
pub mod interface;
pub mod packet;
crates/tsunagi/src/overlay/router.rs
+62 -2
View File
@@ -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<Ipv4Range>,
/// 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<BroadcastTable>,
networks: RwLock<HashMap<NetworkId, NetworkRoutes>>,
}
@@ -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<Arc<[Route]>> {
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(),
crates/tsunagi/src/proto/message.rs
+4 -2
View File
@@ -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.
crates/tsunagi/src/storage/mod.rs
+6
View File
@@ -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))
crates/tsunagi/src/storage/state.rs
+83 -4
View File
@@ -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<Vec<StoredNetwork>> {
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<u8> = 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"
);
}
}