Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d941cdac7a |
No files matched your search
@@ -112,6 +112,23 @@ fresh masking profile and outer keys for each pair connection attempt, including
|
||||
reentry after fallback. Sustained data loss triggers switching independently of
|
||||
the iroh control connection. [Configuration and limits](docs/amnezia.md).
|
||||
|
||||
Change a network's protocol selection without editing its service or restarting:
|
||||
|
||||
```sh
|
||||
tsng protocols # available protocols and settings
|
||||
tsng protocols home # this device's choice for home
|
||||
tsng protocols home awg,wg-quic # prefer AWG, retain QUIC as fallback
|
||||
tsng protocols home none # control plane only
|
||||
```
|
||||
|
||||
Exact network names, full ids and unique id prefixes work here. The choice is
|
||||
local to this device, saved per network and applied immediately; an offline
|
||||
change applies on the next start. In the tray, expand **protocols** on a network
|
||||
tile to toggle protocols or change their preference order. Both endpoints need
|
||||
a common protocol at the same wire version. `tsng up --protocol` supplies the
|
||||
initial defaults for networks without a saved choice; it does not restrict
|
||||
which built-in protocols can be enabled later.
|
||||
|
||||
## Platforms
|
||||
|
||||
| Platform | Build | CLI `tsng` | Tray | Exit node: offer / use | Packages |
|
||||
|
||||
@@ -25,6 +25,7 @@ use tsunagi_wg_quic::WireguardConfig;
|
||||
struct Node {
|
||||
agent: Agent,
|
||||
plugin: Arc<AwgPlugin>,
|
||||
quic: Option<Arc<tsunagi_wg_quic::WireguardPlugin>>,
|
||||
transport: Arc<AwgTransport>,
|
||||
tuns: MemoryTunFactory,
|
||||
loss: Arc<AtomicBool>,
|
||||
@@ -47,6 +48,16 @@ impl Node {
|
||||
interface: &str,
|
||||
fallback: bool,
|
||||
mtu: u32,
|
||||
) -> Self {
|
||||
Self::open_selected(root, discovery, interface, fallback, mtu, None).await
|
||||
}
|
||||
async fn open_selected(
|
||||
root: &Path,
|
||||
discovery: &SharedMemoryDiscovery,
|
||||
interface: &str,
|
||||
fallback: bool,
|
||||
mtu: u32,
|
||||
default_protocols: Option<Vec<String>>,
|
||||
) -> Self {
|
||||
let identity =
|
||||
tsunagi::storage::StateStore::open(tsunagi::StoragePaths::under(root).state_db())
|
||||
@@ -70,9 +81,9 @@ impl Node {
|
||||
let tuns = MemoryTunFactory::new();
|
||||
let mut config = config_with(root, discovery)
|
||||
.with_interface(Arc::new(tuns.clone()), interface, mtu)
|
||||
.with_overlay_ipv4_range(Some(tsunagi::state::DEFAULT_IPV4_RANGE))
|
||||
.with_plugin(plugin.clone() as Arc<dyn IpPlugin>)
|
||||
.with_custom_transport(AWG_PROTOCOL, faults);
|
||||
let mut quic_plugin = None;
|
||||
if fallback {
|
||||
let quic = tsunagi_wg_quic::WireguardPlugin::open(
|
||||
WireguardConfig::new(root.join("wg-quic"))
|
||||
@@ -81,8 +92,10 @@ impl Node {
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
config = config.with_plugin(quic);
|
||||
config = config.with_plugin(quic.clone());
|
||||
quic_plugin = Some(quic);
|
||||
}
|
||||
config.default_protocols = default_protocols;
|
||||
let agent = Agent::spawn(config).await.unwrap();
|
||||
let inbound_agent = agent.clone();
|
||||
let inbound_transport = transport.clone();
|
||||
@@ -96,6 +109,7 @@ impl Node {
|
||||
Self {
|
||||
agent,
|
||||
plugin,
|
||||
quic: quic_plugin,
|
||||
transport,
|
||||
tuns,
|
||||
loss,
|
||||
@@ -163,6 +177,235 @@ async fn exchange(a: &Node, b: &Node, id: NetworkId) {
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_control_only_agent_enables_real_data_without_restart_and_cancels_stale_attempts() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let roots = [TempDir::new().unwrap(), TempDir::new().unwrap()];
|
||||
let a = Node::open_selected(
|
||||
roots[0].path(),
|
||||
&discovery,
|
||||
"initial-none-a",
|
||||
true,
|
||||
1280,
|
||||
Some(vec![]),
|
||||
)
|
||||
.await;
|
||||
let b = Node::open_selected(
|
||||
roots[1].path(),
|
||||
&discovery,
|
||||
"initial-none-b",
|
||||
true,
|
||||
1280,
|
||||
Some(vec![]),
|
||||
)
|
||||
.await;
|
||||
let (name, secret) = network("enable-after-control-only");
|
||||
let id = a.agent.join_network(&name, &secret).await.unwrap();
|
||||
b.agent.join_network(&name, &secret).await.unwrap();
|
||||
wait_for_peers(&a.agent, id, 1).await;
|
||||
wait_for_peers(&b.agent, id, 1).await;
|
||||
assert!(a.agent.overlay().is_none());
|
||||
assert!(b.agent.overlay().is_none());
|
||||
for node in [&a, &b] {
|
||||
assert!(
|
||||
node.agent
|
||||
.network_status(id)
|
||||
.await
|
||||
.unwrap()
|
||||
.protocols
|
||||
.is_empty()
|
||||
);
|
||||
node.agent
|
||||
.set_protocols(id, vec!["awg".into()])
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(node.agent.overlay().is_some());
|
||||
}
|
||||
a.ready(id, 1).await;
|
||||
b.ready(id, 1).await;
|
||||
exchange(&a, &b, id).await;
|
||||
let before = a.agent.overlay().unwrap().interface;
|
||||
// Changes race unfinished opens and prepared plugin callbacks; the last
|
||||
// selection must win even if an earlier open has completed in its queue.
|
||||
for selection in [
|
||||
vec!["wg-quic".into()],
|
||||
vec![],
|
||||
vec!["awg".into()],
|
||||
vec![],
|
||||
vec!["awg".into()],
|
||||
] {
|
||||
a.agent.set_protocols(id, selection).await.unwrap();
|
||||
}
|
||||
a.ready(id, 1).await;
|
||||
b.ready(id, 1).await;
|
||||
exchange(&a, &b, id).await;
|
||||
assert_eq!(a.agent.overlay().unwrap().interface, before);
|
||||
for node in [&a, &b] {
|
||||
let status = node.agent.network_status(id).await.unwrap();
|
||||
assert_eq!(status.protocols, ["awg"]);
|
||||
assert_eq!(status.metrics.sessions_established, 1);
|
||||
assert_eq!(status.metrics.disconnects, 0);
|
||||
}
|
||||
a.agent.shutdown().await;
|
||||
b.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn runtime_protocol_selection_replaces_real_tunnels_and_survives_restart() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let roots = [TempDir::new().unwrap(), TempDir::new().unwrap()];
|
||||
let a = Node::open_with_fallback(roots[0].path(), &discovery, "runtime-a", true).await;
|
||||
let b = Node::open_with_fallback(roots[1].path(), &discovery, "runtime-b", true).await;
|
||||
let (name, secret) = network("runtime-protocols");
|
||||
let id = a.agent.join_network(&name, &secret).await.unwrap();
|
||||
b.agent.join_network(&name, &secret).await.unwrap();
|
||||
a.ready(id, 1).await;
|
||||
b.ready(id, 1).await;
|
||||
exchange(&a, &b, id).await;
|
||||
|
||||
let (other_name, other_secret, other_id) = loop {
|
||||
let (name, secret) = network("unchanged-runtime-network");
|
||||
let other_id = tsunagi::identity::NetworkKeys::derive(&name, &secret).network_id();
|
||||
if tsunagi::state::derived_ipv4_range(id) != tsunagi::state::derived_ipv4_range(other_id) {
|
||||
break (name, secret, other_id);
|
||||
}
|
||||
};
|
||||
a.agent
|
||||
.join_network(&other_name, &other_secret)
|
||||
.await
|
||||
.unwrap();
|
||||
b.agent
|
||||
.join_network(&other_name, &other_secret)
|
||||
.await
|
||||
.unwrap();
|
||||
a.ready(other_id, 1).await;
|
||||
b.ready(other_id, 1).await;
|
||||
let other_before = a
|
||||
.agent
|
||||
.network_status(other_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.metrics
|
||||
.data_links_established;
|
||||
|
||||
for node in [&a, &b] {
|
||||
let before = node.agent.network_status(id).await.unwrap().protocols;
|
||||
for invalid in [vec!["unknown".into()], vec!["awg".into(), "awg".into()]] {
|
||||
assert!(node.agent.set_protocols(id, invalid).await.is_err());
|
||||
assert_eq!(
|
||||
node.agent.network_status(id).await.unwrap().protocols,
|
||||
before
|
||||
);
|
||||
}
|
||||
}
|
||||
// Exercise updates initiated by either endpoint-id role, including an
|
||||
// empty set. Membership and the control session stay alive throughout.
|
||||
for node in [&a, &b] {
|
||||
node.agent.set_protocols(id, Vec::new()).await.unwrap();
|
||||
wait_until("peer accepts control-only selection", || async {
|
||||
let other = if node.agent.endpoint_id() == a.agent.endpoint_id() {
|
||||
&b
|
||||
} else {
|
||||
&a
|
||||
};
|
||||
let status = other.agent.network_status(id).await.ok()?;
|
||||
(status.peers.len() == 1
|
||||
&& status.peers[0].protocols.is_empty()
|
||||
&& matches!(status.peers[0].data_path, tsunagi::agent::DataPath::None))
|
||||
.then_some(())
|
||||
})
|
||||
.await;
|
||||
assert_eq!(node.agent.network_status(id).await.unwrap().peers.len(), 1);
|
||||
node.agent
|
||||
.set_protocols(id, vec!["wg-quic".into(), "awg".into()])
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
a.agent
|
||||
.set_protocols(id, vec!["wg-quic".into()])
|
||||
.await
|
||||
.unwrap();
|
||||
wait_until("real QUIC replaces AWG", || async {
|
||||
let a_status = a.agent.network_status(id).await.ok()?;
|
||||
let b_status = b.agent.network_status(id).await.ok()?;
|
||||
(a_status.peers[0].protocols == ["wg-quic"]
|
||||
&& b_status.peers[0].protocols == ["wg-quic"]
|
||||
&& !matches!(a_status.peers[0].data_path, tsunagi::agent::DataPath::None)
|
||||
&& !matches!(b_status.peers[0].data_path, tsunagi::agent::DataPath::None)
|
||||
&& a.transport.overview(id).iter().all(|p| p.closed)
|
||||
&& b.transport.overview(id).iter().all(|p| p.closed))
|
||||
.then_some(())
|
||||
})
|
||||
.await;
|
||||
// Get addresses from the system allocation: the disabled AWG plugin no
|
||||
// longer has a live view of this network.
|
||||
for (source, dest) in [(&a, &b), (&b, &a)] {
|
||||
wait_until("QUIC end-to-end handshake", || async {
|
||||
source
|
||||
.quic
|
||||
.as_ref()?
|
||||
.overview(id)?
|
||||
.peers
|
||||
.iter()
|
||||
.any(|p| p.tunnel.as_ref().is_some_and(|t| t.health.is_up()))
|
||||
.then_some(())
|
||||
})
|
||||
.await;
|
||||
let addr = |n: &Node| {
|
||||
n.quic
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.overview(id)
|
||||
.unwrap()
|
||||
.overlay_address_v4
|
||||
.unwrap()
|
||||
};
|
||||
let bytes = packet(addr(source), addr(dest), 1280);
|
||||
source.tun().push_from_os(bytes.clone());
|
||||
assert_eq!(
|
||||
tokio::time::timeout(Duration::from_secs(5), dest.tun().pop_to_os())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap(),
|
||||
bytes
|
||||
);
|
||||
}
|
||||
assert_eq!(
|
||||
a.agent
|
||||
.network_status(other_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.metrics
|
||||
.data_links_established,
|
||||
other_before
|
||||
);
|
||||
exchange(&a, &b, other_id).await;
|
||||
a.agent.shutdown().await;
|
||||
let restarted = Node::open_with_fallback(roots[0].path(), &discovery, "runtime-a", true).await;
|
||||
wait_for_peers(&restarted.agent, id, 1).await;
|
||||
assert_eq!(
|
||||
restarted.agent.network_status(id).await.unwrap().protocols,
|
||||
["wg-quic"]
|
||||
);
|
||||
assert_eq!(
|
||||
b.agent.network_status(other_id).await.unwrap().protocols,
|
||||
["awg", "wg-quic"]
|
||||
);
|
||||
restarted.ready(other_id, 1).await;
|
||||
b.ready(other_id, 1).await;
|
||||
exchange(&restarted, &b, other_id).await;
|
||||
restarted
|
||||
.agent
|
||||
.set_protocols(id, vec!["awg".into()])
|
||||
.await
|
||||
.unwrap();
|
||||
restarted.ready(id, 1).await;
|
||||
b.ready(id, 1).await;
|
||||
exchange(&restarted, &b, id).await;
|
||||
restarted.agent.shutdown().await;
|
||||
b.agent.shutdown().await;
|
||||
}
|
||||
#[tokio::test]
|
||||
async fn real_pairs_carry_full_size_packets_and_restart_with_fresh_profiles() {
|
||||
tsunagi::testing::init_tracing();
|
||||
|
||||
@@ -58,6 +58,8 @@ use shutdown::Shutdown;
|
||||
/// Summary of a configured network, whether or not it is running.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ConfiguredNetwork {
|
||||
/// Enabled local data-plane protocols, in preference order.
|
||||
pub protocols: Vec<String>,
|
||||
/// Saved local broadcast participation.
|
||||
pub broadcast: bool,
|
||||
/// Public network identifier.
|
||||
@@ -104,6 +106,37 @@ struct Inner {
|
||||
}
|
||||
|
||||
impl Inner {
|
||||
/// Called under the lifecycle lock after startup: an initially
|
||||
/// control-only agent creates its one interface on first enablement.
|
||||
async fn ensure_interface(self: &Arc<Self>) {
|
||||
if self.interface.get().is_some() {
|
||||
return;
|
||||
}
|
||||
if let Some(factory) = self.config.tun_factory.clone() {
|
||||
let carrier = Arc::new(Carrier(Arc::downgrade(self))) as Arc<dyn PacketCarrier>;
|
||||
match crate::overlay::Interface::start(
|
||||
factory,
|
||||
self.config.interface_name.clone(),
|
||||
self.config.interface_mtu,
|
||||
Arc::clone(&self.routes),
|
||||
carrier,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(interface) => {
|
||||
let _ = self.interface.set(Arc::new(interface));
|
||||
}
|
||||
Err(err) => {
|
||||
let _ = self.events.send(Event::PluginError {
|
||||
network: NetworkId::from_bytes([0u8; 32]),
|
||||
protocol: "overlay".into(),
|
||||
reason: err.to_string(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The name this agent currently answers to.
|
||||
fn read_hostname(&self) -> String {
|
||||
match self.hostname.read() {
|
||||
@@ -265,12 +298,90 @@ impl TransportContext for TransportCtx {
|
||||
}
|
||||
|
||||
impl Agent {
|
||||
fn effective_protocols(&self, saved: Option<&Vec<String>>) -> Vec<String> {
|
||||
saved
|
||||
.or(self.inner.config.default_protocols.as_ref())
|
||||
.cloned()
|
||||
.unwrap_or_else(|| {
|
||||
self.inner
|
||||
.config
|
||||
.plugins
|
||||
.iter()
|
||||
.map(|p| p.protocol_id().to_owned())
|
||||
.collect()
|
||||
})
|
||||
}
|
||||
|
||||
fn selected_plugins(&self, protocols: &[String]) -> Vec<crate::dataplane::SharedPlugin> {
|
||||
protocols
|
||||
.iter()
|
||||
.filter_map(|name| {
|
||||
self.inner
|
||||
.config
|
||||
.plugins
|
||||
.iter()
|
||||
.find(|p| p.protocol_id() == name)
|
||||
.cloned()
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Saves and immediately applies a network's ordered local protocol set.
|
||||
/// An empty set disables data traffic while retaining control sessions.
|
||||
pub async fn set_protocols(&self, network_id: NetworkId, protocols: Vec<String>) -> Result<()> {
|
||||
crate::dataplane::validate_protocols(&protocols, self.inner.limits.max_capabilities)?;
|
||||
for name in &protocols {
|
||||
if !self
|
||||
.inner
|
||||
.config
|
||||
.plugins
|
||||
.iter()
|
||||
.any(|p| p.protocol_id() == name)
|
||||
{
|
||||
return Err(Error::InvalidProtocols(format!(
|
||||
"this agent has no `{name}` plugin"
|
||||
)));
|
||||
}
|
||||
}
|
||||
let _lifecycle = self.inner.lifecycle.lock().await;
|
||||
self.inner
|
||||
.storage
|
||||
.set_protocols(network_id, protocols.clone())
|
||||
.await?;
|
||||
if self.is_active(network_id).await {
|
||||
let (reply, receive) = oneshot::channel();
|
||||
if !protocols.is_empty() {
|
||||
self.inner.ensure_interface().await;
|
||||
}
|
||||
self.command(
|
||||
network_id,
|
||||
NetCommand::SetProtocols {
|
||||
protocols,
|
||||
interface: self.inner.interface.get().cloned(),
|
||||
reply,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
receive.await.map_err(|_| Error::Stopped)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
/// Starts an agent.
|
||||
///
|
||||
/// Opens the state store (taking its ownership lock), restores the
|
||||
/// persistent device identity, binds the iroh endpoint and activates every
|
||||
/// configured network whose auto-start flag is set.
|
||||
pub async fn spawn(config: AgentConfig) -> Result<Self> {
|
||||
if let Some(protocols) = &config.default_protocols {
|
||||
crate::dataplane::validate_protocols(protocols, config.limits.max_capabilities)?;
|
||||
for name in protocols {
|
||||
if !config.plugins.iter().any(|p| p.protocol_id() == name) {
|
||||
return Err(Error::InvalidProtocols(format!(
|
||||
"this agent has no `{name}` plugin"
|
||||
)));
|
||||
}
|
||||
}
|
||||
}
|
||||
let storage = Storage::open(&config.paths)?;
|
||||
let identity = storage.device_identity().await?;
|
||||
let adapter = EndpointAdapter::bind(&config, &identity).await?;
|
||||
@@ -303,31 +414,13 @@ impl Agent {
|
||||
config,
|
||||
});
|
||||
|
||||
// One interface for the agent, if it was given a way to make one.
|
||||
// Failing to is reported and not fatal: the control plane works, and
|
||||
// so do the protocols, they just have nowhere to put packets.
|
||||
if let Some(factory) = inner.config.tun_factory.clone() {
|
||||
let carrier = Arc::new(Carrier(Arc::downgrade(&inner))) as Arc<dyn PacketCarrier>;
|
||||
match crate::overlay::Interface::start(
|
||||
factory,
|
||||
inner.config.interface_name.clone(),
|
||||
inner.config.interface_mtu,
|
||||
Arc::clone(&inner.routes),
|
||||
carrier,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(interface) => {
|
||||
let _ = inner.interface.set(Arc::new(interface));
|
||||
}
|
||||
Err(err) => {
|
||||
let _ = inner.events.send(Event::PluginError {
|
||||
network: NetworkId::from_bytes([0u8; 32]),
|
||||
protocol: "overlay".into(),
|
||||
reason: err.to_string(),
|
||||
});
|
||||
}
|
||||
}
|
||||
if inner
|
||||
.config
|
||||
.default_protocols
|
||||
.as_ref()
|
||||
.is_none_or(|names| !names.is_empty())
|
||||
{
|
||||
inner.ensure_interface().await;
|
||||
}
|
||||
|
||||
// The data plane rides on iroh too, which is where it gets hole
|
||||
@@ -364,8 +457,9 @@ impl Agent {
|
||||
if !inner.config.plugins.is_empty() {
|
||||
let (plugin_tx, plugin_rx) = mpsc::channel(64);
|
||||
let sink = inner
|
||||
.interface
|
||||
.get()
|
||||
.config
|
||||
.tun_factory
|
||||
.as_ref()
|
||||
.map(|_| Arc::new(Sink(Arc::downgrade(&inner))) as Arc<dyn PacketSink>);
|
||||
let context = PluginContext::new(plugin_tx, inner.identity.endpoint_id(), sink);
|
||||
for plugin in &inner.config.plugins {
|
||||
@@ -824,6 +918,12 @@ impl Agent {
|
||||
};
|
||||
|
||||
let broadcast = stored.as_ref().is_none_or(|stored| stored.broadcast);
|
||||
let protocols =
|
||||
self.effective_protocols(stored.as_ref().and_then(|s| s.protocols.as_ref()));
|
||||
let plugins = self.selected_plugins(&protocols);
|
||||
if !plugins.is_empty() {
|
||||
self.inner.ensure_interface().await;
|
||||
}
|
||||
let exit_offer = stored.as_ref().is_some_and(|stored| stored.exit_node);
|
||||
let exit_via = stored
|
||||
.as_ref()
|
||||
@@ -842,7 +942,8 @@ impl Agent {
|
||||
discovery,
|
||||
discovery_interval: self.inner.config.discovery_interval,
|
||||
discovery_policy: self.inner.config.discovery_policy.clone(),
|
||||
plugins: self.inner.config.plugins.clone(),
|
||||
available_plugins: self.inner.config.plugins.clone(),
|
||||
plugins: plugins.clone(),
|
||||
routes: Arc::clone(&self.inner.routes),
|
||||
interface: self.inner.interface.get().cloned(),
|
||||
hostname: self.inner.read_hostname(),
|
||||
@@ -856,7 +957,7 @@ impl Agent {
|
||||
networks.insert(network_id, handle);
|
||||
drop(networks);
|
||||
|
||||
for plugin in &self.inner.config.plugins {
|
||||
for plugin in &plugins {
|
||||
plugin.on_network_activated(network_id);
|
||||
}
|
||||
|
||||
@@ -997,6 +1098,7 @@ impl Agent {
|
||||
.await?
|
||||
.into_iter()
|
||||
.map(|stored| ConfiguredNetwork {
|
||||
protocols: self.effective_protocols(stored.protocols.as_ref()),
|
||||
broadcast: stored.broadcast,
|
||||
network_id: stored.network_id,
|
||||
name: stored.name,
|
||||
@@ -1075,6 +1177,7 @@ impl Agent {
|
||||
let range = self.desired_range(stored.network_id).await?;
|
||||
let range_conflicts = self.range_conflicts(stored.network_id, range).await?;
|
||||
networks.push(NetworkStatus {
|
||||
protocols: self.effective_protocols(stored.protocols.as_ref()),
|
||||
auto_start: stored.auto_start,
|
||||
broadcast: stored.broadcast,
|
||||
exit_offer: stored.exit_node,
|
||||
|
||||
@@ -72,6 +72,11 @@ pub(crate) enum NetCommand {
|
||||
enabled: bool,
|
||||
reply: oneshot::Sender<()>,
|
||||
},
|
||||
SetProtocols {
|
||||
interface: Option<Arc<crate::overlay::Interface>>,
|
||||
protocols: Vec<String>,
|
||||
reply: oneshot::Sender<()>,
|
||||
},
|
||||
/// Offer this agent as an exit node in this network, or stop.
|
||||
SetExitOffer {
|
||||
enabled: bool,
|
||||
@@ -122,6 +127,7 @@ impl std::fmt::Debug for NetCommand {
|
||||
NetCommand::Status { .. } => f.write_str("Status"),
|
||||
NetCommand::Recheck => f.write_str("Recheck"),
|
||||
NetCommand::SetBroadcast { enabled, .. } => write!(f, "SetBroadcast({enabled})"),
|
||||
NetCommand::SetProtocols { protocols, .. } => write!(f, "SetProtocols({protocols:?})"),
|
||||
NetCommand::SetExitOffer { enabled, .. } => write!(f, "SetExitOffer({enabled})"),
|
||||
NetCommand::SetExitVia { via, .. } => write!(f, "SetExitVia({})", via.is_some()),
|
||||
NetCommand::Reannounce => f.write_str("Reannounce"),
|
||||
@@ -171,6 +177,7 @@ pub(crate) struct RuntimeParams {
|
||||
pub(crate) discovery_interval: Duration,
|
||||
pub(crate) discovery_policy: DiscoveryPolicy,
|
||||
pub(crate) plugins: Vec<SharedPlugin>,
|
||||
pub(crate) available_plugins: Vec<SharedPlugin>,
|
||||
/// Who holds which overlay address, shared with every other network.
|
||||
pub(crate) routes: Arc<crate::overlay::RoutingTable>,
|
||||
/// The one interface, when the agent has one.
|
||||
@@ -192,6 +199,7 @@ pub(crate) struct RuntimeParams {
|
||||
|
||||
/// Outcome of one attempt to open a data plane link.
|
||||
struct LinkOutcome {
|
||||
generation: u64,
|
||||
peer: EndpointId,
|
||||
protocol: String,
|
||||
result: Result<SharedLink, String>,
|
||||
@@ -332,6 +340,9 @@ struct Runtime {
|
||||
announced_reach: HashSet<crate::proto::message::ReachableLink>,
|
||||
/// Links currently being opened, so we do not start two.
|
||||
opening: HashSet<(EndpointId, String)>,
|
||||
link_tasks: HashMap<(EndpointId, String), tokio::task::AbortHandle>,
|
||||
link_attempts: HashMap<(EndpointId, String), u64>,
|
||||
next_link_attempt: u64,
|
||||
link_results_tx: mpsc::Sender<LinkOutcome>,
|
||||
link_results_rx: mpsc::Receiver<LinkOutcome>,
|
||||
/// Protocol dial failure count and backoff cooldown per peer and protocol.
|
||||
@@ -425,6 +436,9 @@ impl Runtime {
|
||||
local_capabilities: Vec::new(),
|
||||
announced_reach: HashSet::new(),
|
||||
opening: HashSet::new(),
|
||||
link_tasks: HashMap::new(),
|
||||
link_attempts: HashMap::new(),
|
||||
next_link_attempt: 0,
|
||||
link_results_tx,
|
||||
link_results_rx,
|
||||
protocol_dial_states: HashMap::new(),
|
||||
@@ -511,6 +525,10 @@ impl Runtime {
|
||||
link.close();
|
||||
}
|
||||
self.links.clear();
|
||||
for (_, task) in self.link_tasks.drain() {
|
||||
task.abort();
|
||||
}
|
||||
self.opening.clear();
|
||||
self.pending_health.clear();
|
||||
self.last_failed_protocol.clear();
|
||||
self.protocol_dial_states.clear();
|
||||
@@ -553,6 +571,60 @@ impl Runtime {
|
||||
NetCommand::Status { reply } => {
|
||||
let _ = reply.send(Box::new(self.status()));
|
||||
}
|
||||
NetCommand::SetProtocols {
|
||||
protocols,
|
||||
interface,
|
||||
reply,
|
||||
} => {
|
||||
self.params.interface = interface;
|
||||
let current: Vec<_> = self
|
||||
.params
|
||||
.plugins
|
||||
.iter()
|
||||
.map(|p| p.protocol_id().to_owned())
|
||||
.collect();
|
||||
if current != protocols {
|
||||
let peers: Vec<_> = self.sessions.keys().copied().collect();
|
||||
for peer in &peers {
|
||||
self.reset_peer_data(*peer, "local protocol selection changed");
|
||||
}
|
||||
for plugin in &self.params.plugins {
|
||||
if !protocols.iter().any(|name| name == plugin.protocol_id()) {
|
||||
plugin.on_network_deactivated(self.network_id);
|
||||
}
|
||||
}
|
||||
self.params.plugins = protocols
|
||||
.iter()
|
||||
.filter_map(|name| {
|
||||
self.params
|
||||
.available_plugins
|
||||
.iter()
|
||||
.find(|p| p.protocol_id() == name)
|
||||
.cloned()
|
||||
})
|
||||
.collect();
|
||||
for plugin in &self.params.plugins {
|
||||
if !current.iter().any(|name| name == plugin.protocol_id()) {
|
||||
plugin.on_network_activated(self.network_id);
|
||||
}
|
||||
}
|
||||
self.hub.set_priority(protocols);
|
||||
self.publish_allocations().await;
|
||||
for peer in peers {
|
||||
let caps = self
|
||||
.sessions
|
||||
.get(&peer)
|
||||
.map(|s| s.capabilities.clone())
|
||||
.unwrap_or_default();
|
||||
self.dispatch_capabilities(peer, &caps);
|
||||
}
|
||||
self.reannounce();
|
||||
self.ensure_links();
|
||||
self.update_paths();
|
||||
self.announce_reach(true);
|
||||
}
|
||||
let _ = reply.send(());
|
||||
}
|
||||
NetCommand::SetBroadcast { enabled, reply } => {
|
||||
self.params.broadcast = enabled;
|
||||
self.update_broadcast();
|
||||
@@ -584,7 +656,22 @@ impl Runtime {
|
||||
let _ = reply.send(outcome);
|
||||
}
|
||||
NetCommand::Recheck => self.discovery_round().await,
|
||||
NetCommand::Reannounce => self.reannounce(),
|
||||
NetCommand::Reannounce => {
|
||||
// A newly enabled plugin may prepare its key asynchronously.
|
||||
// Replay allocations and peer identities once it is ready.
|
||||
self.publish_allocations().await;
|
||||
let peers: Vec<_> = self
|
||||
.sessions
|
||||
.iter()
|
||||
.map(|(peer, s)| (*peer, s.capabilities.clone()))
|
||||
.collect();
|
||||
for (peer, caps) in peers {
|
||||
self.dispatch_capabilities(peer, &caps);
|
||||
}
|
||||
self.reannounce();
|
||||
self.ensure_links();
|
||||
self.update_paths();
|
||||
}
|
||||
NetCommand::Release { reply } => {
|
||||
// A tombstone, so the address and the name this agent held
|
||||
// are freed for somebody else instead of staying reserved
|
||||
@@ -1578,13 +1665,16 @@ impl Runtime {
|
||||
if self.links.contains_key(&key) || self.opening.contains(&key) {
|
||||
continue;
|
||||
}
|
||||
self.opening.insert(key);
|
||||
self.opening.insert(key.clone());
|
||||
|
||||
let results = self.link_results_tx.clone();
|
||||
let transport = Arc::clone(&transport);
|
||||
let network = self.network_id;
|
||||
let shutdown = self.shutdown.clone();
|
||||
tokio::spawn(async move {
|
||||
self.next_link_attempt = self.next_link_attempt.wrapping_add(1);
|
||||
let generation = self.next_link_attempt;
|
||||
self.link_attempts.insert(key.clone(), generation);
|
||||
let task = tokio::spawn(async move {
|
||||
let result = tokio::select! {
|
||||
biased;
|
||||
_ = shutdown.wait() => Err("network deactivated".to_string()),
|
||||
@@ -1594,12 +1684,14 @@ impl Runtime {
|
||||
};
|
||||
let _ = results
|
||||
.send(LinkOutcome {
|
||||
generation,
|
||||
peer,
|
||||
protocol,
|
||||
result,
|
||||
})
|
||||
.await;
|
||||
});
|
||||
self.link_tasks.insert(key, task.abort_handle());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1834,7 +1926,15 @@ impl Runtime {
|
||||
|
||||
fn handle_link_result(&mut self, outcome: LinkOutcome) {
|
||||
let key = (outcome.peer, outcome.protocol.clone());
|
||||
if self.link_attempts.get(&key) != Some(&outcome.generation) {
|
||||
if let Ok(link) = outcome.result {
|
||||
link.close();
|
||||
}
|
||||
return;
|
||||
}
|
||||
self.link_attempts.remove(&key);
|
||||
self.opening.remove(&key);
|
||||
self.link_tasks.remove(&key);
|
||||
match outcome.result {
|
||||
Ok(link) => {
|
||||
self.record_protocol_success(outcome.peer, &outcome.protocol);
|
||||
@@ -1903,12 +2003,17 @@ impl Runtime {
|
||||
link.close();
|
||||
return;
|
||||
}
|
||||
if !self
|
||||
.params
|
||||
.plugins
|
||||
.iter()
|
||||
.any(|plugin| plugin.protocol_id() == protocol)
|
||||
{
|
||||
if !self.params.plugins.iter().any(|plugin| {
|
||||
plugin.protocol_id() == protocol
|
||||
&& self.sessions.get(&peer).is_some_and(|s| {
|
||||
s.capabilities.iter().any(|c| {
|
||||
c.enabled
|
||||
&& c.protocol == protocol
|
||||
&& c.version == plugin.protocol_version()
|
||||
})
|
||||
})
|
||||
}) {
|
||||
link.close();
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -1966,14 +2071,25 @@ impl Runtime {
|
||||
|
||||
/// Drops every link to a peer.
|
||||
fn drop_links_for(&mut self, peer: EndpointId) {
|
||||
// The peer itself is going, not just a path to it, so the
|
||||
// protocol's link goes with it.
|
||||
self.reset_peer_data(peer, "peer session ended");
|
||||
self.hub.remove_peer(peer);
|
||||
self.reachable.remove(&peer);
|
||||
self.protocol_dial_states.retain(|(p, _), _| *p != peer);
|
||||
self.last_failed_protocol.remove(&peer);
|
||||
self.pending_health.retain(|(p, _), _| *p != peer);
|
||||
let keys: Vec<(EndpointId, String)> = self
|
||||
}
|
||||
|
||||
/// Retire only the data plane; authenticated membership and topology remain.
|
||||
fn reset_peer_data(&mut self, peer: EndpointId, reason: &str) {
|
||||
self.link_attempts.retain(|(id, _), _| *id != peer);
|
||||
self.opening.retain(|(id, _)| *id != peer);
|
||||
self.link_tasks.retain(|(id, _), task| {
|
||||
if *id == peer {
|
||||
task.abort();
|
||||
false
|
||||
} else {
|
||||
true
|
||||
}
|
||||
});
|
||||
self.hub.clear_peer_links(peer);
|
||||
let keys: Vec<_> = self
|
||||
.links
|
||||
.keys()
|
||||
.filter(|(id, _)| *id == peer)
|
||||
@@ -1987,9 +2103,15 @@ impl Runtime {
|
||||
network: self.network_id,
|
||||
peer,
|
||||
protocol: key.1,
|
||||
reason: "peer session ended".into(),
|
||||
reason: reason.into(),
|
||||
});
|
||||
}
|
||||
self.protocol_dial_states.retain(|(id, _), _| *id != peer);
|
||||
self.last_failed_protocol.remove(&peer);
|
||||
self.pending_health.retain(|(id, _), _| *id != peer);
|
||||
for plugin in &self.params.plugins {
|
||||
plugin.on_peer_gone(self.network_id, peer);
|
||||
}
|
||||
}
|
||||
|
||||
// --------------------------------------------------------------- sessions
|
||||
@@ -2182,9 +2304,6 @@ impl Runtime {
|
||||
self.watch_exit_via();
|
||||
self.update_paths();
|
||||
self.announce_reach(false);
|
||||
for plugin in &self.params.plugins {
|
||||
plugin.on_peer_gone(self.network_id, peer);
|
||||
}
|
||||
// Retry promptly, then back off if it keeps failing.
|
||||
let policy = &self.params.reconnect;
|
||||
let state = self
|
||||
@@ -2277,6 +2396,18 @@ impl Runtime {
|
||||
}
|
||||
ControlMessage::Announce(announcement) => {
|
||||
let capabilities = announcement.capabilities.clone();
|
||||
let changed = self.sessions.get(&peer).is_some_and(|session| {
|
||||
let selection = |caps: &[PluginCapability]| {
|
||||
caps.iter()
|
||||
.filter(|c| c.enabled)
|
||||
.map(|c| (c.protocol.clone(), c.version))
|
||||
.collect::<Vec<_>>()
|
||||
};
|
||||
selection(&session.capabilities) != selection(&capabilities)
|
||||
});
|
||||
if changed {
|
||||
self.reset_peer_data(peer, "peer protocol selection changed");
|
||||
}
|
||||
if let Some(session) = self.sessions.get_mut(&peer) {
|
||||
session.hostname = Some(announcement.hostname.clone());
|
||||
session.broadcast = announcement.broadcast;
|
||||
@@ -2289,6 +2420,7 @@ impl Runtime {
|
||||
self.dispatch_capabilities(peer, &capabilities);
|
||||
self.ensure_links();
|
||||
self.update_paths();
|
||||
self.announce_reach(false);
|
||||
}
|
||||
ControlMessage::Ping { seq, payload } => {
|
||||
let pong = ControlMessage::Pong {
|
||||
@@ -2522,6 +2654,12 @@ impl Runtime {
|
||||
members.dedup_by(|a, b| a.endpoint_id == b.endpoint_id);
|
||||
|
||||
NetworkStatus {
|
||||
protocols: self
|
||||
.params
|
||||
.plugins
|
||||
.iter()
|
||||
.map(|p| p.protocol_id().to_owned())
|
||||
.collect(),
|
||||
auto_start: true,
|
||||
broadcast: self.params.broadcast,
|
||||
exit_offer: self.params.exit_offer,
|
||||
|
||||
@@ -223,6 +223,8 @@ pub struct RangeConflict {
|
||||
/// Status of one network.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct NetworkStatus {
|
||||
/// Enabled local data-plane protocols, in preference order.
|
||||
pub protocols: Vec<String>,
|
||||
/// Whether this network is requested to start or resume automatically.
|
||||
pub auto_start: bool,
|
||||
/// Local broadcast participation in this network.
|
||||
|
||||
@@ -328,6 +328,9 @@ pub struct AgentConfig {
|
||||
pub reconnect: ReconnectPolicy,
|
||||
/// IP plugins whose capabilities are announced and dispatched.
|
||||
pub plugins: Vec<SharedPlugin>,
|
||||
/// Initial ordered selection for networks without a saved override.
|
||||
/// `None` enables all registered plugins; an empty list is control-only.
|
||||
pub default_protocols: Option<Vec<String>>,
|
||||
/// Custom packet transports registered per protocol.
|
||||
pub custom_transports:
|
||||
std::collections::HashMap<String, Arc<dyn crate::dataplane::PacketTransport>>,
|
||||
@@ -376,6 +379,7 @@ impl AgentConfig {
|
||||
limits: Limits::default(),
|
||||
reconnect: ReconnectPolicy::default(),
|
||||
plugins: Vec::new(),
|
||||
default_protocols: None,
|
||||
custom_transports: std::collections::HashMap::new(),
|
||||
tun_factory: None,
|
||||
interface_name: "tsun0".to_string(),
|
||||
|
||||
@@ -40,6 +40,33 @@ pub use tunnel::{GenericTunnelPlugin, TunnelCodec};
|
||||
/// Maximum length of a plugin protocol identifier.
|
||||
pub const MAX_PROTOCOL_ID_LEN: usize = 32;
|
||||
|
||||
/// Validates bounded local protocol preferences before storing or applying them.
|
||||
pub(crate) fn validate_protocols(protocols: &[String], max: usize) -> crate::error::Result<()> {
|
||||
if protocols.len() > max {
|
||||
return Err(crate::Error::InvalidProtocols(format!(
|
||||
"at most {max} protocols are allowed"
|
||||
)));
|
||||
}
|
||||
for (index, name) in protocols.iter().enumerate() {
|
||||
if name.is_empty()
|
||||
|| name.len() > MAX_PROTOCOL_ID_LEN
|
||||
|| !name
|
||||
.bytes()
|
||||
.all(|c| c.is_ascii_alphanumeric() || c == b'-' || c == b'_')
|
||||
{
|
||||
return Err(crate::Error::InvalidProtocols(
|
||||
"names must be 1–32 ASCII letters, digits, hyphens or underscores".into(),
|
||||
));
|
||||
}
|
||||
if protocols[..index].contains(name) {
|
||||
return Err(crate::Error::InvalidProtocols(format!(
|
||||
"`{name}` is listed more than once"
|
||||
)));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// An announcement of one IP plugin's capability.
|
||||
///
|
||||
/// `data` is opaque to the core. Nothing in it may be interpreted as a shell
|
||||
|
||||
@@ -492,11 +492,12 @@ impl RelayHub {
|
||||
}
|
||||
}
|
||||
|
||||
/// Revokes all links and topology involving a departed authenticated peer.
|
||||
pub fn remove_peer(&self, peer: EndpointId) {
|
||||
/// Retires data channels without removing an authenticated peer's topology.
|
||||
pub fn clear_peer_links(&self, peer: EndpointId) {
|
||||
let mut state = self.state();
|
||||
if let Some(links) = state.raw.remove(peer.as_bytes()) {
|
||||
for raw in links {
|
||||
raw.link.close();
|
||||
raw.reader.abort();
|
||||
}
|
||||
}
|
||||
@@ -511,6 +512,13 @@ impl RelayHub {
|
||||
link.close();
|
||||
}
|
||||
}
|
||||
state.publish();
|
||||
}
|
||||
|
||||
/// Revokes all links and topology involving a departed authenticated peer.
|
||||
pub fn remove_peer(&self, peer: EndpointId) {
|
||||
self.clear_peer_links(peer);
|
||||
let mut state = self.state();
|
||||
state.members.remove(peer.as_bytes());
|
||||
state.graph.remove(peer.as_bytes());
|
||||
for neighbors in state.graph.values_mut() {
|
||||
|
||||
@@ -15,6 +15,9 @@ pub type Result<T, E = Error> = std::result::Result<T, E>;
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
#[non_exhaustive]
|
||||
pub enum Error {
|
||||
/// An ordered local data-plane protocol selection is invalid.
|
||||
#[error("invalid protocol selection: {0}")]
|
||||
InvalidProtocols(String),
|
||||
/// The supplied network name does not satisfy the documented rules.
|
||||
#[error("invalid network name: {0}")]
|
||||
InvalidNetworkName(&'static str),
|
||||
|
||||
@@ -171,6 +171,13 @@ pub fn default_control_socket_path() -> PathBuf {
|
||||
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[non_exhaustive]
|
||||
pub enum Request {
|
||||
/// Replace one network's ordered local data-plane selection.
|
||||
SetProtocols {
|
||||
/// Public network identifier.
|
||||
network_id: String,
|
||||
/// Protocol names, best first; empty selects control only.
|
||||
protocols: Vec<String>,
|
||||
},
|
||||
/// Report what the agent is doing.
|
||||
Status,
|
||||
/// Set local broadcast participation in one configured network.
|
||||
@@ -255,6 +262,13 @@ pub enum Request {
|
||||
impl std::fmt::Debug for Request {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Request::SetProtocols {
|
||||
network_id,
|
||||
protocols,
|
||||
} => write!(
|
||||
f,
|
||||
"SetProtocols {{ {network_id}, protocols: {protocols:?} }}"
|
||||
),
|
||||
Request::Status => f.write_str("Status"),
|
||||
Request::SetBroadcast {
|
||||
network_id,
|
||||
@@ -286,6 +300,8 @@ impl std::fmt::Debug for Request {
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[non_exhaustive]
|
||||
pub enum Response {
|
||||
/// Accepted ordered protocol selection.
|
||||
Protocols(Vec<String>),
|
||||
/// A status report.
|
||||
Status(Box<StatusReport>),
|
||||
/// Accepted local broadcast participation.
|
||||
@@ -377,6 +393,8 @@ pub struct LeftReport {
|
||||
/// Everything the agent is doing, in one snapshot.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct StatusReport {
|
||||
/// Data-plane protocols this agent can enable at runtime.
|
||||
pub available_protocols: Vec<ProtocolReport>,
|
||||
/// This device's persistent endpoint id.
|
||||
pub endpoint_id: String,
|
||||
/// Hostname announced to peers.
|
||||
@@ -458,9 +476,20 @@ pub struct RangeConflictReport {
|
||||
pub range: String,
|
||||
}
|
||||
|
||||
/// A data-plane protocol available in the running agent.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ProtocolReport {
|
||||
/// Stable protocol name used by the CLI and network negotiation.
|
||||
pub name: String,
|
||||
/// A short description of encryption and transport.
|
||||
pub summary: String,
|
||||
}
|
||||
|
||||
/// One network.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct NetworkReport {
|
||||
/// Enabled local data-plane protocols, in preference order.
|
||||
pub protocols: Vec<String>,
|
||||
/// Whether the network starts or resumes automatically when its range is free.
|
||||
#[serde(default)]
|
||||
pub auto_start: bool,
|
||||
@@ -720,6 +749,14 @@ impl OverlayPeerReport {
|
||||
/// running and what they can report or change. That is what keeps this module
|
||||
/// free of any knowledge of them.
|
||||
pub trait ReportSource: Send + Sync + 'static {
|
||||
/// Saves and applies a local ordered protocol set for one network.
|
||||
fn set_protocols(
|
||||
&self,
|
||||
_network_id: String,
|
||||
_protocols: Vec<String>,
|
||||
) -> BoxFuture<'_, std::result::Result<Vec<String>, String>> {
|
||||
Box::pin(async { Err("this agent cannot change protocols".into()) })
|
||||
}
|
||||
/// Produces a fresh report.
|
||||
fn report(&self) -> BoxFuture<'_, StatusReport>;
|
||||
|
||||
@@ -849,7 +886,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', 19]);
|
||||
pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 20]);
|
||||
|
||||
/// Reads one request off an accepted stream, answers it, writes the response.
|
||||
///
|
||||
@@ -873,6 +910,13 @@ where
|
||||
}
|
||||
};
|
||||
let response = match request {
|
||||
Request::SetProtocols {
|
||||
network_id,
|
||||
protocols,
|
||||
} => match source.set_protocols(network_id, protocols).await {
|
||||
Ok(protocols) => Response::Protocols(protocols),
|
||||
Err(reason) => Response::Error(reason),
|
||||
},
|
||||
Request::Status => Response::Status(Box::new(source.report().await)),
|
||||
Request::SetBroadcast {
|
||||
network_id,
|
||||
@@ -1028,7 +1072,24 @@ pub async fn join_network_with_broadcast(
|
||||
}
|
||||
}
|
||||
|
||||
/// Changes one network's broadcast participation while the agent runs.
|
||||
/// Replaces the enabled data-plane protocols of one configured network.
|
||||
pub async fn set_protocols(
|
||||
socket: &Path,
|
||||
network_id: &str,
|
||||
protocols: Vec<String>,
|
||||
) -> Result<Vec<String>> {
|
||||
let request = Request::SetProtocols {
|
||||
network_id: network_id.to_owned(),
|
||||
protocols,
|
||||
};
|
||||
match exchange(socket, &request, EXCHANGE_TIMEOUT).await? {
|
||||
Response::Protocols(protocols) => Ok(protocols),
|
||||
Response::Error(reason) => Err(Error::Storage(reason)),
|
||||
other => Err(Error::Storage(format!("unexpected response: {other:?}"))),
|
||||
}
|
||||
}
|
||||
|
||||
/// Changes local broadcast participation in a network.
|
||||
pub async fn set_broadcast(
|
||||
path: impl AsRef<Path>,
|
||||
network_id: &str,
|
||||
|
||||
@@ -254,6 +254,12 @@ impl Storage {
|
||||
.await
|
||||
}
|
||||
|
||||
/// Persists the ordered protocol selection off the async executor.
|
||||
pub async fn set_protocols(&self, network_id: NetworkId, protocols: Vec<String>) -> Result<()> {
|
||||
self.with_state(move |state| state.set_protocols(network_id, &protocols))
|
||||
.await
|
||||
}
|
||||
|
||||
/// Persists whether this agent offers itself as an exit node in a network.
|
||||
pub async fn set_exit_node(&self, network_id: NetworkId, enabled: bool) -> Result<()> {
|
||||
self.with_state(move |state| state.set_exit_node(network_id, enabled))
|
||||
|
||||
@@ -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 = 7;
|
||||
pub const SCHEMA_VERSION: i64 = 8;
|
||||
|
||||
/// Key of the stored hostname setting.
|
||||
const SETTING_HOSTNAME: &str = "hostname";
|
||||
@@ -32,6 +32,8 @@ const SETTING_HOSTNAME: &str = "hostname";
|
||||
/// and zeroizes on drop.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct StoredNetwork {
|
||||
/// Ordered local protocol override; absent inherits the agent's defaults.
|
||||
pub protocols: Option<Vec<String>>,
|
||||
/// Derived public network identifier.
|
||||
pub network_id: NetworkId,
|
||||
/// Network name.
|
||||
@@ -238,6 +240,16 @@ impl StateStore {
|
||||
)
|
||||
.map_err(|err| self.corrupt(format!("cannot migrate schema to 7: {err}")))?;
|
||||
}
|
||||
if found < 8 {
|
||||
self.conn
|
||||
.execute_batch(
|
||||
"BEGIN;
|
||||
ALTER TABLE networks ADD COLUMN protocols TEXT;
|
||||
PRAGMA user_version = 8;
|
||||
COMMIT;",
|
||||
)
|
||||
.map_err(|err| self.corrupt(format!("cannot migrate schema to 8: {err}")))?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -476,6 +488,25 @@ impl StateStore {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Saves a local ordered protocol selection without changing membership.
|
||||
pub fn set_protocols(&self, network_id: NetworkId, protocols: &[String]) -> Result<()> {
|
||||
crate::dataplane::validate_protocols(
|
||||
protocols,
|
||||
crate::config::Limits::default().max_capabilities,
|
||||
)?;
|
||||
let changed = self
|
||||
.conn
|
||||
.execute(
|
||||
"UPDATE networks SET protocols = ?2 WHERE network_id = ?1",
|
||||
params![network_id.as_bytes().as_slice(), protocols.join(",")],
|
||||
)
|
||||
.map_err(|err| Error::Storage(format!("cannot update protocols: {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
|
||||
@@ -536,7 +567,7 @@ impl StateStore {
|
||||
pub fn list_networks(&self) -> Result<Vec<StoredNetwork>> {
|
||||
let mut stmt = self
|
||||
.conn
|
||||
.prepare("SELECT network_id, name, secret, auto_start, broadcast, exit_node, exit_via FROM networks ORDER BY name")
|
||||
.prepare("SELECT network_id, name, secret, auto_start, broadcast, exit_node, exit_via, protocols FROM networks ORDER BY name")
|
||||
.map_err(|err| Error::Storage(format!("cannot list networks: {err}")))?;
|
||||
let rows = stmt
|
||||
.query_map([], |row| {
|
||||
@@ -552,19 +583,37 @@ impl StateStore {
|
||||
row.get::<_, bool>(4)?,
|
||||
row.get::<_, bool>(5)?,
|
||||
row.get::<_, Option<Vec<u8>>>(6)?,
|
||||
row.get::<_, Option<String>>(7)?,
|
||||
))
|
||||
})
|
||||
.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, broadcast, exit_node, exit_via) =
|
||||
let (id, name, secret, auto_start, broadcast, exit_node, exit_via, protocols) =
|
||||
row.map_err(|err| Error::Storage(format!("cannot read network row: {err}")))?;
|
||||
let protocols = protocols
|
||||
.map(|text| {
|
||||
let max = crate::config::Limits::default().max_capabilities;
|
||||
if text.len() > max * (crate::dataplane::MAX_PROTOCOL_ID_LEN + 1) {
|
||||
return Err(self.corrupt("stored protocol selection is too long"));
|
||||
}
|
||||
let names = if text.is_empty() {
|
||||
Vec::new()
|
||||
} else {
|
||||
text.split(',').map(str::to_owned).collect()
|
||||
};
|
||||
crate::dataplane::validate_protocols(&names, max)
|
||||
.map_err(|err| self.corrupt(err))?;
|
||||
Ok(names)
|
||||
})
|
||||
.transpose()?;
|
||||
let id: [u8; 32] = id
|
||||
.as_slice()
|
||||
.try_into()
|
||||
.map_err(|_| self.corrupt("stored network id is not 32 bytes"))?;
|
||||
out.push(StoredNetwork {
|
||||
protocols,
|
||||
network_id: NetworkId::from_bytes(id),
|
||||
name: NetworkName::new(name)?,
|
||||
secret: NetworkSecret::from_bytes(secret)?,
|
||||
@@ -957,6 +1006,68 @@ mod broadcast_storage_tests {
|
||||
assert!(store.peer_hostnames(id).unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn v7_migration_keeps_signed_state_and_protocol_overrides_are_local_and_durable() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("state.sqlite");
|
||||
let store = StateStore::open(&path).unwrap();
|
||||
let identity = store.load_or_create_device_identity().unwrap();
|
||||
let name = NetworkName::new("protocol-migration").unwrap();
|
||||
let secret = NetworkSecret::from_bytes([4; 32]).unwrap();
|
||||
let id = NetworkKeys::derive(&name, &secret).network_id();
|
||||
store.upsert_network(id, &name, &secret, true).unwrap();
|
||||
store.set_broadcast(id, false).unwrap();
|
||||
let record = SignedRecord::sign(&identity.signing_key(), id, 17, RecordBody::Release);
|
||||
store.publish_own_record(&record).unwrap();
|
||||
store
|
||||
.conn
|
||||
.execute_batch("ALTER TABLE networks DROP COLUMN protocols; PRAGMA user_version=7;")
|
||||
.unwrap();
|
||||
drop(store);
|
||||
let store = StateStore::open(&path).unwrap();
|
||||
assert_eq!(
|
||||
store
|
||||
.load_or_create_device_identity()
|
||||
.unwrap()
|
||||
.endpoint_id(),
|
||||
identity.endpoint_id()
|
||||
);
|
||||
assert_eq!(store.signed_records(id).unwrap(), vec![record]);
|
||||
let saved = store.list_networks().unwrap();
|
||||
assert!(saved[0].protocols.is_none());
|
||||
assert!(!saved[0].broadcast);
|
||||
assert!(saved[0].auto_start);
|
||||
assert_eq!(saved[0].secret, secret);
|
||||
let chosen = vec!["awg".into(), "wg-quic".into()];
|
||||
store.set_protocols(id, &chosen).unwrap();
|
||||
assert!(
|
||||
store
|
||||
.set_protocols(id, &["awg".into(), "awg".into()])
|
||||
.is_err()
|
||||
);
|
||||
assert!(
|
||||
store
|
||||
.set_protocols(NetworkId::from_bytes([1; 32]), &[])
|
||||
.is_err()
|
||||
);
|
||||
drop(store);
|
||||
let store = StateStore::open(&path).unwrap();
|
||||
assert_eq!(store.list_networks().unwrap()[0].protocols, Some(chosen));
|
||||
store.upsert_network(id, &name, &secret, true).unwrap();
|
||||
store.set_protocols(id, &[]).unwrap();
|
||||
drop(store);
|
||||
let store = StateStore::open(&path).unwrap();
|
||||
assert_eq!(store.list_networks().unwrap()[0].protocols, Some(vec![]));
|
||||
store
|
||||
.conn
|
||||
.execute("UPDATE networks SET protocols = 'awg,awg'", [])
|
||||
.unwrap();
|
||||
assert!(matches!(
|
||||
store.list_networks(),
|
||||
Err(Error::StateCorrupted { .. })
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn v3_migration_preserves_identity_and_networks_and_defaults_broadcast_on() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
@@ -977,6 +1088,7 @@ mod broadcast_storage_tests {
|
||||
"ALTER TABLE networks DROP COLUMN broadcast; \
|
||||
ALTER TABLE networks DROP COLUMN exit_node; \
|
||||
ALTER TABLE networks DROP COLUMN exit_via; \
|
||||
ALTER TABLE networks DROP COLUMN protocols; \
|
||||
PRAGMA user_version=3;",
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
@@ -21,6 +21,11 @@ const POLL_INTERVAL: Duration = Duration::from_secs(2);
|
||||
|
||||
/// A request from the UI to the agent.
|
||||
pub(crate) enum Command {
|
||||
/// Replace the network's enabled protocols in preference order.
|
||||
SetProtocols {
|
||||
network_id: String,
|
||||
protocols: Vec<String>,
|
||||
},
|
||||
/// Stop serving a network, or start serving it again.
|
||||
SetActive { network_id: String, active: bool },
|
||||
/// Turn local broadcast participation on or off for a network.
|
||||
@@ -106,7 +111,13 @@ impl AgentClient {
|
||||
|
||||
/// Sends a command; a dropped worker simply means nothing happens.
|
||||
pub(crate) fn send(&self, command: Command) {
|
||||
let _ = self.tx.send(command);
|
||||
// Disable protocol editing until the refreshed snapshot arrives;
|
||||
// otherwise a second checkbox click could replace the first edit
|
||||
// using the same stale network selection.
|
||||
lock(&self.shared).busy = true;
|
||||
if self.tx.send(command).is_err() {
|
||||
lock(&self.shared).busy = false;
|
||||
}
|
||||
}
|
||||
|
||||
/// A copy of the latest snapshot.
|
||||
@@ -142,6 +153,8 @@ async fn worker(
|
||||
let Some(command) = command else { return };
|
||||
run(&resolve_socket(), command, &shared, &pending_copy, &ctx).await;
|
||||
refresh(&shared, &ctx).await;
|
||||
lock(&shared).busy = false;
|
||||
ctx.request_repaint();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -179,6 +192,22 @@ async fn run(
|
||||
ctx.request_repaint();
|
||||
|
||||
let outcome: Result<String, String> = match command {
|
||||
Command::SetProtocols {
|
||||
network_id,
|
||||
protocols,
|
||||
} => ipc::set_protocols(socket, &network_id, protocols)
|
||||
.await
|
||||
.map(|names| {
|
||||
format!(
|
||||
"protocols: {}",
|
||||
if names.is_empty() {
|
||||
"none".into()
|
||||
} else {
|
||||
names.join(", ")
|
||||
}
|
||||
)
|
||||
})
|
||||
.map_err(|err| err.to_string()),
|
||||
Command::SetActive { network_id, active } => ipc::set_active(socket, &network_id, active)
|
||||
.await
|
||||
.map(|report| {
|
||||
@@ -260,7 +289,6 @@ async fn run(
|
||||
|
||||
{
|
||||
let mut guard = lock(shared);
|
||||
guard.busy = false;
|
||||
guard.last_action = if matches!(&outcome, Ok(message) if message.is_empty()) {
|
||||
None
|
||||
} else {
|
||||
|
||||
@@ -123,7 +123,7 @@ pub(crate) fn draw(
|
||||
ui.weak("No networks yet — join one below.");
|
||||
}
|
||||
for network in &report.networks {
|
||||
draw_network(ui, agent, state, traffic, network, &report.endpoint_id);
|
||||
draw_network(ui, agent, state, traffic, network, report);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -202,7 +202,7 @@ fn draw_network(
|
||||
state: &mut UiState,
|
||||
traffic: &Traffic,
|
||||
network: &NetworkReport,
|
||||
own_id: &str,
|
||||
report: &StatusReport,
|
||||
) {
|
||||
ui.add_space(6.0);
|
||||
egui::Frame::group(ui.style()).show(ui, |ui| {
|
||||
@@ -244,12 +244,14 @@ fn draw_network(
|
||||
ui.weak("peers");
|
||||
ui.label(format!(
|
||||
"{} online · {} known",
|
||||
devices::online_count(network, own_id),
|
||||
devices::known_count(network, own_id)
|
||||
devices::online_count(network, &report.endpoint_id),
|
||||
devices::known_count(network, &report.endpoint_id)
|
||||
));
|
||||
ui.end_row();
|
||||
});
|
||||
|
||||
draw_protocols(ui, agent, network, &report.available_protocols);
|
||||
|
||||
format::sparkline(ui, series, state.unit, 34.0);
|
||||
|
||||
if !network.range_conflicts.is_empty() {
|
||||
@@ -345,6 +347,83 @@ fn draw_network(
|
||||
});
|
||||
}
|
||||
|
||||
fn draw_protocols(
|
||||
ui: &mut egui::Ui,
|
||||
agent: &AgentClient,
|
||||
network: &NetworkReport,
|
||||
available: &[tsunagi::ipc::ProtocolReport],
|
||||
) {
|
||||
let summary = if network.protocols.is_empty() {
|
||||
"none".into()
|
||||
} else {
|
||||
network.protocols.join(", ")
|
||||
};
|
||||
egui::CollapsingHeader::new(format!("protocols {summary}"))
|
||||
.id_salt(("protocols", &network.network_id))
|
||||
.show(ui, |ui| {
|
||||
ui.weak("Preference order; changes apply immediately and are saved.");
|
||||
let mut selected = network.protocols.clone();
|
||||
let mut changed = false;
|
||||
ui.add_enabled_ui(!agent.snapshot().busy, |ui| {
|
||||
// Enabled items follow the user's preference; unchecked items
|
||||
// follow the agent's catalog. The worker refreshes after edits.
|
||||
let ordered: Vec<_> = network
|
||||
.protocols
|
||||
.iter()
|
||||
.filter_map(|name| available.iter().find(|p| &p.name == name))
|
||||
.chain(
|
||||
available
|
||||
.iter()
|
||||
.filter(|p| !network.protocols.contains(&p.name)),
|
||||
)
|
||||
.collect();
|
||||
for protocol in ordered {
|
||||
ui.horizontal(|ui| {
|
||||
let mut on = selected.contains(&protocol.name);
|
||||
if ui.checkbox(&mut on, &protocol.name).changed() {
|
||||
if on {
|
||||
selected.push(protocol.name.clone());
|
||||
} else {
|
||||
selected.retain(|p| p != &protocol.name);
|
||||
}
|
||||
changed = true;
|
||||
}
|
||||
if let Some(index) = selected.iter().position(|p| p == &protocol.name) {
|
||||
if ui
|
||||
.add_enabled(index > 0, egui::Button::new("↑"))
|
||||
.on_hover_text("prefer earlier")
|
||||
.clicked()
|
||||
{
|
||||
selected.swap(index, index - 1);
|
||||
changed = true;
|
||||
}
|
||||
if ui
|
||||
.add_enabled(index + 1 < selected.len(), egui::Button::new("↓"))
|
||||
.on_hover_text("prefer later")
|
||||
.clicked()
|
||||
{
|
||||
selected.swap(index, index + 1);
|
||||
changed = true;
|
||||
}
|
||||
}
|
||||
});
|
||||
ui.indent(("description", &network.network_id, &protocol.name), |ui| {
|
||||
ui.weak(&protocol.summary);
|
||||
});
|
||||
}
|
||||
});
|
||||
if selected.is_empty() {
|
||||
ui.weak("Control only; no IP traffic in this network.");
|
||||
}
|
||||
if changed {
|
||||
agent.send(Command::SetProtocols {
|
||||
network_id: network.network_id.clone(),
|
||||
protocols: selected,
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// A prominent line about an exit node. One that carries a command copies it
|
||||
/// when clicked.
|
||||
fn draw_exit_warning(ui: &mut egui::Ui, warning: &exit::Warning) {
|
||||
|
||||
+192
-61
@@ -69,8 +69,8 @@ enum Command {
|
||||
Up(Box<UpArgs>),
|
||||
/// Reports this device, what the agent is doing, and what this host can do.
|
||||
Status(StatusArgs),
|
||||
/// Shows the protocols this build can carry packets with.
|
||||
Protocols,
|
||||
/// Lists protocols, or shows and changes one network's selection at runtime.
|
||||
Protocols(ProtocolsArgs),
|
||||
/// Makes a network, or joins one — the command that does the work.
|
||||
///
|
||||
/// The same as `tsng network join`, at the top level because it is
|
||||
@@ -107,6 +107,23 @@ struct DnsArgs {
|
||||
action: Option<DnsAction>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Args)]
|
||||
#[command(
|
||||
after_help = "Examples:\n tsng protocols\n tsng protocols home\n tsng protocols home awg,wg-quic\n tsng protocols home none\n\nThe selection is local to this device, saved per network and applied immediately.\nBoth endpoints need a common protocol at the same wire version."
|
||||
)]
|
||||
struct ProtocolsArgs {
|
||||
#[command(flatten)]
|
||||
paths: PathArgs,
|
||||
/// Control socket to talk to; uses the running service by default.
|
||||
#[arg(long, env = "TSUNAGI_CONTROL_SOCKET")]
|
||||
control_socket: Option<PathBuf>,
|
||||
/// Network's exact name, id or unique id prefix.
|
||||
network: Option<String>,
|
||||
/// Replace the selection, best first: awg,wg-quic; none disables data traffic.
|
||||
#[arg(value_name = "LIST", requires = "network")]
|
||||
protocols: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Subcommand)]
|
||||
enum DnsAction {
|
||||
/// Starts serving, now and after every restart.
|
||||
@@ -465,11 +482,13 @@ struct UpArgs {
|
||||
#[arg(long, help_heading = "System")]
|
||||
no_tun: bool,
|
||||
|
||||
/// Protocols to carry packets with, best first.
|
||||
/// Initial protocol selection for networks without a saved choice, best first.
|
||||
///
|
||||
/// A pair of peers uses one they both have at the same wire version. A
|
||||
/// peer with none in common keeps its control plane and gets no data
|
||||
/// plane. `none` runs the control plane alone.
|
||||
/// plane. `none` runs the control plane alone. `tsng protocols NETWORK LIST`
|
||||
/// changes and saves each network's choice while this agent runs; any
|
||||
/// built-in protocol can be enabled, even if absent from this initial list.
|
||||
#[arg(
|
||||
long = "protocol",
|
||||
value_name = "LIST",
|
||||
@@ -897,7 +916,7 @@ async fn run(command: Command) -> Result<(), Box<dyn std::error::Error>> {
|
||||
Command::Id(args) => id(args).await,
|
||||
Command::Up(args) => up(*args).await,
|
||||
Command::Status(args) => status(args).await,
|
||||
Command::Protocols => show_protocols(),
|
||||
Command::Protocols(args) => protocols_command(args).await,
|
||||
Command::Join(args) => join_command(args).await,
|
||||
Command::Network(args) => network_command(args).await,
|
||||
Command::Dns(args) => dns_command(args).await,
|
||||
@@ -1454,6 +1473,7 @@ const PROTOCOLS: &[ProtocolSpec] = &[
|
||||
];
|
||||
|
||||
/// One `-o` setting, and the protocol it was aimed at.
|
||||
#[derive(Clone)]
|
||||
struct Setting {
|
||||
/// `Some` when written as `protocol:key=value`.
|
||||
protocol: Option<String>,
|
||||
@@ -1704,6 +1724,23 @@ impl tsunagi::ipc::ReportSource for AgentControl {
|
||||
})
|
||||
}
|
||||
|
||||
fn set_protocols(
|
||||
&self,
|
||||
network_id: String,
|
||||
protocols: Vec<String>,
|
||||
) -> tsunagi::BoxFuture<'_, Result<Vec<String>, String>> {
|
||||
Box::pin(async move {
|
||||
let id = network_id
|
||||
.parse()
|
||||
.map_err(|err| format!("invalid network id: {err}"))?;
|
||||
self.agent
|
||||
.set_protocols(id, protocols.clone())
|
||||
.await
|
||||
.map_err(|err| err.to_string())?;
|
||||
Ok(protocols)
|
||||
})
|
||||
}
|
||||
|
||||
fn set_exit_offer(
|
||||
&self,
|
||||
network_id: String,
|
||||
@@ -1866,45 +1903,138 @@ fn resolve_network<'a>(
|
||||
networks: &'a [tsunagi::storage::StoredNetwork],
|
||||
wanted: &str,
|
||||
) -> Result<&'a tsunagi::storage::StoredNetwork, String> {
|
||||
let entries: Vec<_> = networks
|
||||
.iter()
|
||||
.map(|n| (n.network_id.to_string(), n.name.as_str().to_owned()))
|
||||
.collect();
|
||||
resolve_network_index(&entries, wanted).map(|i| &networks[i])
|
||||
}
|
||||
|
||||
/// Shared selector rules for stored networks and secret-free live reports.
|
||||
fn resolve_network_index(networks: &[(String, String)], wanted: &str) -> Result<usize, String> {
|
||||
let wanted = wanted.trim_matches(|c: char| c.is_ascii_whitespace());
|
||||
if wanted.is_empty() {
|
||||
return Err("select a network by name or id; `tsng network` lists them".to_string());
|
||||
return Err("select a network by name or id; `tsng network` lists them".into());
|
||||
}
|
||||
if let Some(network) = networks
|
||||
.iter()
|
||||
.find(|network| network.network_id.to_string() == wanted)
|
||||
{
|
||||
return Ok(network);
|
||||
if let Some(index) = networks.iter().position(|(id, _)| id == wanted) {
|
||||
return Ok(index);
|
||||
}
|
||||
// Only id prefixes lose a copied report ellipsis; names are used verbatim.
|
||||
let prefix = wanted.trim_end_matches('…');
|
||||
let matched: Vec<&tsunagi::storage::StoredNetwork> = networks
|
||||
let matches: Vec<_> = networks
|
||||
.iter()
|
||||
.filter(|network| {
|
||||
network.name.as_str() == wanted
|
||||
|| (!prefix.is_empty() && network.network_id.to_string().starts_with(prefix))
|
||||
})
|
||||
.enumerate()
|
||||
.filter(|(_, (id, name))| name == wanted || (!prefix.is_empty() && id.starts_with(prefix)))
|
||||
.collect();
|
||||
match matched.as_slice() {
|
||||
[one] => Ok(one),
|
||||
match matches.as_slice() {
|
||||
[(index, _)] => Ok(*index),
|
||||
[] => Err(format!(
|
||||
"no configured network matches name or id `{wanted}`; \
|
||||
`tsng network` lists them"
|
||||
"no configured network matches name or id `{wanted}`; `tsng network` lists them"
|
||||
)),
|
||||
several => {
|
||||
let candidates = several
|
||||
several => Err(format!(
|
||||
"`{wanted}` matches {} networks; use the full id or a unique id prefix:\n{}",
|
||||
several.len(),
|
||||
several
|
||||
.iter()
|
||||
.map(|network| format!(" `{}` {}", network.name, network.network_id))
|
||||
.map(|(_, (id, name))| format!(" `{name}` {id}"))
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n");
|
||||
Err(format!(
|
||||
"`{wanted}` matches {} networks; use the full id or a unique id prefix:\n{candidates}",
|
||||
several.len()
|
||||
))
|
||||
}
|
||||
.join("\n")
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_protocol_list(text: &str) -> Result<Vec<String>, String> {
|
||||
if text.eq_ignore_ascii_case("none") {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
let mut names = Vec::new();
|
||||
for name in text.split(',') {
|
||||
let name = name.trim();
|
||||
let name = if name.eq_ignore_ascii_case("wireguard") {
|
||||
WG_PROTOCOL
|
||||
} else {
|
||||
name
|
||||
};
|
||||
if !PROTOCOLS.iter().any(|spec| spec.name == name) {
|
||||
return Err(format!(
|
||||
"unknown protocol `{name}`; use wg,awg,wg-quic,tcp-tls or `none` alone. Run `tsng protocols` for details"
|
||||
));
|
||||
}
|
||||
if names.iter().any(|n| n == name) {
|
||||
return Err(format!("protocol `{name}` is listed more than once"));
|
||||
}
|
||||
names.push(name.to_owned());
|
||||
}
|
||||
Ok(names)
|
||||
}
|
||||
|
||||
async fn protocols_command(args: ProtocolsArgs) -> Result<(), Box<dyn std::error::Error>> {
|
||||
let Some(selector) = args.network else {
|
||||
return show_protocols();
|
||||
};
|
||||
let requested = args
|
||||
.protocols
|
||||
.as_deref()
|
||||
.map(parse_protocol_list)
|
||||
.transpose()?;
|
||||
let custom = args.paths.state_dir.is_some();
|
||||
let paths = args.paths.resolve()?;
|
||||
let socket = control_socket(&paths, args.control_socket.as_ref(), custom);
|
||||
if tsunagi::ipc::is_serving(&socket).await {
|
||||
let report = tsunagi::ipc::request_status(&socket).await?;
|
||||
let entries: Vec<_> = report
|
||||
.networks
|
||||
.iter()
|
||||
.map(|n| (n.network_id.clone(), n.name.clone()))
|
||||
.collect();
|
||||
let network = &report.networks[resolve_network_index(&entries, &selector)?];
|
||||
let protocols = match requested {
|
||||
Some(protocols) => {
|
||||
tsunagi::ipc::set_protocols(&socket, &network.network_id, protocols).await?
|
||||
}
|
||||
None => network.protocols.clone(),
|
||||
};
|
||||
println!(
|
||||
"protocols {} for `{}` ({}) · preference order",
|
||||
if protocols.is_empty() {
|
||||
"none".into()
|
||||
} else {
|
||||
protocols.join(",")
|
||||
},
|
||||
network.name,
|
||||
network.network_id
|
||||
);
|
||||
} else {
|
||||
// Take the directory ownership lock even for an offline edit. Never
|
||||
// write behind a live agent whose socket cannot be reached.
|
||||
let storage = tsunagi::storage::Storage::open(&paths)?;
|
||||
let configured = storage.list_networks().await?;
|
||||
let network = resolve_network(&configured, &selector)?;
|
||||
let protocols = match requested {
|
||||
Some(protocols) => {
|
||||
storage
|
||||
.set_protocols(network.network_id, protocols.clone())
|
||||
.await?;
|
||||
Some(protocols)
|
||||
}
|
||||
None => network.protocols.clone(),
|
||||
};
|
||||
let shown = protocols
|
||||
.map(|names| {
|
||||
if names.is_empty() {
|
||||
"none".into()
|
||||
} else {
|
||||
names.join(",")
|
||||
}
|
||||
})
|
||||
.unwrap_or_else(|| "defaults from tsng up --protocol".into());
|
||||
println!(
|
||||
"protocols {shown} for `{}` ({}) · agent stopped; applies on next start",
|
||||
network.name, network.network_id
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// `tsng network`: what this device belongs to, and leaving it.
|
||||
async fn network_command(args: NetworkArgs) -> Result<(), Box<dyn std::error::Error>> {
|
||||
let paths = args.paths.resolve()?;
|
||||
@@ -4445,43 +4575,24 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
|
||||
// What the user asked for is checked first, before anything that could
|
||||
// fail for a reason of its own: a misspelled protocol or setting is
|
||||
// their mistake to see, not something to bury under a privilege error.
|
||||
let wanted: Vec<&ProtocolSpec> = {
|
||||
let mut wanted = Vec::new();
|
||||
for name in args
|
||||
.protocols
|
||||
.iter()
|
||||
.filter(|name| !name.eq_ignore_ascii_case("none"))
|
||||
{
|
||||
// Accept "wireguard" as a convenience alias for the canonical "wg".
|
||||
let canonical = if name.eq_ignore_ascii_case("wireguard") {
|
||||
WG_PROTOCOL
|
||||
} else {
|
||||
name.as_str()
|
||||
};
|
||||
let Some(spec) = PROTOCOLS.iter().find(|spec| spec.name == canonical) else {
|
||||
let known: Vec<&str> = PROTOCOLS.iter().map(|spec| spec.name).collect();
|
||||
return Err(format!(
|
||||
"this build has no protocol called `{name}`; it has {}. \
|
||||
Run `tsng protocols` to see what each one takes.",
|
||||
known.join(", ")
|
||||
)
|
||||
.into());
|
||||
};
|
||||
wanted.push(spec);
|
||||
}
|
||||
wanted
|
||||
};
|
||||
let initial = parse_protocol_list(&args.protocols.join(","))?;
|
||||
let wanted: Vec<&ProtocolSpec> = initial
|
||||
.iter()
|
||||
.filter_map(|name| PROTOCOLS.iter().find(|spec| spec.name == name))
|
||||
.collect();
|
||||
config.default_protocols = Some(initial);
|
||||
|
||||
let settings = parse_settings(&args.protocol_options)?;
|
||||
// A setting nobody takes is a mistake, not a preference: one that was
|
||||
// silently dropped looks exactly like one that did not work.
|
||||
for setting in &settings {
|
||||
if !wanted
|
||||
if !PROTOCOLS
|
||||
.iter()
|
||||
.filter(|spec| setting.protocol.is_some() || wanted.iter().any(|p| p.name == spec.name))
|
||||
.any(|spec| !settings_for(spec, std::slice::from_ref(setting)).is_empty())
|
||||
{
|
||||
return Err(format!(
|
||||
"no selected protocol takes `{}`; run `tsng protocols` to see what they do",
|
||||
"no configured protocol takes `{}`; qualify disabled protocol settings as protocol:key=value. Run `tsng protocols` for details",
|
||||
setting.key
|
||||
)
|
||||
.into());
|
||||
@@ -4497,7 +4608,7 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
|
||||
let mut tcp_tls_plugin = None;
|
||||
let mut pure_wg_plugin: Option<Arc<PureWgPlugin>> = None;
|
||||
let mut pure_wg_transport: Option<Arc<PureWgTransport>> = None;
|
||||
if !wanted.is_empty() {
|
||||
if !PROTOCOLS.is_empty() {
|
||||
let tun_factory: Arc<dyn TunFactory> = if args.no_tun {
|
||||
Arc::new(MemoryTunFactory::new())
|
||||
} else {
|
||||
@@ -4507,8 +4618,17 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
|
||||
config = config.with_interface(tun_factory, args.interface.clone(), mtu);
|
||||
}
|
||||
|
||||
for spec in &wanted {
|
||||
let options = settings_for(spec, &settings);
|
||||
for spec in PROTOCOLS {
|
||||
// Unqualified options keep targeting the initial selection; explicitly
|
||||
// scoped options also configure protocols that will be enabled later.
|
||||
let scoped: Vec<_> = settings
|
||||
.iter()
|
||||
.filter(|setting| {
|
||||
setting.protocol.is_some() || wanted.iter().any(|p| p.name == spec.name)
|
||||
})
|
||||
.cloned()
|
||||
.collect();
|
||||
let options = settings_for(spec, &scoped);
|
||||
match spec.name {
|
||||
AWG_PROTOCOL => {
|
||||
let mut wg = WireguardConfig::new(paths.state_dir.join("awg"));
|
||||
@@ -5112,6 +5232,7 @@ async fn build_report(
|
||||
};
|
||||
|
||||
NetworkReport {
|
||||
protocols: network.protocols.clone(),
|
||||
auto_start: network.auto_start,
|
||||
exit: exit_report(network),
|
||||
broadcast: network.broadcast,
|
||||
@@ -5182,6 +5303,13 @@ async fn build_report(
|
||||
.collect();
|
||||
|
||||
StatusReport {
|
||||
available_protocols: PROTOCOLS
|
||||
.iter()
|
||||
.map(|p| tsunagi::ipc::ProtocolReport {
|
||||
name: p.name.into(),
|
||||
summary: p.summary.into(),
|
||||
})
|
||||
.collect(),
|
||||
endpoint_id: status.endpoint_id.to_string(),
|
||||
hostname: status.hostname.clone(),
|
||||
bound_sockets: status
|
||||
@@ -5536,6 +5664,7 @@ mod status_tests {
|
||||
/// The situation that prompted this: one peer left and came back.
|
||||
fn network_after_a_peer_returned() -> NetworkReport {
|
||||
NetworkReport {
|
||||
protocols: vec!["wg-quic".into()],
|
||||
auto_start: true,
|
||||
exit: Default::default(),
|
||||
hostname_conflict: None,
|
||||
@@ -6219,6 +6348,7 @@ mod network_tests {
|
||||
NetworkSecret::from_bytes(&[secret.as_bytes(), &[0u8; 32]].concat()[..32]).unwrap();
|
||||
let keys = NetworkKeys::derive(&name, &secret);
|
||||
StoredNetwork {
|
||||
protocols: None,
|
||||
exit_node: false,
|
||||
exit_via: None,
|
||||
broadcast: true,
|
||||
@@ -6442,6 +6572,7 @@ mod network_context_tests {
|
||||
let secret = NetworkSecret::from_bytes([seed; 32]).unwrap();
|
||||
let keys = NetworkKeys::derive(&name, &secret);
|
||||
StoredNetwork {
|
||||
protocols: None,
|
||||
exit_node: false,
|
||||
exit_via: None,
|
||||
broadcast: true,
|
||||
|
||||
@@ -54,6 +54,10 @@ fn start_bare(port: u16) -> Running {
|
||||
}
|
||||
|
||||
fn start_bare_range(port: u16, range: Option<&str>) -> Running {
|
||||
start_bare_config(port, range, None)
|
||||
}
|
||||
|
||||
fn start_bare_config(port: u16, range: Option<&str>, protocols: Option<&str>) -> Running {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let mut command = std::process::Command::new(env!("CARGO_BIN_EXE_tsng"));
|
||||
command
|
||||
@@ -71,6 +75,9 @@ fn start_bare_range(port: u16, range: Option<&str>) -> Running {
|
||||
if let Some(range) = range {
|
||||
command.args(["--ipv4-range", range]);
|
||||
}
|
||||
if let Some(protocols) = protocols {
|
||||
command.args(["--protocol", protocols]);
|
||||
}
|
||||
let child = command.spawn().expect("the agent binary starts");
|
||||
let mut agent = Running { child, dir };
|
||||
let runtime = tokio::runtime::Builder::new_current_thread()
|
||||
@@ -112,6 +119,147 @@ fn start(network: &str, port: u16) -> Running {
|
||||
agent
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn protocols_absent_from_startup_can_be_enabled_and_changed_offline() {
|
||||
let mut agent = start_bare_config(0, None, Some("none"));
|
||||
assert!(
|
||||
agent
|
||||
.run(&[
|
||||
"join",
|
||||
"--network",
|
||||
"from-control-only",
|
||||
"--secret",
|
||||
"a-control-only-test-secret"
|
||||
])
|
||||
.status
|
||||
.success()
|
||||
);
|
||||
let initial = agent.run(&["protocols", "from-control-only"]);
|
||||
assert!(initial.status.success());
|
||||
assert!(String::from_utf8_lossy(&initial.stdout).contains("protocols none"));
|
||||
let enabled = agent.run(&["protocols", "from-control-only", "awg"]);
|
||||
assert!(
|
||||
enabled.status.success(),
|
||||
"{}",
|
||||
String::from_utf8_lossy(&enabled.stderr)
|
||||
);
|
||||
assert_eq!(
|
||||
read_networks(&agent.dir)[0].protocols,
|
||||
Some(vec!["awg".into()])
|
||||
);
|
||||
agent.child.kill().unwrap();
|
||||
agent.child.wait().unwrap();
|
||||
let saved = agent.run(&["protocols", "from-control-only", "wg-quic,awg"]);
|
||||
assert!(
|
||||
saved.status.success(),
|
||||
"{}",
|
||||
String::from_utf8_lossy(&saved.stderr)
|
||||
);
|
||||
assert!(String::from_utf8_lossy(&saved.stdout).contains("agent stopped"));
|
||||
assert_eq!(
|
||||
read_networks(&agent.dir)[0].protocols,
|
||||
Some(vec!["wg-quic".into(), "awg".into()])
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn protocols_are_changed_live_per_network_and_saved_without_changing_the_service() {
|
||||
let agent = start("protocol-choice", 0);
|
||||
assert!(
|
||||
agent
|
||||
.run(&[
|
||||
"join",
|
||||
"--network",
|
||||
"other-choice",
|
||||
"--secret",
|
||||
"other-secret-for-cli-test"
|
||||
])
|
||||
.status
|
||||
.success()
|
||||
);
|
||||
let before = read_networks(&agent.dir);
|
||||
let other_id = before
|
||||
.iter()
|
||||
.find(|n| n.name.as_str() == "other-choice")
|
||||
.unwrap()
|
||||
.network_id;
|
||||
let changed = agent.run(&["protocols", "protocol-choice", "awg,wg-quic"]);
|
||||
assert!(
|
||||
changed.status.success(),
|
||||
"{}",
|
||||
String::from_utf8_lossy(&changed.stderr)
|
||||
);
|
||||
let shown = agent.run(&["protocols", "protocol-choice"]);
|
||||
assert!(shown.status.success());
|
||||
assert!(String::from_utf8_lossy(&shown.stdout).contains("awg,wg-quic"));
|
||||
let saved = read_networks(&agent.dir);
|
||||
assert_eq!(
|
||||
saved
|
||||
.iter()
|
||||
.find(|n| n.name.as_str() == "protocol-choice")
|
||||
.unwrap()
|
||||
.protocols,
|
||||
Some(vec!["awg".into(), "wg-quic".into()])
|
||||
);
|
||||
assert_eq!(
|
||||
saved
|
||||
.iter()
|
||||
.find(|n| n.network_id == other_id)
|
||||
.unwrap()
|
||||
.protocols,
|
||||
before
|
||||
.iter()
|
||||
.find(|n| n.network_id == other_id)
|
||||
.unwrap()
|
||||
.protocols
|
||||
);
|
||||
for invalid in ["unknown", "awg,awg", "none,awg"] {
|
||||
assert!(
|
||||
!agent
|
||||
.run(&["protocols", "protocol-choice", invalid])
|
||||
.status
|
||||
.success()
|
||||
);
|
||||
}
|
||||
assert!(!agent.run(&["protocols", "missing", "awg"]).status.success());
|
||||
let disabled = agent.run(&["protocols", "protocol-choice", "none"]);
|
||||
assert!(
|
||||
disabled.status.success(),
|
||||
"{}",
|
||||
String::from_utf8_lossy(&disabled.stderr)
|
||||
);
|
||||
assert_eq!(
|
||||
read_networks(&agent.dir)
|
||||
.iter()
|
||||
.find(|n| n.name.as_str() == "protocol-choice")
|
||||
.unwrap()
|
||||
.protocols,
|
||||
Some(Vec::new())
|
||||
);
|
||||
assert!(
|
||||
agent
|
||||
.run(&["network", "stop", "protocol-choice"])
|
||||
.status
|
||||
.success()
|
||||
);
|
||||
assert!(
|
||||
agent
|
||||
.run(&["protocols", "protocol-choice", "wg-quic"])
|
||||
.status
|
||||
.success()
|
||||
);
|
||||
assert!(
|
||||
agent
|
||||
.run(&["network", "start", "protocol-choice"])
|
||||
.status
|
||||
.success()
|
||||
);
|
||||
assert!(
|
||||
String::from_utf8_lossy(&agent.run(&["protocols", "protocol-choice"]).stdout)
|
||||
.contains("wg-quic")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn joining_with_no_secret_makes_one_and_prints_what_to_send() {
|
||||
// The hurry case: a network with somebody, for as long as it is needed,
|
||||
|
||||
@@ -8,6 +8,21 @@ tsng up --protocol awg
|
||||
tsng up --protocol wg,awg,wg-quic,tcp-tls
|
||||
```
|
||||
|
||||
The startup list is the initial per-network preference. A saved network choice
|
||||
overrides it and can enable any built-in protocol at runtime:
|
||||
|
||||
```sh
|
||||
tsng protocols home awg,wg-quic
|
||||
tsng protocols home awg
|
||||
```
|
||||
|
||||
Changes are local to this device, persist across restarts, and retire existing
|
||||
data links and end-to-end channels for that network before renegotiation.
|
||||
Authenticated control sessions, membership, addresses and other networks stay
|
||||
live. Removing every protocol (`none` in the CLI, all checkboxes off in the tray)
|
||||
leaves the network control-only. This is different from automatic carrier
|
||||
fallback, which preserves the selected end-to-end channel.
|
||||
|
||||
`awg` is Tsunagi AWG wire version 1. Its packet masking engine is adapted from
|
||||
[RunaRoo/AmneziaWG-RS](https://github.com/RunaRoo/AmneziaWG-RS), with the pinned
|
||||
source and license recorded in
|
||||
|
||||
@@ -184,3 +184,17 @@ without substituting iroh, membership, encryption or SQLite. It checks runtime
|
||||
loss detection, fallback and fresh profiles, in addition to restart and input
|
||||
validation. Tests are offline; public STUN/DPI/NAT effectiveness needs separate
|
||||
real-network validation.
|
||||
|
||||
Runtime protocol selection tests switch real AWG and iroh/QUIC carriers and
|
||||
end-to-end WireGuard tunnels from either endpoint-id role, carry full-size
|
||||
packets, keep another active network's tunnels unchanged, disable and re-enable
|
||||
the data plane without losing control sessions, and preserve selection after
|
||||
restart. CLI tests use a real agent process and control socket to exercise
|
||||
selection by network name, stopped networks, invalid lists, and persisted
|
||||
control-only configuration, offline updates, and enabling a protocol absent
|
||||
from the startup list. A real AWG test starts without a TUN, creates the single
|
||||
memory interface on enablement, and races rapid changes against pending opens.
|
||||
SQLite tests migrate schema 7 to 8 without changing
|
||||
identity, secrets or signed records, preserve order and explicit empty choices,
|
||||
and reject malformed stored selections. These checks use memory TUNs and do
|
||||
not establish GUI rendering or behavior on an actual OS interface.
|
||||
Reference in new issue
Block a user