feat(credentials): manage declarative credentials through TOML (#2515)

* feat(credentials): manage declarative credentials through TOML

Make managed credentials part of the canonical TOML configuration and
load them before peers can authenticate.

Reuse ConfigRpc hot patches to durably replace the configured credential
set without restarting the instance. Serialize credential mutations so
base, managed, and ephemeral keys cannot race into conflicts.

Remove the managed overlay file format, digest protocol, capability
negotiation, force reconciliation, and database CAS machinery. Redact
credential secrets from debug output and management events. Write
credential-bearing files atomically with private permissions.

* fix(core): release JoinSet reapers with their owners

Pass weak task-set references into background reapers so they cannot
retain the JoinSet they are meant to collect. This lets stale smoltcp
bridge tasks terminate when an IPv4 generation is replaced.

Add ownership and TCP generation-replacement regressions covering the
production port-forward failure.
This commit is contained in:
KKRainbow
2026-08-22 16:30:51 +08:00
committed by GitHub
parent 8794e12a26
commit 3fe427bc99
44 changed files with 1600 additions and 259 deletions
+7
View File
@@ -54,6 +54,7 @@ jobs:
- uses: actions-rust-lang/setup-rust-toolchain@v1 - uses: actions-rust-lang/setup-rust-toolchain@v1
with: with:
components: rustfmt,clippy components: rustfmt,clippy
target: wasm32-wasip1
rustflags: '' rustflags: ''
- uses: taiki-e/install-action@cargo-hack - uses: taiki-e/install-action@cargo-hack
@@ -70,6 +71,12 @@ jobs:
if: ${{ !cancelled() }} if: ${{ !cancelled() }}
run: cargo hack check --package easytier --each-feature --exclude-features macos-ne --verbose run: cargo hack check --package easytier --each-feature --exclude-features macos-ne --verbose
- name: Check WASI
if: ${{ !cancelled() }}
run: >-
cargo check --package easytier-core --lib --target wasm32-wasip1
--features management-rpc,proxy-smoltcp-stack,ring-crypto,wasi-crypto-offload
- name: Check Cargo.lock is up to date - name: Check Cargo.lock is up to date
if: ${{ !cancelled() }} if: ${{ !cancelled() }}
run: | run: |
+48
View File
@@ -118,6 +118,19 @@ pub fn network_config_from_toml(config: &TomlConfig) -> NetworkConfig {
result.credential_file = config result.credential_file = config
.get_credential_file() .get_credential_file()
.map(|path| path.to_string_lossy().into_owned()); .map(|path| path.to_string_lossy().into_owned());
result.managed_credentials = config
.get_managed_credentials()
.into_iter()
.map(|credential| manage::ManagedCredentialConfig {
credential_id: credential.credential_id,
credential_secret: credential.credential_secret,
groups: credential.groups,
allow_relay: credential.allow_relay,
allowed_proxy_cidrs: credential.allowed_proxy_cidrs,
expiry_unix: credential.expiry_unix,
reusable: Some(credential.reusable),
})
.collect();
let flags = config.get_flags(); let flags = config.get_flags();
let default_flags = default_config.get_flags(); let default_flags = default_config.get_flags();
@@ -172,3 +185,38 @@ pub fn network_config_from_toml(config: &TomlConfig) -> NetworkConfig {
result result
} }
#[cfg(test)]
mod tests {
use super::*;
use crate::config::toml::ManagedCredentialConfig;
#[test]
fn includes_managed_credentials() {
let config = TomlConfig::default();
config.set_managed_credentials(vec![ManagedCredentialConfig {
credential_id: "managed-a".to_owned(),
credential_secret: "credential-secret".to_owned(),
groups: vec!["ops".to_owned()],
allow_relay: true,
allowed_proxy_cidrs: vec!["10.0.0.0/24".to_owned()],
expiry_unix: 2_000_000_000,
reusable: false,
}]);
let projected = network_config_from_toml(&config);
assert_eq!(
projected.managed_credentials,
vec![manage::ManagedCredentialConfig {
credential_id: "managed-a".to_owned(),
credential_secret: "credential-secret".to_owned(),
groups: vec!["ops".to_owned()],
allow_relay: true,
allowed_proxy_cidrs: vec!["10.0.0.0/24".to_owned()],
expiry_unix: 2_000_000_000,
reusable: Some(false),
}]
);
}
}
+51 -2
View File
@@ -8,8 +8,8 @@ use easytier_proto::api::manage;
use crate::config::{ use crate::config::{
MappedListenerPolicy, normalize_secure_mode_config, MappedListenerPolicy, normalize_secure_mode_config,
toml::{ toml::{
ConfigLoader, NetworkIdentity, PeerConfig, PortForwardConfig, TomlConfigLoader, ConfigLoader, ManagedCredentialConfig, NetworkIdentity, PeerConfig, PortForwardConfig,
VpnPortalClientConfig, VpnPortalConfig, gen_default_flags, TomlConfigLoader, VpnPortalClientConfig, VpnPortalConfig, gen_default_flags,
}, },
}; };
@@ -298,6 +298,21 @@ impl NetworkConfigExt for NetworkConfig {
cfg.set_credential_file(Some(credential_file.into())); cfg.set_credential_file(Some(credential_file.into()));
} }
cfg.set_managed_credentials(
self.managed_credentials
.iter()
.map(|credential| ManagedCredentialConfig {
credential_id: credential.credential_id.clone(),
credential_secret: credential.credential_secret.clone(),
groups: credential.groups.clone(),
allow_relay: credential.allow_relay,
allowed_proxy_cidrs: credential.allowed_proxy_cidrs.clone(),
expiry_unix: credential.expiry_unix,
reusable: credential.reusable.unwrap_or(true),
})
.collect(),
);
if let Some(credential_secret) = credential_secret { if let Some(credential_secret) = credential_secret {
cfg.set_secure_mode(Some(normalize_secure_mode_config( cfg.set_secure_mode(Some(normalize_secure_mode_config(
easytier_proto::common::SecureModeConfig { easytier_proto::common::SecureModeConfig {
@@ -606,6 +621,19 @@ impl NetworkConfigExt for NetworkConfig {
result.credential_file = config result.credential_file = config
.get_credential_file() .get_credential_file()
.map(|path| path.to_string_lossy().into_owned()); .map(|path| path.to_string_lossy().into_owned());
result.managed_credentials = config
.get_managed_credentials()
.into_iter()
.map(|credential| manage::ManagedCredentialConfig {
credential_id: credential.credential_id,
credential_secret: credential.credential_secret,
groups: credential.groups,
allow_relay: credential.allow_relay,
allowed_proxy_cidrs: credential.allowed_proxy_cidrs,
expiry_unix: credential.expiry_unix,
reusable: Some(credential.reusable),
})
.collect();
let flags = config.get_flags(); let flags = config.get_flags();
let default_flags = default_config.get_flags(); let default_flags = default_config.get_flags();
result.latency_first = Some(flags.latency_first); result.latency_first = Some(flags.latency_first);
@@ -714,6 +742,27 @@ mod tests {
assert_eq!(output.enable_vpn_portal, None); assert_eq!(output.enable_vpn_portal, None);
} }
#[test]
fn managed_credentials_round_trip_through_toml_model() {
let input = NetworkConfig {
managed_credentials: vec![manage::ManagedCredentialConfig {
credential_id: "managed-a".to_owned(),
credential_secret: "secret".to_owned(),
groups: vec!["ops".to_owned()],
allow_relay: true,
allowed_proxy_cidrs: vec!["10.0.0.0/24".to_owned()],
expiry_unix: 2_000_000_000,
reusable: None,
}],
..standalone_config()
};
let config = input.gen_config().unwrap();
let output = NetworkConfig::new_from_config(&config).unwrap();
assert_eq!(output.managed_credentials[0].credential_id, "managed-a");
assert_eq!(output.managed_credentials[0].reusable, Some(true));
}
#[test] #[test]
fn legacy_enabled_vpn_portal_config_reports_migration_error() { fn legacy_enabled_vpn_portal_config_reports_migration_error() {
let error = NetworkConfig { let error = NetworkConfig {
+85
View File
@@ -270,6 +270,11 @@ pub trait ConfigLoader: Send + Sync {
} }
fn set_credential_file(&self, _path: Option<std::path::PathBuf>) {} fn set_credential_file(&self, _path: Option<std::path::PathBuf>) {}
fn get_managed_credentials(&self) -> Vec<ManagedCredentialConfig> {
Vec::new()
}
fn set_managed_credentials(&self, _credentials: Vec<ManagedCredentialConfig>) {}
fn get_network_config_source(&self) -> ConfigSource { fn get_network_config_source(&self) -> ConfigSource {
ConfigSource::User ConfigSource::User
} }
@@ -471,6 +476,41 @@ pub struct VpnPortalClientConfig {
pub groups: Vec<String>, pub groups: Vec<String>,
} }
fn default_true() -> bool {
true
}
#[derive(Clone, Deserialize, Serialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct ManagedCredentialConfig {
pub credential_id: String,
pub credential_secret: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub groups: Vec<String>,
#[serde(default)]
pub allow_relay: bool,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub allowed_proxy_cidrs: Vec<String>,
pub expiry_unix: i64,
#[serde(default = "default_true")]
pub reusable: bool,
}
impl std::fmt::Debug for ManagedCredentialConfig {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("ManagedCredentialConfig")
.field("credential_id", &self.credential_id)
.field("credential_secret", &"<redacted>")
.field("groups", &self.groups)
.field("allow_relay", &self.allow_relay)
.field("allowed_proxy_cidrs", &self.allowed_proxy_cidrs)
.field("expiry_unix", &self.expiry_unix)
.field("reusable", &self.reusable)
.finish()
}
}
#[derive(Debug, Clone, PartialEq, Deserialize)] #[derive(Debug, Clone, PartialEq, Deserialize)]
#[cfg_attr(feature = "config-write", derive(Serialize))] #[cfg_attr(feature = "config-write", derive(Serialize))]
struct Config { struct Config {
@@ -516,6 +556,8 @@ struct Config {
stun_servers_v6: Option<Vec<String>>, stun_servers_v6: Option<Vec<String>>,
credential_file: Option<PathBuf>, credential_file: Option<PathBuf>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
managed_credentials: Vec<ManagedCredentialConfig>,
source: Option<ConfigSourceConfig>, source: Option<ConfigSourceConfig>,
} }
@@ -628,6 +670,11 @@ impl TomlConfig {
} }
} }
} }
for credential in &mut config.managed_credentials {
if !credential.credential_secret.is_empty() {
credential.credential_secret = REDACTED.to_owned();
}
}
} }
pub fn new_from_str(config_str: &str) -> Result<Self, anyhow::Error> { pub fn new_from_str(config_str: &str) -> Result<Self, anyhow::Error> {
@@ -1088,6 +1135,14 @@ impl ConfigLoader for TomlConfig {
self.config.lock().unwrap().credential_file = path; self.config.lock().unwrap().credential_file = path;
} }
fn get_managed_credentials(&self) -> Vec<ManagedCredentialConfig> {
self.config.lock().unwrap().managed_credentials.clone()
}
fn set_managed_credentials(&self, credentials: Vec<ManagedCredentialConfig>) {
self.config.lock().unwrap().managed_credentials = credentials;
}
fn get_network_config_source(&self) -> ConfigSource { fn get_network_config_source(&self) -> ConfigSource {
self.config self.config
.lock() .lock()
@@ -1254,6 +1309,36 @@ group_secret = "group-secret"
assert_eq!(redacted.matches("<redacted>").count(), 4); assert_eq!(redacted.matches("<redacted>").count(), 4);
} }
#[cfg(feature = "config-write")]
#[test]
fn managed_credentials_round_trip_and_redact_secret() {
let config = TomlConfig::new_from_str(
r#"
[[managed_credentials]]
credential_id = "managed-a"
credential_secret = "private-key-material"
groups = ["ops"]
allow_relay = true
allowed_proxy_cidrs = ["10.0.0.0/24"]
expiry_unix = 2000000000
"#,
)
.unwrap();
let dumped = config.dump();
let restored = TomlConfig::new_from_str(&dumped).unwrap();
assert_eq!(
restored.get_managed_credentials(),
config.get_managed_credentials()
);
assert!(dumped.contains("private-key-material"));
let redacted = config.dump_redacted();
assert!(!redacted.contains("private-key-material"));
assert!(redacted.contains("<redacted>"));
assert!(!TomlConfig::default().dump().contains("managed_credentials"));
}
#[test] #[test]
fn hostname_normalization_is_portable_and_has_no_host_fallback() { fn hostname_normalization_is_portable_and_has_no_host_fallback() {
let absent = TomlConfig::default(); let absent = TomlConfig::default();
@@ -474,7 +474,7 @@ where
self.stopping.store(false, Ordering::Release); self.stopping.store(false, Ordering::Release);
} }
reaper.replace(AbortOnDropHandle::new(tokio::spawn( reaper.replace(AbortOnDropHandle::new(tokio::spawn(
reap_joinset_background(self.tasks.clone(), "tcp hole punch"), reap_joinset_background(Arc::downgrade(&self.tasks), "tcp hole punch"),
))); )));
} }
@@ -56,7 +56,10 @@ where
socket_context: SocketContext, socket_context: SocketContext,
) -> Self { ) -> Self {
let tasks = Arc::new(Mutex::new(JoinSet::new())); let tasks = Arc::new(Mutex::new(JoinSet::new()));
tokio::spawn(reap_joinset_background(tasks.clone(), "UdpSocketArray")); tokio::spawn(reap_joinset_background(
Arc::downgrade(&tasks),
"UdpSocketArray",
));
Self { Self {
sockets: Arc::new(DashMap::new()), sockets: Arc::new(DashMap::new()),
+16 -3
View File
@@ -1,6 +1,6 @@
use std::{ use std::{
result::Result, result::Result,
sync::{Arc, Mutex, atomic::Ordering}, sync::{Arc, Mutex, Weak, atomic::Ordering},
time::Duration, time::Duration,
}; };
@@ -15,11 +15,10 @@ use tokio::{
}; };
use tokio_util::task::AbortOnDropHandle; use tokio_util::task::AbortOnDropHandle;
pub(crate) async fn reap_joinset_background<T>(tasks: Arc<Mutex<JoinSet<T>>>, origin: &'static str) pub(crate) async fn reap_joinset_background<T>(tasks: Weak<Mutex<JoinSet<T>>>, origin: &'static str)
where where
T: Send + 'static, T: Send + 'static,
{ {
let tasks = Arc::downgrade(&tasks);
loop { loop {
crate::foundation::time::sleep(Duration::from_secs(1)).await; crate::foundation::time::sleep(Duration::from_secs(1)).await;
let Some(tasks) = tasks.upgrade() else { let Some(tasks) = tasks.upgrade() else {
@@ -282,6 +281,20 @@ mod tests {
} }
} }
#[tokio::test]
async fn joinset_reaper_does_not_keep_task_set_alive() {
let tasks = Arc::new(Mutex::new(JoinSet::new()));
let weak_tasks = Arc::downgrade(&tasks);
tasks
.lock()
.unwrap()
.spawn(reap_joinset_background(weak_tasks.clone(), "test"));
drop(tasks);
assert!(weak_tasks.upgrade().is_none());
}
#[tokio::test] #[tokio::test]
async fn peer_task_manager_is_cold_and_joins_children_on_stop() { async fn peer_task_manager_is_cold_and_joins_children_on_stop() {
let active_tasks = Arc::new(AtomicUsize::new(0)); let active_tasks = Arc::new(AtomicUsize::new(0));
+5 -1
View File
@@ -433,7 +433,11 @@ mod tests {
} }
/// Test high load with concurrent access /// Test high load with concurrent access
#[tokio::test(flavor = "multi_thread", worker_threads = 4)] #[cfg_attr(
not(target_os = "wasi"),
tokio::test(flavor = "multi_thread", worker_threads = 4)
)]
#[cfg_attr(target_os = "wasi", tokio::test)]
async fn test_concurrent_access() { async fn test_concurrent_access() {
let bucket = TokenBucket::new(10_000, 1); let bucket = TokenBucket::new(10_000, 1);
let mut handles = vec![]; let mut handles = vec![];
+1 -1
View File
@@ -506,7 +506,7 @@ where
.lock() .lock()
.unwrap() .unwrap()
.spawn(reap_joinset_background( .spawn(reap_joinset_background(
self.runtime_tasks.clone(), Arc::downgrade(&self.runtime_tasks),
"data plane runtime", "data plane runtime",
)); ));
self.run_net_update_task().await; self.run_net_update_task().await;
+1 -1
View File
@@ -105,7 +105,7 @@ impl SmoltcpPlane {
let forward_tasks = Arc::new(std::sync::Mutex::new(forward_tasks)); let forward_tasks = Arc::new(std::sync::Mutex::new(forward_tasks));
forward_tasks.lock().unwrap().spawn(reap_joinset_background( forward_tasks.lock().unwrap().spawn(reap_joinset_background(
forward_tasks.clone(), Arc::downgrade(&forward_tasks),
"SmoltcpPlane", "SmoltcpPlane",
)); ));
@@ -653,6 +653,53 @@ async fn immediate_consumer_reacquire_never_leases_closing_generation() {
endpoint.peer_manager.clear_resources().await; endpoint.peer_manager.clear_resources().await;
} }
#[tokio::test]
async fn tcp_connect_survives_ipv4_generation_replacement() {
let (a, b) = setup_data_plane_pair().await;
let _consumer = b.gateway.acquire_consumer_lease().unwrap();
for ip in ["10.126.127.2", "10.126.126.2"] {
let ip: IpAddr = ip.parse().unwrap();
b.gateway.runtime_config.update_peer_with(|peer| {
peer.runtime.core.routes.ipv4 = Some(IpPrefix::new(ip, 24).unwrap());
});
tokio::time::timeout(Duration::from_secs(1), async {
loop {
if b.gateway
.net
.lock()
.await
.as_ref()
.is_some_and(|plane| IpAddr::V4(plane.ipv4_addr.address()) == ip)
{
break;
}
tokio::task::yield_now().await;
}
})
.await
.expect("data-plane IPv4 generation did not update");
}
let timeout = Duration::from_secs(10);
let mut listener = b.gateway.data_plane_tcp_bind(0, timeout).await.unwrap();
let listen_addr = SocketAddr::new(b.ip.address().into(), listener.local_addr().port());
let (accepted, client) = tokio::join!(
listener.accept(),
a.gateway.data_plane_tcp_connect(listen_addr, timeout),
);
let (mut server, _) = accepted.unwrap();
let mut client = client.unwrap();
client.write_all(b"ping").await.unwrap();
client.flush().await.unwrap();
let mut buf = [0u8; 4];
server.read_exact(&mut buf).await.unwrap();
assert_eq!(&buf, b"ping");
stop_data_plane_pair(&a, &b).await;
}
#[tokio::test] #[tokio::test]
async fn ipv4_change_closes_existing_generation_with_typed_error() { async fn ipv4_change_closes_existing_generation_with_typed_error() {
let host = Arc::new(TestHost::default()); let host = Arc::new(TestHost::default());
+7 -3
View File
@@ -148,7 +148,7 @@ where
return Ok(()); return Ok(());
} }
self.tasks.lock().unwrap().spawn(reap_joinset_background( self.tasks.lock().unwrap().spawn(reap_joinset_background(
self.tasks.clone(), Arc::downgrade(&self.tasks),
"port-forward adapter", "port-forward adapter",
)); ));
self.start_udp_reaper(); self.start_udp_reaper();
@@ -246,7 +246,7 @@ where
let data_plane = self.data_plane.clone(); let data_plane = self.data_plane.clone();
let connections = Arc::new(std::sync::Mutex::new(JoinSet::new())); let connections = Arc::new(std::sync::Mutex::new(JoinSet::new()));
connections.lock().unwrap().spawn(reap_joinset_background( connections.lock().unwrap().spawn(reap_joinset_background(
connections.clone(), Arc::downgrade(&connections),
"TCP port-forward connections", "TCP port-forward connections",
)); ));
self.tasks.lock().unwrap().spawn(async move { self.tasks.lock().unwrap().spawn(async move {
@@ -624,7 +624,11 @@ mod tests {
assert_eq!(slots.available_permits(), 2); assert_eq!(slots.available_permits(), 2);
} }
#[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[cfg_attr(
not(target_os = "wasi"),
tokio::test(flavor = "multi_thread", worker_threads = 2)
)]
#[cfg_attr(target_os = "wasi", tokio::test)]
async fn udp_client_admission_covers_client_and_response_task_publication() { async fn udp_client_admission_covers_client_and_response_task_publication() {
let slots = Arc::new(Semaphore::new(1)); let slots = Arc::new(Semaphore::new(1));
let admission = Arc::new(Mutex::new(())); let admission = Arc::new(Mutex::new(()));
+1 -1
View File
@@ -159,7 +159,7 @@ where
let consumer_lease = self.data_plane.acquire_consumer_lease()?; let consumer_lease = self.data_plane.acquire_consumer_lease()?;
self.tasks.lock().unwrap().spawn(reap_joinset_background( self.tasks.lock().unwrap().spawn(reap_joinset_background(
self.tasks.clone(), Arc::downgrade(&self.tasks),
"SOCKS5 gateway adapter", "SOCKS5 gateway adapter",
)); ));
let data_plane = self.data_plane.clone(); let data_plane = self.data_plane.clone();
@@ -1144,6 +1144,7 @@ mod tests {
let peer = Arc::new( let peer = Arc::new(
PeerManagerCore::new( PeerManagerCore::new(
portable, portable,
Vec::new(),
store.clone(), store.clone(),
Arc::new(()), Arc::new(()),
packet_sender, packet_sender,
+29
View File
@@ -173,6 +173,12 @@ impl CoreInstanceConfig {
let flags = host.runtime_flags(config.get_flags()); let flags = host.runtime_flags(config.get_flags());
let instance_id = config.get_id(); let instance_id = config.get_id();
let identity: crate::config::NetworkIdentity = config.get_network_identity().into(); let identity: crate::config::NetworkIdentity = config.get_network_identity().into();
let managed_credentials = config.get_managed_credentials();
if !managed_credentials.is_empty() && identity.network_secret.is_none() {
anyhow::bail!(
"only admin nodes with a network_secret can configure managed credentials"
);
}
let network_name = identity.network_name.clone(); let network_name = identity.network_name.clone();
let socket_context = SocketContext::default() let socket_context = SocketContext::default()
.with_socket_mark(flags.socket_mark) .with_socket_mark(flags.socket_mark)
@@ -325,6 +331,7 @@ impl CoreInstanceConfig {
Ok(Self { Ok(Self {
instance_name: config.get_inst_name(), instance_name: config.get_inst_name(),
peer, peer,
managed_credentials,
vpn_portal: (!host.ignore_unsupported_config || host.vpn_portal_enabled) vpn_portal: (!host.ignore_unsupported_config || host.vpn_portal_enabled)
.then(|| config.get_vpn_portal_config()) .then(|| config.get_vpn_portal_config())
.flatten() .flatten()
@@ -391,6 +398,8 @@ impl CoreInstanceConfig {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD};
use super::*; use super::*;
#[test] #[test]
@@ -456,6 +465,26 @@ stun_servers_v6 = ["custom-v6.example.com:3478"]
); );
} }
#[test]
fn credential_nodes_cannot_declare_managed_credentials() {
let config = TomlConfig::default();
config.set_network_identity(crate::config::toml::NetworkIdentity::new_credential(
"credential-network".to_owned(),
));
config.set_managed_credentials(vec![crate::config::toml::ManagedCredentialConfig {
credential_id: "managed".to_owned(),
credential_secret: BASE64_STANDARD.encode([1u8; 32]),
groups: Vec::new(),
allow_relay: false,
allowed_proxy_cidrs: Vec::new(),
expiry_unix: 2_000_000_000,
reusable: true,
}]);
let error = CoreInstanceConfig::from_toml(&config).unwrap_err();
assert!(error.to_string().contains("only admin nodes"));
}
#[cfg(feature = "config-write")] #[cfg(feature = "config-write")]
#[test] #[test]
fn explicit_stun_servers_survive_dump_reload() { fn explicit_stun_servers_survive_dump_reload() {
+16 -2
View File
@@ -164,7 +164,8 @@ where
options.ttl, options.ttl,
options.credential_id, options.credential_id,
options.reusable, options.reusable,
); )
.map_err(anyhow::Error::msg)?;
self.peer_manager.notify_credential_changed(); self.peer_manager.notify_credential_changed();
Ok(generated) Ok(generated)
} }
@@ -176,7 +177,8 @@ where
let revoked = self let revoked = self
.peer_manager .peer_manager
.credential_manager() .credential_manager()
.revoke_credential(credential_id); .revoke_credential(credential_id)
.map_err(anyhow::Error::msg)?;
if revoked { if revoked {
self.peer_manager.notify_credential_changed(); self.peer_manager.notify_credential_changed();
} }
@@ -202,6 +204,18 @@ where
self.peer_manager.credential_manager().list_credentials() self.peer_manager.credential_manager().list_credentials()
} }
#[cfg(feature = "web-client")]
pub(crate) fn credential_manager(
&self,
) -> Arc<crate::peers::credential_manager::CredentialManager> {
self.peer_manager.credential_manager()
}
#[cfg(feature = "web-client")]
pub(crate) fn notify_credential_changed(&self) {
self.peer_manager.notify_credential_changed();
}
pub fn metric_snapshots(&self) -> Vec<MetricSnapshot> { pub fn metric_snapshots(&self) -> Vec<MetricSnapshot> {
self.peer_manager.stats_manager().get_all_metrics() self.peer_manager.stats_manager().get_all_metrics()
} }
+6 -6
View File
@@ -291,6 +291,12 @@ impl<F: InstanceFactory> InstanceManager<F> {
.remove(&instance_id) .remove(&instance_id)
} }
pub fn config_control(&self, instance_id: Uuid) -> Option<ConfigFileControl> {
self.config_controls
.get(&instance_id)
.map(|control| control.clone())
}
pub fn mutation_lock(&self) -> Arc<tokio::sync::Mutex<()>> { pub fn mutation_lock(&self) -> Arc<tokio::sync::Mutex<()>> {
self.mutation_lock.clone() self.mutation_lock.clone()
} }
@@ -405,12 +411,6 @@ where
self.list() self.list()
} }
pub fn config_control(&self, instance_id: Uuid) -> Option<ConfigFileControl> {
self.config_controls
.get(&instance_id)
.map(|control| control.clone())
}
pub fn attach_tun_fd(&self, instance_id: Uuid, fd: i32) -> anyhow::Result<()> { pub fn attach_tun_fd(&self, instance_id: Uuid, fd: i32) -> anyhow::Result<()> {
self.get(instance_id) self.get(instance_id)
.ok_or_else(|| anyhow::anyhow!("instance {instance_id} not found"))? .ok_or_else(|| anyhow::anyhow!("instance {instance_id} not found"))?
+3
View File
@@ -184,6 +184,8 @@ pub struct CoreInstanceConfig {
pub connectivity: CoreConnectivityConfig, pub connectivity: CoreConnectivityConfig,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub vpn_portal: Option<PortalRuntimeConfig>, pub vpn_portal: Option<PortalRuntimeConfig>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub managed_credentials: Vec<crate::config::toml::ManagedCredentialConfig>,
} }
#[cfg(any(test, feature = "test-utils"))] #[cfg(any(test, feature = "test-utils"))]
@@ -503,6 +505,7 @@ where
)); ));
let peer_manager = Arc::new(PeerManagerCore::new( let peer_manager = Arc::new(PeerManagerCore::new(
config.peer, config.peer,
config.managed_credentials,
runtime_config.clone(), runtime_config.clone(),
Arc::new(CoreStunPeerInfoSource(peer_stun)), Arc::new(CoreStunPeerInfoSource(peer_stun)),
packet_tx, packet_tx,
+169
View File
@@ -174,6 +174,7 @@ fn core_instance_config_round_trips_as_normalized_json() {
peer, peer,
connectivity: CoreConnectivityConfig::default(), connectivity: CoreConnectivityConfig::default(),
vpn_portal: None, vpn_portal: None,
managed_credentials: Vec::new(),
}; };
let mut config = config; let mut config = config;
@@ -328,6 +329,7 @@ mod portable_runtime {
peer, peer,
connectivity, connectivity,
vpn_portal: None, vpn_portal: None,
managed_credentials: Vec::new(),
} }
} }
#[cfg(feature = "vpn-portal")] #[cfg(feature = "vpn-portal")]
@@ -452,6 +454,28 @@ mod portable_runtime {
fn build_instance(config: CoreInstanceConfig) -> anyhow::Result<Arc<CoreInstance<TestHost>>> { fn build_instance(config: CoreInstanceConfig) -> anyhow::Result<Arc<CoreInstance<TestHost>>> {
build_with_engines(config, WrappedTransportEngines::default()) build_with_engines(config, WrappedTransportEngines::default())
} }
#[cfg(feature = "management")]
struct RecordingConfigPatchPersistence {
writes: std::sync::Mutex<Vec<String>>,
fail: AtomicBool,
}
#[cfg(feature = "management")]
#[async_trait]
impl crate::management::ConfigPatchPersistence for RecordingConfigPatchPersistence {
async fn persist(
&self,
_instance_id: uuid::Uuid,
config: &TomlConfig,
) -> anyhow::Result<()> {
if self.fail.load(Ordering::Relaxed) {
anyhow::bail!("injected config persistence failure");
}
self.writes.lock().unwrap().push(config.dump());
Ok(())
}
}
#[cfg(feature = "vpn-portal")] #[cfg(feature = "vpn-portal")]
#[tokio::test] #[tokio::test]
async fn runtime_update_rejects_portal_client_address_conflict() { async fn runtime_update_rejects_portal_client_address_conflict() {
@@ -613,12 +637,14 @@ mod portable_runtime {
instance::manager::{InstanceFactory, InstanceManager}, instance::manager::{InstanceFactory, InstanceManager},
management::InstanceManagementRpc, management::InstanceManagementRpc,
}; };
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD};
use easytier_proto::{ use easytier_proto::{
api::config::{ConfigRpc, GetConfigRequest, InstanceConfigPatch, PatchConfigRequest}, api::config::{ConfigRpc, GetConfigRequest, InstanceConfigPatch, PatchConfigRequest},
api::instance::{ api::instance::{
PeerManageRpc, ShowNodeInfoRequest, PeerManageRpc, ShowNodeInfoRequest,
instance_identifier::{InstanceSelector, Selector}, instance_identifier::{InstanceSelector, Selector},
}, },
api::manage::{ManagedCredentialConfig, ManagedCredentialSet},
rpc_types::controller::BaseController, rpc_types::controller::BaseController,
}; };
@@ -648,6 +674,10 @@ mod portable_runtime {
r#" r#"
instance_name = "managed-by-name" instance_name = "managed-by-name"
hostname = "core-owned-config" hostname = "core-owned-config"
[network_identity]
network_name = "managed-network"
network_secret = "network-secret"
"#, "#,
) )
.unwrap(); .unwrap();
@@ -703,6 +733,46 @@ hostname = "core-owned-config"
response.config.unwrap().hostname.as_deref(), response.config.unwrap().hostname.as_deref(),
Some("patched-in-core") Some("patched-in-core")
); );
let secret = BASE64_STANDARD.encode([9u8; 32]);
rpc.patch_config(
BaseController::default(),
PatchConfigRequest {
patch: Some(InstanceConfigPatch {
managed_credentials: Some(ManagedCredentialSet {
entries: vec![ManagedCredentialConfig {
credential_id: "pathless".to_owned(),
credential_secret: secret.clone(),
expiry_unix: i64::MAX,
..Default::default()
}],
}),
..Default::default()
}),
instance: Some(selector()),
},
)
.await
.unwrap();
let response = rpc
.get_config(
BaseController::default(),
GetConfigRequest {
instance: Some(selector()),
},
)
.await
.unwrap();
assert_eq!(
response.config.unwrap().managed_credentials,
vec![ManagedCredentialConfig {
credential_id: "pathless".to_owned(),
credential_secret: secret,
expiry_unix: i64::MAX,
reusable: Some(true),
..Default::default()
}]
);
let runtime = instance.runtime_config.snapshot(); let runtime = instance.runtime_config.snapshot();
assert!(runtime.services.proxy.enable_exit_node); assert!(runtime.services.proxy.enable_exit_node);
assert!(runtime.services.public_ipv6_provider.provider_supported); assert!(runtime.services.public_ipv6_provider.provider_supported);
@@ -720,6 +790,7 @@ hostname = "core-owned-config"
hostname: Some("too-early".to_owned()), hostname: Some("too-early".to_owned()),
..Default::default() ..Default::default()
}, },
None,
) )
.await .await
.unwrap_err(); .unwrap_err();
@@ -727,6 +798,101 @@ hostname = "core-owned-config"
assert!(error.to_string().contains("instance is not ready")); assert!(error.to_string().contains("instance is not ready"));
} }
#[cfg(feature = "management")]
#[tokio::test]
async fn managed_credential_patch_is_durable_atomic_and_does_not_restart() {
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD};
use easytier_proto::api::{
config::InstanceConfigPatch,
manage::{ManagedCredentialConfig, ManagedCredentialSet},
};
let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
let config = TomlConfig::new_from_str(
r#"
[network_identity]
network_name = "managed-network"
network_secret = "network-secret"
[source]
source = "web"
"#,
)
.unwrap();
let instance =
CoreInstance::from_toml(config, adapters(None, Arc::new(packet_sink))).unwrap();
instance.start().await.unwrap();
let peer_id = instance.peer_id();
let secret = BASE64_STANDARD.encode([7u8; 32]);
let patch = InstanceConfigPatch {
managed_credentials: Some(ManagedCredentialSet {
entries: vec![ManagedCredentialConfig {
credential_id: "managed".to_owned(),
credential_secret: secret.clone(),
groups: vec!["ops".to_owned()],
allow_relay: false,
allowed_proxy_cidrs: Vec::new(),
expiry_unix: 2_000_000_000,
reusable: Some(true),
}],
}),
..Default::default()
};
let persistence = RecordingConfigPatchPersistence {
writes: std::sync::Mutex::new(Vec::new()),
fail: AtomicBool::new(true),
};
let error =
crate::management::apply_config_patch(&instance, patch.clone(), Some(&persistence))
.await
.unwrap_err();
assert!(
error
.to_string()
.contains("injected config persistence failure")
);
assert!(
instance
.toml_config()
.unwrap()
.get_managed_credentials()
.is_empty()
);
let private_bytes: [u8; 32] = BASE64_STANDARD.decode(&secret).unwrap().try_into().unwrap();
let public_key =
x25519_dalek::PublicKey::from(&x25519_dalek::StaticSecret::from(private_bytes));
assert!(
!instance
.credential_manager()
.is_pubkey_trusted(public_key.as_bytes())
);
persistence.fail.store(false, Ordering::Relaxed);
crate::management::apply_config_patch(&instance, patch, Some(&persistence))
.await
.unwrap();
assert_eq!(instance.peer_id(), peer_id);
assert_eq!(instance.state(), CoreInstanceState::Running);
assert!(
instance
.credential_manager()
.is_pubkey_trusted(public_key.as_bytes())
);
assert_eq!(
instance
.toml_config()
.unwrap()
.get_managed_credentials()
.len(),
1
);
let persisted = persistence.writes.lock().unwrap();
assert_eq!(persisted.len(), 1);
assert!(persisted[0].contains(&secret));
}
#[cfg(all(feature = "management", not(feature = "proxy-smoltcp-stack")))] #[cfg(all(feature = "management", not(feature = "proxy-smoltcp-stack")))]
#[tokio::test] #[tokio::test]
async fn unavailable_gateway_patch_does_not_commit_shared_toml() { async fn unavailable_gateway_patch_does_not_commit_shared_toml() {
@@ -769,6 +935,7 @@ hostname = "core-owned-config"
}], }],
..Default::default() ..Default::default()
}, },
None,
) )
.await .await
.unwrap_err(); .unwrap_err();
@@ -833,6 +1000,7 @@ virtual_ip = "10.82.0.2"
ipv4: Some("10.82.0.2/24".parse::<cidr::Ipv4Inet>().unwrap().into()), ipv4: Some("10.82.0.2/24".parse::<cidr::Ipv4Inet>().unwrap().into()),
..Default::default() ..Default::default()
}, },
None,
) )
.await .await
.unwrap_err(); .unwrap_err();
@@ -909,6 +1077,7 @@ virtual_ip = "10.82.0.2"
}], }],
..Default::default() ..Default::default()
}, },
None,
) )
.await .await
.unwrap(); .unwrap();
@@ -13,6 +13,7 @@ use easytier_proto::{
}; };
use super::super::instance_rpc::InstanceManagementRpc; use super::super::instance_rpc::InstanceManagementRpc;
use super::ConfigFileStorage;
use crate::{ use crate::{
instance::{ instance::{
CoreInstance, CoreInstanceHost, CoreInstance, CoreInstanceHost,
@@ -26,11 +27,12 @@ use crate::{
pub fn register_instance_management_rpc<F, H>( pub fn register_instance_management_rpc<F, H>(
manager: Arc<InstanceManager<F>>, manager: Arc<InstanceManager<F>>,
registry: &ServiceRegistry, registry: &ServiceRegistry,
storage: Arc<dyn ConfigFileStorage>,
) where ) where
F: InstanceFactory<Instance = CoreInstance<H>>, F: InstanceFactory<Instance = CoreInstance<H>>,
H: CoreInstanceHost, H: CoreInstanceHost,
{ {
let rpc = InstanceManagementRpc::<F>::new(manager.clone()); let rpc = InstanceManagementRpc::<F>::new_with_config_storage(manager.clone(), storage);
registry.register(PeerManageRpcServer::new(rpc.clone()), ""); registry.register(PeerManageRpcServer::new(rpc.clone()), "");
registry.register(ConnectorManageRpcServer::new(rpc.clone()), ""); registry.register(ConnectorManageRpcServer::new(rpc.clone()), "");
registry.register(MappedListenerManageRpcServer::new(rpc.clone()), ""); registry.register(MappedListenerManageRpcServer::new(rpc.clone()), "");
+113 -12
View File
@@ -10,14 +10,21 @@ use crate::{
config::{ config::{
peers::AclRuleConfig, peers::AclRuleConfig,
runtime::CoreInstanceRuntimeConfig, runtime::CoreInstanceRuntimeConfig,
toml::{ConfigLoader as _, TomlConfig}, toml::{ConfigLoader as _, ManagedCredentialConfig, TomlConfig},
}, },
instance::{CoreInstance, CoreInstanceConfig, CoreInstanceHost, CoreInstanceState}, instance::{CoreInstance, CoreInstanceConfig, CoreInstanceHost, CoreInstanceState},
peers::credential_manager::CredentialManager,
}; };
#[async_trait::async_trait]
pub trait ConfigPatchPersistence: Send + Sync {
async fn persist(&self, instance_id: uuid::Uuid, config: &TomlConfig) -> anyhow::Result<()>;
}
pub async fn apply_config_patch<H>( pub async fn apply_config_patch<H>(
instance: &Arc<CoreInstance<H>>, instance: &Arc<CoreInstance<H>>,
patch: InstanceConfigPatch, patch: InstanceConfigPatch,
persistence: Option<&dyn ConfigPatchPersistence>,
) -> anyhow::Result<()> ) -> anyhow::Result<()>
where where
H: CoreInstanceHost, H: CoreInstanceHost,
@@ -33,11 +40,15 @@ where
let candidate = config.detached_snapshot(); let candidate = config.detached_snapshot();
let parsed_prefix = let parsed_prefix =
parse_ipv6_public_addr_prefix_patch(patch.ipv6_public_addr_prefix.as_deref())?; parse_ipv6_public_addr_prefix_patch(patch.ipv6_public_addr_prefix.as_deref())?;
let patch_for_host = patch.clone(); // Take the credential set out first so the host-facing copy below never
// clones secret material.
let mut patch = patch;
let managed_credentials = patch.managed_credentials.take();
let patch_for_host = patch_without_managed_credentials(&patch);
// Preserve the existing ordered partial-commit contract: earlier valid // Preserve the existing ordered partial-commit contract: earlier valid
// sub-patches remain applied if a later sub-patch fails. // sub-patches remain applied if a later sub-patch fails.
let patch_result: anyhow::Result<bool> = async { let patch_result: anyhow::Result<(bool, bool)> = async {
let result = patch_port_forwards(&candidate, patch.port_forwards); let result = patch_port_forwards(&candidate, patch.port_forwards);
validate_and_commit_candidate(instance, &config, &candidate)?; validate_and_commit_candidate(instance, &config, &candidate)?;
result?; result?;
@@ -95,6 +106,7 @@ where
candidate.set_ipv6_public_addr_prefix(prefix); candidate.set_ipv6_public_addr_prefix(prefix);
provider_config_changed = true; provider_config_changed = true;
} }
let mut managed_credentials_changed = false;
// Runs last so client validation sees the fully patched candidate, // Runs last so client validation sees the fully patched candidate,
// including routes and the node IPv4 set earlier in this request. // including routes and the node IPv4 set earlier in this request.
@@ -124,22 +136,84 @@ where
validate_and_commit_candidate(instance, &config, &candidate)?; validate_and_commit_candidate(instance, &config, &candidate)?;
} }
let normalized = validate_and_commit_candidate(instance, &config, &candidate)?; if let Some(managed) = &managed_credentials {
// Managed credential patch transaction: validate and reserve →
// persist → install. The reservation prevents base or ephemeral
// credential mutations from invalidating the replacement while
// the durable write is in flight, without holding a synchronous
// lock across the await. Dropping the replacement before install
// releases the reservation.
//
// Accepted consistency limits:
//
// 1. A persistence implementation may finish its write after this
// RPC future is cancelled. The reservation is then released and
// the running instance keeps its previous credentials even if
// the durable file contains the replacement. A retry, controller
// reconcile, or restart is required to converge; until then a
// removed credential may remain trusted by the running instance.
//
// 2. This instance operation is not serialized with a process-level
// instance overwrite. The built-in web reconciler serializes its
// own actions, but independently concurrent admin RPCs are
// last-writer-wins and may leave the running instance and durable
// file on different config generations. A restart aligns runtime
// with the file; controller reconcile is required to restore its
// desired generation.
let credential_manager = instance.credential_manager();
let entries = managed
.entries
.iter()
.map(|credential| ManagedCredentialConfig {
credential_id: credential.credential_id.clone(),
credential_secret: credential.credential_secret.clone(),
groups: credential.groups.clone(),
allow_relay: credential.allow_relay,
allowed_proxy_cidrs: credential.allowed_proxy_cidrs.clone(),
expiry_unix: credential.expiry_unix,
reusable: credential.reusable.unwrap_or(true),
})
.collect::<Vec<_>>();
let replacement = credential_manager
.validate_managed_credentials(&entries)
.map_err(anyhow::Error::msg)?;
candidate.set_managed_credentials(entries);
validate_candidate(instance, &candidate)?;
// File-backed configs persist every successful patch, so the
// durable file and the shared TOML model can never diverge.
persistence
.ok_or_else(|| anyhow::anyhow!("durable config patching is unavailable"))?
.persist(instance.instance_id(), &candidate)
.await?;
config.replace_from_snapshot(&candidate);
managed_credentials_changed =
CredentialManager::install_managed_credentials(replacement);
} else {
validate_and_commit_candidate(instance, &config, &candidate)?;
}
let normalized = validate_candidate(instance, &candidate)?;
let runtime = runtime_config_from_normalized(&normalized); let runtime = runtime_config_from_normalized(&normalized);
instance if patch_for_host != InstanceConfigPatch::default() {
.instance_runtime instance
.synchronize_config(&patch_for_host, &runtime); .instance_runtime
Ok(provider_config_changed) .synchronize_config(&patch_for_host, &runtime);
}
Ok((provider_config_changed, managed_credentials_changed))
} }
.await; .await;
instance instance
.update_runtime_config_under_operation(runtime_config_from_toml(instance, &config)?) .update_runtime_config_under_operation(runtime_config_from_toml(instance, &config)?)
.await?; .await?;
let provider_config_changed = patch_result?; let (provider_config_changed, managed_credentials_changed) = patch_result?;
instance if patch_for_host != InstanceConfigPatch::default() {
.instance_runtime instance
.publish_config_patch(patch_for_host); .instance_runtime
.publish_config_patch(patch_for_host);
}
if managed_credentials_changed {
instance.notify_credential_changed();
}
#[cfg(feature = "public-ipv6-provider")] #[cfg(feature = "public-ipv6-provider")]
if provider_config_changed && instance.state() == CoreInstanceState::Running { if provider_config_changed && instance.state() == CoreInstanceState::Running {
instance.reconcile_public_ipv6_provider().await; instance.reconcile_public_ipv6_provider().await;
@@ -149,6 +223,12 @@ where
Ok(()) Ok(())
} }
fn patch_without_managed_credentials(patch: &InstanceConfigPatch) -> InstanceConfigPatch {
let mut patch = patch.clone();
patch.managed_credentials = None;
patch
}
fn validate_candidate<H>( fn validate_candidate<H>(
instance: &CoreInstance<H>, instance: &CoreInstance<H>,
candidate: &TomlConfig, candidate: &TomlConfig,
@@ -227,6 +307,27 @@ fn trace_patchables<T: Debug>(patches: &[Patchable<T>]) {
} }
} }
#[cfg(test)]
mod managed_credential_tests {
use easytier_proto::api::manage::ManagedCredentialSet;
use super::*;
#[test]
fn event_patch_drops_managed_credential_secrets() {
let patch = InstanceConfigPatch {
managed_credentials: Some(ManagedCredentialSet::default()),
..Default::default()
};
assert!(
patch_without_managed_credentials(&patch)
.managed_credentials
.is_none()
);
}
}
fn patch_port_forwards(config: &TomlConfig, patches: Vec<PortForwardPatch>) -> anyhow::Result<()> { fn patch_port_forwards(config: &TomlConfig, patches: Vec<PortForwardPatch>) -> anyhow::Result<()> {
if patches.is_empty() { if patches.is_empty() {
return Ok(()); return Ok(());
+6 -3
View File
@@ -37,7 +37,7 @@ use super::{
#[cfg(feature = "management")] #[cfg(feature = "management")]
pub use compiled::register_instance_management_rpc; pub use compiled::register_instance_management_rpc;
pub use config_patch::apply_config_patch; pub use config_patch::{ConfigPatchPersistence, apply_config_patch};
pub use instance_info::network_instance_running_info; pub use instance_info::network_instance_running_info;
#[cfg(feature = "management")] #[cfg(feature = "management")]
pub use logger_rpc::{ pub use logger_rpc::{
@@ -82,7 +82,7 @@ pub fn register_management_rpc<F, H>(
F::Error: std::fmt::Debug + std::fmt::Display + Send + Sync + 'static, F::Error: std::fmt::Debug + std::fmt::Display + Send + Sync + 'static,
H: CoreInstanceHost, H: CoreInstanceHost,
{ {
register_instance_management_rpc(instances.clone(), registry); register_instance_management_rpc(instances.clone(), registry, storage.clone());
registry.register(LoggerRpcServer::new(LoggerManagementRpc::new(logger)), ""); registry.register(LoggerRpcServer::new(LoggerManagementRpc::new(logger)), "");
registry.register( registry.register(
WebClientServiceServer::new(ProcessManagementRpc::<F>::new(instances, hooks, storage)), WebClientServiceServer::new(ProcessManagementRpc::<F>::new(instances, hooks, storage)),
@@ -102,7 +102,10 @@ pub(crate) fn register_web_client_rpc<F, H>(
F::Error: std::fmt::Debug + std::fmt::Display + Send + Sync + 'static, F::Error: std::fmt::Debug + std::fmt::Display + Send + Sync + 'static,
H: CoreInstanceHost, H: CoreInstanceHost,
{ {
let config_rpc = super::instance_rpc::InstanceManagementRpc::<F>::new(instances.clone()); let config_rpc = super::instance_rpc::InstanceManagementRpc::<F>::new_with_config_storage(
instances.clone(),
storage.clone(),
);
registry.register(ConfigRpcServer::new(config_rpc), ""); registry.register(ConfigRpcServer::new(config_rpc), "");
registry.register( registry.register(
WebClientServiceServer::new(ProcessManagementRpc::<F>::new(instances, hooks, storage)), WebClientServiceServer::new(ProcessManagementRpc::<F>::new(instances, hooks, storage)),
@@ -63,6 +63,8 @@ pub trait ConfigFileStorage: Send + Sync + 'static {
async fn read(&self, path: &Path) -> anyhow::Result<Option<Vec<u8>>>; async fn read(&self, path: &Path) -> anyhow::Result<Option<Vec<u8>>>;
/// Atomically replaces the file and restricts newly created files to the
/// current user when the Host supports file permissions.
async fn write(&self, path: &Path, contents: &[u8]) -> anyhow::Result<()>; async fn write(&self, path: &Path, contents: &[u8]) -> anyhow::Result<()>;
async fn remove(&self, path: &Path) -> anyhow::Result<()>; async fn remove(&self, path: &Path) -> anyhow::Result<()>;
@@ -26,7 +26,7 @@ where
) -> rpc_types::error::Result<PatchConfigResponse> { ) -> rpc_types::error::Result<PatchConfigResponse> {
let instance = self.instance(request.instance.as_ref())?; let instance = self.instance(request.instance.as_ref())?;
if let Some(patch) = request.patch { if let Some(patch) = request.patch {
apply_config_patch(&instance, patch).await?; apply_config_patch(&instance, patch, self.config_patch_persistence.as_deref()).await?;
} }
Ok(PatchConfigResponse::default()) Ok(PatchConfigResponse::default())
} }
@@ -16,7 +16,10 @@ use easytier_proto::{
}; };
use crate::{ use crate::{
config::{IpPrefix, ProxyNetworkConfig}, config::{
IpPrefix, ProxyNetworkConfig,
toml::{ConfigLoader as _, TomlConfig},
},
connectivity::manual::{ManualConnectorSnapshot, ManualConnectorStatus}, connectivity::manual::{ManualConnectorSnapshot, ManualConnectorStatus},
instance::{ instance::{
CoreInstance, CoreInstanceHost, CoreInstance, CoreInstanceHost,
@@ -26,6 +29,8 @@ use crate::{
}; };
use super::resolve_instance; use super::resolve_instance;
#[cfg(feature = "web-client")]
use super::{ConfigFileStorage, ConfigPatchPersistence};
#[cfg(feature = "web-client")] #[cfg(feature = "web-client")]
mod config; mod config;
@@ -124,6 +129,8 @@ where
#[doc(hidden)] #[doc(hidden)]
pub struct ResolvedInstanceManagementRpc<R> { pub struct ResolvedInstanceManagementRpc<R> {
resolver: R, resolver: R,
#[cfg(feature = "web-client")]
config_patch_persistence: Option<Arc<dyn ConfigPatchPersistence>>,
} }
impl<R> Clone for ResolvedInstanceManagementRpc<R> impl<R> Clone for ResolvedInstanceManagementRpc<R>
@@ -133,6 +140,8 @@ where
fn clone(&self) -> Self { fn clone(&self) -> Self {
Self { Self {
resolver: self.resolver.clone(), resolver: self.resolver.clone(),
#[cfg(feature = "web-client")]
config_patch_persistence: self.config_patch_persistence.clone(),
} }
} }
} }
@@ -158,8 +167,30 @@ where
H: CoreInstanceHost, H: CoreInstanceHost,
{ {
pub fn new(manager: Arc<InstanceManager<F>>) -> Self { pub fn new(manager: Arc<InstanceManager<F>>) -> Self {
#[cfg(feature = "web-client")]
let persistence = Arc::new(ManagerPathlessConfigPatchPersistence {
manager: manager.clone(),
_host: std::marker::PhantomData,
});
Self { Self {
resolver: ManagerInstanceResolver { manager }, resolver: ManagerInstanceResolver { manager },
#[cfg(feature = "web-client")]
config_patch_persistence: Some(persistence),
}
}
#[cfg(feature = "web-client")]
pub fn new_with_config_storage(
manager: Arc<InstanceManager<F>>,
storage: Arc<dyn ConfigFileStorage>,
) -> Self {
let persistence = Arc::new(ManagerConfigPatchPersistence {
manager: manager.clone(),
storage,
_host: std::marker::PhantomData,
});
Self {
resolver: ManagerInstanceResolver { manager },
config_patch_persistence: Some(persistence),
} }
} }
@@ -178,6 +209,179 @@ where
{ {
ResolvedInstanceManagementRpc { ResolvedInstanceManagementRpc {
resolver: BoundInstanceResolver { instance }, resolver: BoundInstanceResolver { instance },
#[cfg(feature = "web-client")]
config_patch_persistence: Some(Arc::new(InMemoryConfigPatchPersistence)),
}
}
#[cfg(all(feature = "web-client", target_os = "wasi"))]
struct InMemoryConfigPatchPersistence;
#[async_trait::async_trait]
#[cfg(all(feature = "web-client", target_os = "wasi"))]
impl ConfigPatchPersistence for InMemoryConfigPatchPersistence {
async fn persist(&self, _instance_id: uuid::Uuid, _config: &TomlConfig) -> anyhow::Result<()> {
Ok(())
}
}
#[cfg(feature = "web-client")]
struct ManagerPathlessConfigPatchPersistence<F, H>
where
F: InstanceFactory,
H: CoreInstanceHost,
{
manager: Arc<InstanceManager<F>>,
_host: std::marker::PhantomData<fn() -> H>,
}
#[async_trait::async_trait]
#[cfg(feature = "web-client")]
impl<F, H> ConfigPatchPersistence for ManagerPathlessConfigPatchPersistence<F, H>
where
F: InstanceFactory<Instance = CoreInstance<H>>,
H: CoreInstanceHost,
{
async fn persist(&self, instance_id: uuid::Uuid, _config: &TomlConfig) -> anyhow::Result<()> {
let Some(control) = self.manager.config_control(instance_id) else {
return Ok(());
};
if control.is_read_only() {
anyhow::bail!("configuration file is read-only");
}
if let Some(path) = control.path {
anyhow::bail!(
"config file {} requires a durable config storage backend",
path.display()
);
}
Ok(())
}
}
#[cfg(feature = "web-client")]
struct ManagerConfigPatchPersistence<F, H>
where
F: InstanceFactory,
H: CoreInstanceHost,
{
manager: Arc<InstanceManager<F>>,
storage: Arc<dyn ConfigFileStorage>,
_host: std::marker::PhantomData<fn() -> H>,
}
#[cfg(feature = "web-client")]
async fn persist_config_patch(
storage: &dyn ConfigFileStorage,
control: &crate::instance::manager::ConfigFileControl,
config: &TomlConfig,
) -> anyhow::Result<()> {
if control.is_read_only() {
anyhow::bail!("configuration file is read-only");
}
let Some(path) = control.path.as_deref() else {
return Ok(());
};
if storage.inspect(path).await.is_read_only() {
anyhow::bail!(
"config file {} is read-only, cannot be overwritten",
path.display()
);
}
storage.write(path, config.dump().as_bytes()).await
}
#[async_trait::async_trait]
#[cfg(feature = "web-client")]
impl<F, H> ConfigPatchPersistence for ManagerConfigPatchPersistence<F, H>
where
F: InstanceFactory<Instance = CoreInstance<H>>,
H: CoreInstanceHost,
{
async fn persist(&self, instance_id: uuid::Uuid, config: &TomlConfig) -> anyhow::Result<()> {
let control = self
.manager
.config_control(instance_id)
.ok_or_else(|| anyhow::anyhow!("configuration file control is unavailable"))?;
persist_config_patch(self.storage.as_ref(), &control, config).await
}
}
#[cfg(all(test, feature = "web-client"))]
mod config_patch_persistence_tests {
use std::{
path::{Path, PathBuf},
sync::atomic::{AtomicBool, AtomicUsize, Ordering},
};
use super::*;
use crate::{
instance::manager::{ConfigFileControl, ConfigFilePermission},
management::ConfigFileStorage,
};
#[derive(Default)]
struct RecordingStorage {
read_only: AtomicBool,
inspections: AtomicUsize,
writes: AtomicUsize,
}
#[async_trait::async_trait]
impl ConfigFileStorage for RecordingStorage {
async fn inspect(&self, path: &Path) -> ConfigFileControl {
self.inspections.fetch_add(1, Ordering::Relaxed);
let permission = if self.read_only.load(Ordering::Relaxed) {
ConfigFilePermission::from(ConfigFilePermission::READ_ONLY)
} else {
ConfigFilePermission::default()
};
ConfigFileControl::new(Some(path.to_owned()), permission)
}
async fn read(&self, _path: &Path) -> anyhow::Result<Option<Vec<u8>>> {
unreachable!("config patch persistence does not read files")
}
async fn write(&self, _path: &Path, _contents: &[u8]) -> anyhow::Result<()> {
self.writes.fetch_add(1, Ordering::Relaxed);
Ok(())
}
async fn remove(&self, _path: &Path) -> anyhow::Result<()> {
unreachable!("config patch persistence does not remove files")
}
}
#[tokio::test]
async fn pathless_config_patch_skips_persistence() {
let storage = RecordingStorage::default();
let control = ConfigFileControl::new(None, ConfigFilePermission::default());
persist_config_patch(&storage, &control, &TomlConfig::default())
.await
.unwrap();
assert_eq!(storage.inspections.load(Ordering::Relaxed), 0);
assert_eq!(storage.writes.load(Ordering::Relaxed), 0);
}
#[tokio::test]
async fn config_patch_rechecks_file_permission_before_write() {
let storage = RecordingStorage::default();
storage.read_only.store(true, Ordering::Relaxed);
let control = ConfigFileControl::new(
Some(PathBuf::from("managed.toml")),
ConfigFilePermission::default(),
);
let error = persist_config_patch(&storage, &control, &TomlConfig::default())
.await
.unwrap_err();
assert!(error.to_string().contains("managed.toml is read-only"));
assert_eq!(storage.inspections.load(Ordering::Relaxed), 1);
assert_eq!(storage.writes.load(Ordering::Relaxed), 0);
} }
} }
+3 -3
View File
@@ -34,9 +34,9 @@ pub(crate) use full::register_web_client_rpc;
pub use full::remote_client; pub use full::remote_client;
#[cfg(feature = "web-client")] #[cfg(feature = "web-client")]
pub use full::{ pub use full::{
ConfigFileStorage, ConfigServerEndpoint, InstanceMutationHooks, InstanceMutationResult, ConfigFileStorage, ConfigPatchPersistence, ConfigServerEndpoint, InstanceMutationHooks,
ProcessManagement, ProcessManagementRpc, UnsupportedConfigFileStorage, WebClient, InstanceMutationResult, ProcessManagement, ProcessManagementRpc, UnsupportedConfigFileStorage,
WebClientConfig, apply_config_patch, config_source_from_rpc, config_source_to_rpc, WebClient, WebClientConfig, apply_config_patch, config_source_from_rpc, config_source_to_rpc,
network_instance_running_info, network_instance_running_info,
}; };
#[cfg(feature = "management")] #[cfg(feature = "management")]
+11 -10
View File
@@ -189,16 +189,15 @@ impl AttachedPeerRuntime {
let runtime_handle = Handle::current(); let runtime_handle = Handle::current();
let network = network_runtime_config.snapshot(); let network = network_runtime_config.snapshot();
let (peer_snapshot, credential_public_key) = build_peer_snapshot(&network, &config)?; let (peer_snapshot, credential_public_key) = build_peer_snapshot(&network, &config)?;
let credential_registration = credential_public_key let credential_registration = match credential_public_key {
.map(|public_key| { Some(public_key) => Some(AttachedCredentialRegistration::register(
AttachedCredentialRegistration::register( network_peer_manager.clone(),
network_peer_manager.clone(), network_runtime_config.clone(),
network_runtime_config.clone(), public_key,
public_key, config.groups.clone(),
config.groups.clone(), )?),
) None => None,
}) };
.transpose()?;
let services = build_attached_services(&network.services, credential_public_key.is_some()); let services = build_attached_services(&network.services, credential_public_key.is_some());
let runtime_config = CoreRuntimeConfigStore::new(services, Arc::new(peer_snapshot.clone())); let runtime_config = CoreRuntimeConfigStore::new(services, Arc::new(peer_snapshot.clone()));
let (packet_sender, packet_receiver) = host_packet_channel(); let (packet_sender, packet_receiver) = host_packet_channel();
@@ -212,6 +211,7 @@ impl AttachedPeerRuntime {
exit_nodes: Vec::new(), exit_nodes: Vec::new(),
foreign_context_default_flags: flags, foreign_context_default_flags: flags,
}, },
Vec::new(),
runtime_config, runtime_config,
Arc::new(()), Arc::new(()),
packet_sender, packet_sender,
@@ -623,6 +623,7 @@ mod tests {
let peer_manager = Arc::new( let peer_manager = Arc::new(
PeerManagerCore::new( PeerManagerCore::new(
portable, portable,
Vec::new(),
store.clone(), store.clone(),
Arc::new(()), Arc::new(()),
packet_sender, packet_sender,
@@ -372,7 +372,11 @@ mod tests {
); );
} }
#[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[cfg_attr(
not(target_os = "wasi"),
tokio::test(flavor = "multi_thread", worker_threads = 2)
)]
#[cfg_attr(target_os = "wasi", tokio::test)]
async fn echoed_business_traffic_keeps_connection_alive_when_pongs_are_lost() { async fn echoed_business_traffic_keeps_connection_alive_when_pongs_are_lost() {
let local_liveness = PeerConnLiveness::new(); let local_liveness = PeerConnLiveness::new();
let remote_liveness = PeerConnLiveness::new(); let remote_liveness = PeerConnLiveness::new();
+510 -179
View File
@@ -1,5 +1,5 @@
use std::{ use std::{
collections::HashMap, collections::{HashMap, HashSet},
sync::{Arc, Mutex}, sync::{Arc, Mutex},
time::{Duration, SystemTime, UNIX_EPOCH}, time::{Duration, SystemTime, UNIX_EPOCH},
}; };
@@ -9,7 +9,10 @@ use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256}; use sha2::{Digest, Sha256};
use x25519_dalek::{PublicKey, StaticSecret}; use x25519_dalek::{PublicKey, StaticSecret};
use crate::proto::peer_rpc::{TrustedCredentialPubkey, TrustedCredentialPubkeyProof}; use crate::{
config::toml::ManagedCredentialConfig,
proto::peer_rpc::{TrustedCredentialPubkey, TrustedCredentialPubkeyProof},
};
fn default_true() -> bool { fn default_true() -> bool {
true true
@@ -43,7 +46,7 @@ pub struct CredentialUpsertOptions {
pub reusable: bool, pub reusable: bool,
} }
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct CredentialEntry { pub(crate) struct CredentialEntry {
pubkey: String, pubkey: String,
#[serde(default)] #[serde(default)]
@@ -85,6 +88,33 @@ impl CredentialEntry {
.unwrap_or_default(), .unwrap_or_default(),
} }
} }
fn from_managed(entry: &ManagedCredentialConfig) -> Result<Self, String> {
let credential_id = entry.credential_id.trim();
let private_bytes: [u8; 32] = BASE64_STANDARD
.decode(entry.credential_secret.trim())
.map_err(|_| format!("credential_secret for {credential_id} must be base64"))?
.try_into()
.map_err(|_| format!("credential_secret for {credential_id} must contain 32 bytes"))?;
let private = StaticSecret::from(private_bytes);
let mut allowed_proxy_cidrs = Vec::with_capacity(entry.allowed_proxy_cidrs.len());
for cidr in &entry.allowed_proxy_cidrs {
let cidr = cidr.trim();
cidr.parse::<cidr::IpCidr>()
.map_err(|_| format!("invalid allowed_proxy_cidr for {credential_id}: {cidr}"))?;
allowed_proxy_cidrs.push(cidr.to_owned());
}
Ok(Self {
pubkey: BASE64_STANDARD.encode(PublicKey::from(&private).as_bytes()),
secret: BASE64_STANDARD.encode(private.as_bytes()),
groups: entry.groups.clone(),
allow_relay: entry.allow_relay,
allowed_proxy_cidrs,
reusable: entry.reusable,
expiry_unix: entry.expiry_unix,
created_at_unix: 0,
})
}
} }
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
@@ -114,11 +144,35 @@ pub trait CredentialStorage: Send + Sync + 'static {
fn store(&self, serialized_credentials: &str) -> anyhow::Result<()>; fn store(&self, serialized_credentials: &str) -> anyhow::Result<()>;
} }
#[derive(Default)]
struct CredentialState {
base: HashMap<String, CredentialEntry>,
managed: HashMap<String, CredentialEntry>,
pending_managed: Option<HashMap<String, CredentialEntry>>,
ephemeral: HashMap<uuid::Uuid, CredentialEntry>,
}
pub(crate) struct CredentialManager { pub(crate) struct CredentialManager {
credentials: Mutex<HashMap<String, CredentialEntry>>, state: Mutex<CredentialState>,
ephemeral_credentials: Mutex<HashMap<uuid::Uuid, CredentialEntry>>,
storage: Option<Arc<dyn CredentialStorage>>, storage: Option<Arc<dyn CredentialStorage>>,
storage_write: Mutex<()>, storage_load_error: Option<String>,
}
/// A validated managed credential replacement awaiting installation.
#[cfg(feature = "web-client")]
pub(crate) struct ManagedCredentialReplacement<'a> {
manager: &'a CredentialManager,
changed: bool,
installed: bool,
}
#[cfg(feature = "web-client")]
impl Drop for ManagedCredentialReplacement<'_> {
fn drop(&mut self) {
if self.changed && !self.installed {
self.manager.state.lock().unwrap().pending_managed = None;
}
}
} }
impl Default for CredentialManager { impl Default for CredentialManager {
@@ -130,38 +184,35 @@ impl Default for CredentialManager {
impl CredentialManager { impl CredentialManager {
pub fn new() -> Self { pub fn new() -> Self {
Self { Self {
credentials: Mutex::new(HashMap::new()), state: Mutex::new(CredentialState::default()),
ephemeral_credentials: Mutex::new(HashMap::new()),
storage: None, storage: None,
storage_write: Mutex::new(()), storage_load_error: None,
} }
} }
pub fn from_storage(storage: Arc<dyn CredentialStorage>) -> Self { pub fn from_storage(storage: Arc<dyn CredentialStorage>) -> Self {
let credentials = match storage.load() { let loaded = match storage.load() {
Ok(Some(serialized)) => serde_json::from_str(&serialized).unwrap_or_else(|error| { Ok(Some(serialized)) => serde_json::from_str(&serialized).map_err(anyhow::Error::from),
tracing::warn!(?error, "failed to parse stored credentials"); Ok(None) => Ok(HashMap::new()),
HashMap::new() Err(error) => Err(error),
}), };
Ok(None) => HashMap::new(), let (base, storage_load_error) = match loaded {
Ok(base) => (base, None),
Err(error) => { Err(error) => {
tracing::warn!(?error, "failed to load stored credentials"); tracing::error!(?error, "credential storage is unavailable");
HashMap::new() (HashMap::new(), Some(error.to_string()))
} }
}; };
Self { Self {
credentials: Mutex::new(credentials), state: Mutex::new(CredentialState {
ephemeral_credentials: Mutex::new(HashMap::new()), base,
..Default::default()
}),
storage: Some(storage), storage: Some(storage),
storage_write: Mutex::new(()), storage_load_error,
} }
} }
pub fn with_entries<R>(&self, f: impl FnOnce(&HashMap<String, CredentialEntry>) -> R) -> R {
let credentials = self.credentials.lock().unwrap();
f(&credentials)
}
pub fn generate_credential_with_options( pub fn generate_credential_with_options(
&self, &self,
groups: Vec<String>, groups: Vec<String>,
@@ -170,61 +221,68 @@ impl CredentialManager {
ttl: Duration, ttl: Duration,
credential_id: Option<String>, credential_id: Option<String>,
reusable: bool, reusable: bool,
) -> GeneratedCredential { ) -> Result<GeneratedCredential, String> {
self.remove_expired_credentials(); self.ensure_storage_available()
self.generate_credential_with_options_after_cleanup( .map_err(|error| error.to_string())?;
groups, let mut state = self.state.lock().unwrap();
allow_relay, let now = current_unix_timestamp();
allowed_proxy_cidrs, let mut updated = state.base.clone();
ttl, updated.retain(|_, entry| entry.is_active_at(now));
credential_id, let id = if let Some(id) = credential_id
reusable, .map(|x| x.trim().to_string())
) .filter(|x| !x.is_empty())
} {
if Self::managed_contains_id(&state, &id) {
pub fn generate_credential_with_options_after_cleanup( return Err(format!("credential_id {id} is managed by configuration"));
&self, }
groups: Vec<String>, if let Some(existing) = updated.get(&id)
allow_relay: bool, && !existing.secret.is_empty()
allowed_proxy_cidrs: Vec<String>,
ttl: Duration,
credential_id: Option<String>,
reusable: bool,
) -> GeneratedCredential {
let generated = {
let mut credentials = self.credentials.lock().unwrap();
let id = if let Some(id) = credential_id
.map(|x| x.trim().to_string())
.filter(|x| !x.is_empty())
{ {
if let Some(existing) = credentials.get(&id) return Ok(GeneratedCredential {
&& !existing.secret.is_empty() credential_id: id,
{ secret: existing.secret.clone(),
return GeneratedCredential { expiry_unix: existing.expiry_unix,
credential_id: id, changed: false,
secret: existing.secret.clone(), });
expiry_unix: existing.expiry_unix, }
changed: false, id
}; } else {
loop {
let id = uuid::Uuid::new_v4().to_string();
if !updated.contains_key(&id) && !Self::managed_contains_id(&state, &id) {
break id;
} }
id
} else {
uuid::Uuid::new_v4().to_string()
};
let (entry, secret) =
Self::build_entry(groups, allow_relay, allowed_proxy_cidrs, reusable, ttl);
let expiry_unix = entry.expiry_unix;
credentials.insert(id.clone(), entry);
GeneratedCredential {
credential_id: id,
secret,
expiry_unix,
changed: true,
} }
}; };
self.persist();
generated let (entry, secret) = loop {
let generated = Self::build_entry(
groups.clone(),
allow_relay,
allowed_proxy_cidrs.clone(),
reusable,
ttl,
);
let public_key_in_use = updated
.values()
.chain(Self::managed_values(&state))
.chain(state.ephemeral.values())
.any(|existing| existing.pubkey == generated.0.pubkey);
if !public_key_in_use {
break generated;
}
};
let expiry_unix = entry.expiry_unix;
updated.insert(id.clone(), entry);
self.store_base(&updated)
.map_err(|error| format!("failed to store credentials: {error}"))?;
state.base = updated;
Ok(GeneratedCredential {
credential_id: id,
secret,
expiry_unix,
changed: true,
})
} }
fn build_entry( fn build_entry(
@@ -258,17 +316,19 @@ impl CredentialManager {
(entry, secret) (entry, secret)
} }
pub fn revoke_credential(&self, credential_id: &str) -> bool { pub fn revoke_credential(&self, credential_id: &str) -> Result<bool, String> {
let removed = self self.ensure_storage_available()
.credentials .map_err(|error| error.to_string())?;
.lock() let mut state = self.state.lock().unwrap();
.unwrap() if !state.base.contains_key(credential_id) {
.remove(credential_id) return Ok(false);
.is_some();
if removed {
self.persist();
} }
removed let mut updated = state.base.clone();
updated.remove(credential_id);
self.store_base(&updated)
.map_err(|error| format!("failed to store credentials: {error}"))?;
state.base = updated;
Ok(true)
} }
pub fn register_ephemeral_credential( pub fn register_ephemeral_credential(
@@ -290,17 +350,14 @@ impl CredentialManager {
created_at_unix: current_unix_timestamp(), created_at_unix: current_unix_timestamp(),
}; };
let _storage_write = self.storage_write.lock().unwrap(); let mut state = self.state.lock().unwrap();
if self if state
.credentials .base
.lock()
.unwrap()
.values() .values()
.chain(Self::managed_values(&state))
.any(|existing| existing.pubkey == entry.pubkey) .any(|existing| existing.pubkey == entry.pubkey)
|| self || state
.ephemeral_credentials .ephemeral
.lock()
.unwrap()
.values() .values()
.any(|existing| existing.pubkey == entry.pubkey) .any(|existing| existing.pubkey == entry.pubkey)
{ {
@@ -308,10 +365,7 @@ impl CredentialManager {
} }
let credential_id = uuid::Uuid::new_v4(); let credential_id = uuid::Uuid::new_v4();
self.ephemeral_credentials state.ephemeral.insert(credential_id, entry);
.lock()
.unwrap()
.insert(credential_id, entry);
Ok(credential_id) Ok(credential_id)
} }
@@ -320,8 +374,8 @@ impl CredentialManager {
credential_id: uuid::Uuid, credential_id: uuid::Uuid,
groups: Vec<String>, groups: Vec<String>,
) -> Option<bool> { ) -> Option<bool> {
let mut credentials = self.ephemeral_credentials.lock().unwrap(); let mut state = self.state.lock().unwrap();
let credential = credentials.get_mut(&credential_id)?; let credential = state.ephemeral.get_mut(&credential_id)?;
if credential.groups == groups { if credential.groups == groups {
return Some(false); return Some(false);
} }
@@ -330,9 +384,10 @@ impl CredentialManager {
} }
pub fn revoke_ephemeral_credential(&self, credential_id: uuid::Uuid) -> bool { pub fn revoke_ephemeral_credential(&self, credential_id: uuid::Uuid) -> bool {
self.ephemeral_credentials self.state
.lock() .lock()
.unwrap() .unwrap()
.ephemeral
.remove(&credential_id) .remove(&credential_id)
.is_some() .is_some()
} }
@@ -372,23 +427,32 @@ impl CredentialManager {
created_at_unix: current_unix_timestamp(), created_at_unix: current_unix_timestamp(),
}; };
let _storage_write = self.storage_write.lock().unwrap(); self.ensure_storage_available()
let mut credentials = self.credentials.lock().unwrap(); .map_err(|error| error.to_string())?;
if credentials.iter().any(|(existing_id, existing)| { let mut state = self.state.lock().unwrap();
existing_id != &credential_id && existing.pubkey == entry.pubkey if Self::managed_contains_id(&state, &credential_id) {
}) { return Err(format!(
"credential_id {credential_id} is managed by configuration"
));
}
if state
.base
.iter()
.chain(Self::managed_entries(&state))
.any(|(existing_id, existing)| {
existing_id != &credential_id && existing.pubkey == entry.pubkey
})
{
return Err("credential_secret is already used by another credential_id".to_string()); return Err("credential_secret is already used by another credential_id".to_string());
} }
if self if state
.ephemeral_credentials .ephemeral
.lock()
.unwrap()
.values() .values()
.any(|existing| existing.pubkey == entry.pubkey) .any(|existing| existing.pubkey == entry.pubkey)
{ {
return Err("credential public key is already registered".to_owned()); return Err("credential public key is already registered".to_owned());
} }
let changed = credentials.get(&credential_id).is_none_or(|existing| { let changed = state.base.get(&credential_id).is_none_or(|existing| {
existing.secret != entry.secret existing.secret != entry.secret
|| existing.pubkey != entry.pubkey || existing.pubkey != entry.pubkey
|| existing.groups != entry.groups || existing.groups != entry.groups
@@ -401,18 +465,11 @@ impl CredentialManager {
return Ok(false); return Ok(false);
} }
if let Some(storage) = &self.storage { let mut updated = state.base.clone();
let mut updated = credentials.clone(); updated.insert(credential_id, entry);
updated.insert(credential_id, entry); self.store_base(&updated)
let serialized = serde_json::to_string_pretty(&updated) .map_err(|error| format!("failed to store credentials: {error}"))?;
.map_err(|error| format!("failed to serialize credentials: {error}"))?; state.base = updated;
storage
.store(&serialized)
.map_err(|error| format!("failed to store credentials: {error}"))?;
*credentials = updated;
} else {
credentials.insert(credential_id, entry);
}
Ok(true) Ok(true)
} }
@@ -421,15 +478,18 @@ impl CredentialManager {
} }
fn remove_expired_credentials_at(&self, now: i64) -> bool { fn remove_expired_credentials_at(&self, now: i64) -> bool {
let mut credentials = self.credentials.lock().unwrap(); let mut state = self.state.lock().unwrap();
let before = credentials.len(); let mut updated = state.base.clone();
credentials.retain(|_, entry| entry.is_active_at(now)); updated.retain(|_, entry| entry.is_active_at(now));
let changed = before != credentials.len(); if updated == state.base {
drop(credentials); return false;
if changed {
self.persist();
} }
changed if let Err(error) = self.store_base(&updated) {
tracing::warn!(?error, "failed to remove expired credentials");
return false;
}
state.base = updated;
true
} }
pub fn get_trusted_pubkeys(&self, network_secret: &str) -> Vec<TrustedCredentialPubkeyProof> { pub fn get_trusted_pubkeys(&self, network_secret: &str) -> Vec<TrustedCredentialPubkeyProof> {
@@ -439,36 +499,29 @@ impl CredentialManager {
TrustedCredentialPubkeyProof::new_signed(credential, network_secret) TrustedCredentialPubkeyProof::new_signed(credential, network_secret)
}) })
}; };
let mut trusted = self let state = self.state.lock().unwrap();
.credentials let mut trusted = state
.lock() .base
.unwrap()
.values() .values()
.chain(state.managed.values())
.filter(|entry| entry.is_active_at(now)) .filter(|entry| entry.is_active_at(now))
.filter_map(to_proof) .filter_map(to_proof)
.collect::<Vec<_>>(); .collect::<Vec<_>>();
trusted.extend( trusted.extend(state.ephemeral.values().filter_map(to_proof));
self.ephemeral_credentials
.lock()
.unwrap()
.values()
.filter_map(to_proof),
);
trusted trusted
} }
pub fn is_pubkey_trusted(&self, pubkey: &[u8]) -> bool { pub fn is_pubkey_trusted(&self, pubkey: &[u8]) -> bool {
let now = current_unix_timestamp(); let now = current_unix_timestamp();
let encoded = BASE64_STANDARD.encode(pubkey); let encoded = BASE64_STANDARD.encode(pubkey);
self.credentials let state = self.state.lock().unwrap();
.lock() state
.unwrap() .base
.values() .values()
.chain(state.managed.values())
.any(|entry| entry.pubkey == encoded && entry.is_active_at(now)) .any(|entry| entry.pubkey == encoded && entry.is_active_at(now))
|| self || state
.ephemeral_credentials .ephemeral
.lock()
.unwrap()
.values() .values()
.any(|entry| entry.pubkey == encoded) .any(|entry| entry.pubkey == encoded)
} }
@@ -476,13 +529,136 @@ impl CredentialManager {
pub fn list_credentials(&self) -> Vec<CredentialInfo> { pub fn list_credentials(&self) -> Vec<CredentialInfo> {
let now = current_unix_timestamp(); let now = current_unix_timestamp();
self.credentials let state = self.state.lock().unwrap();
.lock() let mut credentials = state
.unwrap() .base
.iter() .iter()
.chain(state.managed.iter())
.filter(|(_, entry)| entry.is_active_at(now)) .filter(|(_, entry)| entry.is_active_at(now))
.map(|(id, entry)| entry.to_credential_info(id)) .map(|(id, entry)| entry.to_credential_info(id))
.collect() .collect::<Vec<_>>();
credentials.sort_unstable_by(|left, right| left.credential_id.cmp(&right.credential_id));
credentials
}
pub fn install_initial_managed_credentials(
&self,
credentials: &[ManagedCredentialConfig],
) -> Result<(), String> {
self.ensure_storage_available()
.map_err(|error| error.to_string())?;
let replacement = Self::build_managed_entries(credentials)?;
let mut state = self.state.lock().unwrap();
Self::validate_managed_conflicts(&state, &replacement)?;
state.managed = replacement;
Ok(())
}
/// Fallible checks for a managed credential replacement (secret parsing,
/// duplicate IDs/keys, conflicts with base/ephemeral credentials). Must
/// run before the candidate is persisted so a rejected patch never
/// reaches disk.
#[cfg(feature = "web-client")]
pub fn validate_managed_credentials(
&self,
credentials: &[ManagedCredentialConfig],
) -> Result<ManagedCredentialReplacement<'_>, String> {
let replacement = Self::build_managed_entries(credentials)?;
let mut state = self.state.lock().unwrap();
if state.pending_managed.is_some() {
return Err("managed credential replacement is already pending".to_owned());
}
Self::validate_managed_conflicts(&state, &replacement)?;
let changed = state.managed != replacement;
if changed {
state.pending_managed = Some(replacement);
}
Ok(ManagedCredentialReplacement {
manager: self,
changed,
installed: false,
})
}
/// Installs a replacement whose IDs and public keys were reserved by
/// [`Self::validate_managed_credentials`].
#[cfg(feature = "web-client")]
pub fn install_managed_credentials(mut replacement: ManagedCredentialReplacement<'_>) -> bool {
if !replacement.changed {
return false;
}
let mut state = replacement.manager.state.lock().unwrap();
state.managed = state
.pending_managed
.take()
.expect("validated managed credential replacement must remain reserved");
replacement.installed = true;
true
}
fn managed_contains_id(state: &CredentialState, credential_id: &str) -> bool {
state.managed.contains_key(credential_id)
|| state
.pending_managed
.as_ref()
.is_some_and(|pending| pending.contains_key(credential_id))
}
fn managed_entries(
state: &CredentialState,
) -> impl Iterator<Item = (&String, &CredentialEntry)> {
state.managed.iter().chain(
state
.pending_managed
.iter()
.flat_map(|pending| pending.iter()),
)
}
fn managed_values(state: &CredentialState) -> impl Iterator<Item = &CredentialEntry> {
Self::managed_entries(state).map(|(_, entry)| entry)
}
fn build_managed_entries(
credentials: &[ManagedCredentialConfig],
) -> Result<HashMap<String, CredentialEntry>, String> {
let mut entries = HashMap::with_capacity(credentials.len());
let mut public_keys = HashSet::with_capacity(credentials.len());
for credential in credentials {
let credential_id = credential.credential_id.trim().to_owned();
if credential_id.is_empty() {
return Err("credential_id must not be empty".to_owned());
}
let entry = CredentialEntry::from_managed(credential)?;
if !public_keys.insert(entry.pubkey.clone()) {
return Err("credential_secret is assigned to multiple credential IDs".to_owned());
}
if entries.insert(credential_id.clone(), entry).is_some() {
return Err(format!("duplicate managed credential_id: {credential_id}"));
}
}
Ok(entries)
}
fn validate_managed_conflicts(
state: &CredentialState,
replacement: &HashMap<String, CredentialEntry>,
) -> Result<(), String> {
if let Some(credential_id) = replacement.keys().find(|id| state.base.contains_key(*id)) {
return Err(format!(
"credential_id {credential_id} is already owned by the credential file"
));
}
if replacement.values().any(|entry| {
state
.base
.values()
.chain(state.ephemeral.values())
.any(|existing| existing.pubkey == entry.pubkey)
}) {
return Err("credential public key is already registered".to_owned());
}
Ok(())
} }
fn decode_pubkey_b64(s: &str) -> Option<Vec<u8>> { fn decode_pubkey_b64(s: &str) -> Option<Vec<u8>> {
@@ -503,21 +679,19 @@ impl CredentialManager {
) )
} }
fn persist(&self) { fn ensure_storage_available(&self) -> anyhow::Result<()> {
let Some(storage) = &self.storage else { if let Some(error) = &self.storage_load_error {
return; anyhow::bail!("credential storage is unavailable: {error}");
};
let _storage_write = self.storage_write.lock().unwrap();
let serialized = match self.with_entries(serde_json::to_string_pretty) {
Ok(serialized) => serialized,
Err(error) => {
tracing::warn!(?error, "failed to serialize credentials");
return;
}
};
if let Err(error) = storage.store(&serialized) {
tracing::warn!(?error, "failed to store credentials");
} }
Ok(())
}
fn store_base(&self, base: &HashMap<String, CredentialEntry>) -> anyhow::Result<()> {
self.ensure_storage_available()?;
let Some(storage) = &self.storage else {
return Ok(());
};
storage.store(&serde_json::to_string_pretty(base)?)
} }
} }
@@ -525,6 +699,32 @@ impl CredentialManager {
mod tests { mod tests {
use super::*; use super::*;
fn managed_credential(
credential_id: &str,
secret_byte: u8,
expiry_unix: i64,
) -> ManagedCredentialConfig {
ManagedCredentialConfig {
credential_id: credential_id.to_owned(),
credential_secret: BASE64_STANDARD.encode([secret_byte; 32]),
groups: vec!["ops".to_owned()],
allow_relay: false,
allowed_proxy_cidrs: vec!["10.0.0.0/24".to_owned()],
expiry_unix,
reusable: true,
}
}
#[test]
fn managed_credential_trims_allowed_proxy_cidrs() {
let mut credential = managed_credential("managed", 1, i64::MAX);
credential.allowed_proxy_cidrs = vec![" 10.0.0.0/24 ".to_owned()];
let entry = CredentialEntry::from_managed(&credential).unwrap();
assert_eq!(entry.allowed_proxy_cidrs, ["10.0.0.0/24"]);
}
impl CredentialManager { impl CredentialManager {
pub(crate) fn generate_credential( pub(crate) fn generate_credential(
&self, &self,
@@ -541,6 +741,7 @@ mod tests {
None, None,
true, true,
) )
.unwrap()
} }
fn generate_credential_with_id( fn generate_credential_with_id(
@@ -559,6 +760,7 @@ mod tests {
credential_id, credential_id,
true, true,
) )
.unwrap()
} }
} }
@@ -641,7 +843,7 @@ mod tests {
); );
assert_eq!(trusted[0].credential.as_ref().unwrap().reusable, Some(true)); assert_eq!(trusted[0].credential.as_ref().unwrap().reusable, Some(true));
assert!(mgr.revoke_credential(&generated.credential_id)); assert!(mgr.revoke_credential(&generated.credential_id).unwrap());
assert!(!mgr.is_pubkey_trusted(&pubkey_bytes)); assert!(!mgr.is_pubkey_trusted(&pubkey_bytes));
assert!(mgr.get_trusted_pubkeys("sec").is_empty()); assert!(mgr.get_trusted_pubkeys("sec").is_empty());
} }
@@ -684,14 +886,16 @@ mod tests {
#[test] #[test]
fn upsert_credential_preserves_key_attributes_and_storage() { fn upsert_credential_preserves_key_attributes_and_storage() {
let source = CredentialManager::new(); let source = CredentialManager::new();
let generated = source.generate_credential_with_options( let generated = source
vec!["users".to_string()], .generate_credential_with_options(
false, vec!["users".to_string()],
vec!["10.0.0.0/8".to_string()], false,
Duration::from_secs(3600), vec!["10.0.0.0/8".to_string()],
Some("shared-id".to_string()), Duration::from_secs(3600),
false, Some("shared-id".to_string()),
); false,
)
.unwrap();
let source_info = source.list_credentials().remove(0); let source_info = source.list_credentials().remove(0);
let options = CredentialUpsertOptions { let options = CredentialUpsertOptions {
credential_id: generated.credential_id, credential_id: generated.credential_id,
@@ -802,13 +1006,13 @@ mod tests {
generated.credential_id generated.credential_id
); );
assert!(manager.revoke_credential(&generated.credential_id)); assert!(manager.revoke_credential(&generated.credential_id).unwrap());
let reloaded = CredentialManager::from_storage(storage); let reloaded = CredentialManager::from_storage(storage);
assert!(reloaded.list_credentials().is_empty()); assert!(reloaded.list_credentials().is_empty());
} }
#[test] #[test]
fn malformed_storage_starts_with_empty_credentials() { fn malformed_storage_is_fail_closed() {
let storage = Arc::new(MemoryCredentialStorage { let storage = Arc::new(MemoryCredentialStorage {
serialized: Mutex::new(Some("not json".to_owned())), serialized: Mutex::new(Some("not json".to_owned())),
}); });
@@ -816,6 +1020,7 @@ mod tests {
let manager = CredentialManager::from_storage(storage); let manager = CredentialManager::from_storage(storage);
assert!(manager.list_credentials().is_empty()); assert!(manager.list_credentials().is_empty());
assert!(manager.install_initial_managed_credentials(&[]).is_err());
} }
#[test] #[test]
@@ -846,4 +1051,130 @@ mod tests {
assert!(manager.get_trusted_pubkeys("network-secret").is_empty()); assert!(manager.get_trusted_pubkeys("network-secret").is_empty());
assert!(storage.serialized.lock().unwrap().is_none()); assert!(storage.serialized.lock().unwrap().is_none());
} }
#[cfg(feature = "web-client")]
#[test]
fn managed_credentials_work_without_base_storage_and_expire_in_place() {
let manager = CredentialManager::new();
let active = managed_credential("active", 1, current_unix_timestamp() + 60);
let expired = managed_credential("expired", 2, current_unix_timestamp() - 1);
manager
.install_initial_managed_credentials(&[active.clone(), expired])
.unwrap();
assert_eq!(manager.list_credentials().len(), 1);
assert_eq!(manager.list_credentials()[0].credential_id, "active");
let private_bytes: [u8; 32] = BASE64_STANDARD
.decode(active.credential_secret)
.unwrap()
.try_into()
.unwrap();
let public = PublicKey::from(&StaticSecret::from(private_bytes));
assert!(manager.is_pubkey_trusted(public.as_bytes()));
let replacement = manager.validate_managed_credentials(&[]).unwrap();
assert!(replacement.changed);
assert!(CredentialManager::install_managed_credentials(replacement));
assert!(manager.list_credentials().is_empty());
}
#[cfg(feature = "web-client")]
#[test]
fn pending_managed_replacement_reserves_ids_and_public_keys() {
let manager = CredentialManager::new();
let pending = managed_credential("pending", 5, current_unix_timestamp() + 60);
let private_bytes: [u8; 32] = BASE64_STANDARD
.decode(&pending.credential_secret)
.unwrap()
.try_into()
.unwrap();
let public = PublicKey::from(&StaticSecret::from(private_bytes));
let replacement = manager
.validate_managed_credentials(std::slice::from_ref(&pending))
.unwrap();
let error = manager
.generate_credential_with_options(
Vec::new(),
false,
Vec::new(),
Duration::from_secs(60),
Some("pending".to_owned()),
true,
)
.unwrap_err();
assert!(error.contains("managed by configuration"));
assert!(
manager
.register_ephemeral_credential(
*public.as_bytes(),
Vec::new(),
false,
Vec::new(),
false,
)
.is_err()
);
assert!(!manager.is_pubkey_trusted(public.as_bytes()));
drop(replacement);
assert!(
manager
.register_ephemeral_credential(
*public.as_bytes(),
Vec::new(),
false,
Vec::new(),
false,
)
.is_ok()
);
}
#[cfg(feature = "web-client")]
#[test]
fn managed_and_base_credentials_must_be_disjoint() {
let manager = CredentialManager::new();
manager
.install_initial_managed_credentials(&[managed_credential(
"managed",
3,
current_unix_timestamp() + 60,
)])
.unwrap();
let error = manager
.generate_credential_with_options(
Vec::new(),
false,
Vec::new(),
Duration::from_secs(60),
Some("managed".to_owned()),
true,
)
.unwrap_err();
assert!(error.contains("managed by configuration"));
let generated =
manager.generate_credential(Vec::new(), false, Vec::new(), Duration::from_secs(60));
let conflicting = ManagedCredentialConfig {
credential_id: "other".to_owned(),
credential_secret: generated.secret,
..managed_credential("other", 4, current_unix_timestamp() + 60)
};
let error = manager
.validate_managed_credentials(&[conflicting])
.err()
.unwrap();
assert_eq!(error, "credential public key is already registered");
// The rejected replacement must not have touched existing state.
assert!(
manager
.list_credentials()
.iter()
.any(|info| info.credential_id == "managed")
);
}
} }
@@ -811,7 +811,7 @@ impl ForeignNetworkManager {
let tasks = Arc::new(std::sync::Mutex::new(JoinSet::new())); let tasks = Arc::new(std::sync::Mutex::new(JoinSet::new()));
let task_reaper = tokio::spawn(reap_joinset_background( let task_reaper = tokio::spawn(reap_joinset_background(
tasks.clone(), Arc::downgrade(&tasks),
"ForeignNetworkManager", "ForeignNetworkManager",
)); ));
+9
View File
@@ -28,6 +28,7 @@ use crate::{
PeerRuntimeSnapshot, PeerRuntimeSnapshot,
}, },
runtime::{CoreInstanceRuntimeConfig, CoreRuntimeConfigStore}, runtime::{CoreInstanceRuntimeConfig, CoreRuntimeConfigStore},
toml::ManagedCredentialConfig,
}, },
events::CoreEventSink, events::CoreEventSink,
foundation::task::ExternalTaskSignal, foundation::task::ExternalTaskSignal,
@@ -811,6 +812,7 @@ impl PeerManagerCore {
#[allow(clippy::too_many_arguments)] #[allow(clippy::too_many_arguments)]
pub(crate) fn new( pub(crate) fn new(
mut config: PortablePeerManagerConfig, mut config: PortablePeerManagerConfig,
managed_credentials: Vec<ManagedCredentialConfig>,
runtime_config: CoreRuntimeConfigStore, runtime_config: CoreRuntimeConfigStore,
stun_info_source: Arc<dyn PeerStunInfoSource>, stun_info_source: Arc<dyn PeerStunInfoSource>,
nic_channel: HostPacketSender, nic_channel: HostPacketSender,
@@ -922,6 +924,10 @@ impl PeerManagerCore {
credential_storage, credential_storage,
}, },
)); ));
context
.credential_manager()
.install_initial_managed_credentials(&managed_credentials)
.map_err(anyhow::Error::msg)?;
let peer_manager = Self::assemble( let peer_manager = Self::assemble(
config.route_algo, config.route_algo,
my_peer_id, my_peer_id,
@@ -3563,6 +3569,7 @@ mod tests {
let stun_info_source = Arc::new(RuntimeConfigStunInfoSource(runtime_config.clone())); let stun_info_source = Arc::new(RuntimeConfigStunInfoSource(runtime_config.clone()));
Self::new( Self::new(
config, config,
Vec::new(),
runtime_config, runtime_config,
stun_info_source, stun_info_source,
nic_channel, nic_channel,
@@ -3879,6 +3886,7 @@ mod tests {
let core = PeerManagerCore::new( let core = PeerManagerCore::new(
config, config,
Vec::new(),
runtime_config, runtime_config,
Arc::new(()), Arc::new(()),
packet_tx, packet_tx,
@@ -4099,6 +4107,7 @@ mod tests {
admin_a admin_a
.credential_manager() .credential_manager()
.revoke_credential(&generated.credential_id) .revoke_credential(&generated.credential_id)
.unwrap()
); );
admin_b admin_b
.context .context
+1 -1
View File
@@ -123,7 +123,7 @@ impl Server {
self.stopped.store(false, Ordering::Relaxed); self.stopped.store(false, Ordering::Relaxed);
let handler_tasks = self.handler_tasks.clone(); let handler_tasks = self.handler_tasks.clone();
self.tasks.lock().unwrap().spawn(reap_joinset_background( self.tasks.lock().unwrap().spawn(reap_joinset_background(
handler_tasks.clone(), Arc::downgrade(&handler_tasks),
"rpc server handlers", "rpc server handlers",
)); ));
+8 -1
View File
@@ -127,6 +127,7 @@ fn hosted_network_config(config: &NetworkConfig) -> NetworkConfig {
disable_upnp: config.disable_upnp, disable_upnp: config.disable_upnp,
disable_relay_data: config.disable_relay_data, disable_relay_data: config.disable_relay_data,
enable_udp_broadcast_relay: config.enable_udp_broadcast_relay, enable_udp_broadcast_relay: config.enable_udp_broadcast_relay,
managed_credentials: config.managed_credentials.clone(),
peers, peers,
..Default::default() ..Default::default()
} }
@@ -388,7 +389,7 @@ impl WasiWebClientRuntime {
mod tests { mod tests {
use super::*; use super::*;
use crate::proto::{ use crate::proto::{
api::manage::NetworkPeerConfig, api::manage::{ManagedCredentialConfig, NetworkPeerConfig},
common::{CompressionAlgoPb, SecureModeConfig}, common::{CompressionAlgoPb, SecureModeConfig},
}; };
@@ -425,6 +426,11 @@ mod tests {
enable_private_mode: Some(true), enable_private_mode: Some(true),
disable_relay_data: Some(true), disable_relay_data: Some(true),
proxy_cidrs: vec!["10.88.0.0/24".to_owned()], proxy_cidrs: vec!["10.88.0.0/24".to_owned()],
managed_credentials: vec![ManagedCredentialConfig {
credential_id: "managed".to_owned(),
credential_secret: "secret".to_owned(),
..Default::default()
}],
port_forwards: vec![crate::proto::api::manage::PortForwardConfig { port_forwards: vec![crate::proto::api::manage::PortForwardConfig {
proto: "tcp".to_owned(), proto: "tcp".to_owned(),
bind_ip: "127.0.0.1".to_owned(), bind_ip: "127.0.0.1".to_owned(),
@@ -450,6 +456,7 @@ mod tests {
assert_eq!(hosted.enable_private_mode, Some(true)); assert_eq!(hosted.enable_private_mode, Some(true));
assert_eq!(hosted.disable_relay_data, Some(true)); assert_eq!(hosted.disable_relay_data, Some(true));
assert_eq!(hosted.proxy_cidrs, original.proxy_cidrs); assert_eq!(hosted.proxy_cidrs, original.proxy_cidrs);
assert_eq!(hosted.managed_credentials, original.managed_credentials);
assert_eq!(hosted.port_forwards, original.port_forwards); assert_eq!(hosted.port_forwards, original.port_forwards);
assert_eq!(hosted.enable_vpn_portal, None); assert_eq!(hosted.enable_vpn_portal, None);
assert_eq!(hosted.data_compress_algo, None); assert_eq!(hosted.data_compress_algo, None);
+1
View File
@@ -128,6 +128,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
".common.Ipv4Addr", ".common.Ipv4Addr",
".common.Ipv6Addr", ".common.Ipv6Addr",
".common.UUID", ".common.UUID",
".api.manage.ManagedCredentialConfig",
".api.manage.VpnPortalConfig", ".api.manage.VpnPortalConfig",
]); ]);
+1
View File
@@ -29,6 +29,7 @@ message InstanceConfigPatch {
optional string ipv6_public_addr_prefix = 13; optional string ipv6_public_addr_prefix = 13;
optional bool disable_relay_data = 14; optional bool disable_relay_data = 14;
repeated VpnPortalClientPatch vpn_portal_clients = 15; repeated VpnPortalClientPatch vpn_portal_clients = 15;
api.manage.ManagedCredentialSet managed_credentials = 16;
} }
message VpnPortalClientPatch { message VpnPortalClientPatch {
+15
View File
@@ -104,6 +104,21 @@ message NetworkConfig {
optional uint32 socket_mark = 67; optional uint32 socket_mark = 67;
repeated NetworkPeerConfig peers = 68; repeated NetworkPeerConfig peers = 68;
optional VpnPortalConfig vpn_portal_config = 69; optional VpnPortalConfig vpn_portal_config = 69;
repeated ManagedCredentialConfig managed_credentials = 71;
}
message ManagedCredentialConfig {
string credential_id = 1;
string credential_secret = 2;
repeated string groups = 3;
bool allow_relay = 4;
repeated string allowed_proxy_cidrs = 5;
int64 expiry_unix = 6;
optional bool reusable = 7;
}
message ManagedCredentialSet {
repeated ManagedCredentialConfig entries = 1;
} }
message VpnPortalClientConfig { message VpnPortalClientConfig {
+28
View File
@@ -335,6 +335,21 @@ pub mod manage {
#[cfg(feature = "json-rpc")] #[cfg(feature = "json-rpc")]
include!(concat!(env!("OUT_DIR"), "/api.manage.serde.rs")); include!(concat!(env!("OUT_DIR"), "/api.manage.serde.rs"));
impl std::fmt::Debug for ManagedCredentialConfig {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("ManagedCredentialConfig")
.field("credential_id", &self.credential_id)
.field("credential_secret", &"<redacted>")
.field("groups", &self.groups)
.field("allow_relay", &self.allow_relay)
.field("allowed_proxy_cidrs", &self.allowed_proxy_cidrs)
.field("expiry_unix", &self.expiry_unix)
.field("reusable", &self.reusable)
.finish()
}
}
impl std::fmt::Debug for VpnPortalConfig { impl std::fmt::Debug for VpnPortalConfig {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter formatter
@@ -379,6 +394,19 @@ mod tests {
assert!(!debug.contains("private-key-material")); assert!(!debug.contains("private-key-material"));
} }
#[test]
fn managed_credential_debug_redacts_secret() {
let credential = super::manage::ManagedCredentialConfig {
credential_id: "managed".to_owned(),
credential_secret: "private-key-material".to_owned(),
..Default::default()
};
let debug = format!("{credential:?}");
assert!(debug.contains("<redacted>"));
assert!(!debug.contains("private-key-material"));
}
#[derive(Clone, Default)] #[derive(Clone, Default)]
struct WebClientServiceJsonCallHandler; struct WebClientServiceJsonCallHandler;
@@ -15,7 +15,8 @@ use easytier::{
}, },
instance::{InstanceIdentifier, instance_identifier}, instance::{InstanceIdentifier, instance_identifier},
manage::{ manage::{
ConfigSource as RpcConfigSource, GetNetworkInstanceConfigRequest, NetworkConfig, ConfigSource as RpcConfigSource, GetNetworkInstanceConfigRequest,
ManagedCredentialConfig, ManagedCredentialSet, NetworkConfig,
RunNetworkInstanceRequest, RunNetworkInstanceRequest,
}, },
}, },
@@ -66,6 +67,7 @@ fn hot_patch_base(config: &NetworkConfig) -> anyhow::Result<NetworkConfig> {
// VPN portal clients are diffed separately; the listener identity // VPN portal clients are diffed separately; the listener identity
// (address and private key) decides between patch and recreate. // (address and private key) decides between patch and recreate.
config.vpn_portal_config = None; config.vpn_portal_config = None;
config.managed_credentials.clear();
if config.dhcp.unwrap_or_default() { if config.dhcp.unwrap_or_default() {
config.virtual_ipv4 = None; config.virtual_ipv4 = None;
config.network_length = None; config.network_length = None;
@@ -256,6 +258,12 @@ fn client_name_only(name: &str) -> easytier::proto::api::manage::VpnPortalClient
} }
} }
fn normalized_managed_credentials(
config: &NetworkConfig,
) -> anyhow::Result<Vec<ManagedCredentialConfig>> {
Ok(NetworkConfig::new_from_config(config.gen_config()?)?.managed_credentials)
}
fn web_source_runtime_patch( fn web_source_runtime_patch(
current: &NetworkConfig, current: &NetworkConfig,
desired: &NetworkConfig, desired: &NetworkConfig,
@@ -328,6 +336,13 @@ fn web_source_runtime_patch(
(Some(_), None) | (None, Some(_)) => return Ok(None), (Some(_), None) | (None, Some(_)) => return Ok(None),
(None, None) => {} (None, None) => {}
} }
let current_managed_credentials = normalized_managed_credentials(current)?;
let desired_managed_credentials = normalized_managed_credentials(desired)?;
if current_managed_credentials != desired_managed_credentials {
patch.managed_credentials = Some(ManagedCredentialSet {
entries: desired_managed_credentials,
});
}
Ok(Some(patch)) Ok(Some(patch))
} }
@@ -339,7 +354,7 @@ fn ensure_runtime_config_converged(
let patch = web_source_runtime_patch(current, desired)?; let patch = web_source_runtime_patch(current, desired)?;
match patch { match patch {
Some(patch) if patch == InstanceConfigPatch::default() => Ok(()), Some(patch) if patch == InstanceConfigPatch::default() => Ok(()),
Some(patch) => anyhow::bail!("runtime config still needs patch after reconcile: {patch:?}"), Some(_) => anyhow::bail!("runtime config still needs patch after reconcile"),
None => anyhow::bail!("runtime config still needs full overwrite after reconcile"), None => anyhow::bail!("runtime config still needs full overwrite after reconcile"),
} }
} }
@@ -771,6 +786,27 @@ mod tests {
assert!(patch.is_none()); assert!(patch.is_none());
} }
#[test]
fn runtime_patch_replaces_managed_credentials_without_full_run() {
let current = config_with_port_forwards(Vec::new());
let mut desired = current.clone();
desired.managed_credentials = vec![ManagedCredentialConfig {
credential_id: "managed".to_owned(),
credential_secret: "credential-secret".to_owned(),
expiry_unix: 2_000_000_000,
..Default::default()
}];
let action = prepare_web_source_runtime_reconcile_from_current(&current, desired)
.expect("prepare reconcile");
let RuntimeReconcileAction::Patch(patch) = action else {
panic!("managed credential change must use a hot patch");
};
let managed = patch.managed_credentials.expect("managed credential patch");
assert_eq!(managed.entries.len(), 1);
assert_eq!(managed.entries[0].credential_id, "managed");
}
#[test] #[test]
fn runtime_patch_rejects_routes_change() { fn runtime_patch_rejects_routes_change() {
let mut current = config_with_port_forwards(Vec::new()); let mut current = config_with_port_forwards(Vec::new());
+32
View File
@@ -11,6 +11,8 @@ use sea_orm::{
}; };
use sea_orm_migration::MigratorTrait as _; use sea_orm_migration::MigratorTrait as _;
use sqlx::{Sqlite, SqlitePool, migrate::MigrateDatabase as _, types::chrono}; use sqlx::{Sqlite, SqlitePool, migrate::MigrateDatabase as _, types::chrono};
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt as _;
use uuid::Uuid; use uuid::Uuid;
use crate::migrator; use crate::migrator;
@@ -18,6 +20,35 @@ use async_trait::async_trait;
pub type UserIdInDb = i32; pub type UserIdInDb = i32;
#[cfg(unix)]
fn restrict_database_file_permissions(db_path: &str) -> anyhow::Result<()> {
if db_path.ends_with(":memory:") || db_path.contains("mode=memory") {
return Ok(());
}
let path = db_path
.strip_prefix("sqlite://")
.or_else(|| db_path.strip_prefix("sqlite:"))
.unwrap_or(db_path);
let path = path
.strip_prefix("file:")
.unwrap_or(path)
.split('?')
.next()
.filter(|path| !path.is_empty());
let Some(path) = path else {
return Ok(());
};
let mut permissions = std::fs::metadata(path)?.permissions();
permissions.set_mode(0o600);
std::fs::set_permissions(path, permissions)?;
Ok(())
}
#[cfg(not(unix))]
fn restrict_database_file_permissions(_db_path: &str) -> anyhow::Result<()> {
Ok(())
}
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct Db { pub struct Db {
db_path: String, db_path: String,
@@ -48,6 +79,7 @@ impl Db {
tracing::info!("Database not found, creating a new one"); tracing::info!("Database not found, creating a new one");
Sqlite::create_database(db_path).await?; Sqlite::create_database(db_path).await?;
} }
restrict_database_file_permissions(db_path)?;
let db = sqlx::pool::PoolOptions::new() let db = sqlx::pool::PoolOptions::new()
.max_lifetime(None) .max_lifetime(None)
+39 -4
View File
@@ -1,6 +1,6 @@
use std::{io::Write, path::PathBuf, sync::Arc}; use std::{io::Write, path::PathBuf, sync::Arc};
use atomic_write_file::AtomicWriteFile; use atomic_write_file::{AtomicWriteFile, OpenOptions};
use easytier_core::peers::credential_manager::CredentialStorage; use easytier_core::peers::credential_manager::CredentialStorage;
struct FileCredentialStorage { struct FileCredentialStorage {
@@ -9,21 +9,33 @@ struct FileCredentialStorage {
impl CredentialStorage for FileCredentialStorage { impl CredentialStorage for FileCredentialStorage {
fn load(&self) -> anyhow::Result<Option<String>> { fn load(&self) -> anyhow::Result<Option<String>> {
let Ok(serialized) = std::fs::read_to_string(&self.path) else { let serialized = match std::fs::read_to_string(&self.path) {
return Ok(None); Ok(serialized) => serialized,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(error.into()),
}; };
tracing::info!(path = %self.path.display(), "loaded credentials"); tracing::info!(path = %self.path.display(), "loaded credentials");
Ok(Some(serialized)) Ok(Some(serialized))
} }
fn store(&self, serialized_credentials: &str) -> anyhow::Result<()> { fn store(&self, serialized_credentials: &str) -> anyhow::Result<()> {
let mut file = AtomicWriteFile::open(&self.path)?; let mut file = restricted_atomic_file(&self.path)?;
file.write_all(serialized_credentials.as_bytes())?; file.write_all(serialized_credentials.as_bytes())?;
file.commit()?; file.commit()?;
Ok(()) Ok(())
} }
} }
fn restricted_atomic_file(path: &std::path::Path) -> std::io::Result<AtomicWriteFile> {
let mut options = OpenOptions::new();
#[cfg(unix)]
{
atomic_write_file::unix::OpenOptionsExt::preserve_mode(&mut options, false);
std::os::unix::fs::OpenOptionsExt::mode(&mut options, 0o600);
}
options.open(path)
}
pub(crate) fn runtime_credential_storage( pub(crate) fn runtime_credential_storage(
path: Option<PathBuf>, path: Option<PathBuf>,
) -> Option<Arc<dyn CredentialStorage>> { ) -> Option<Arc<dyn CredentialStorage>> {
@@ -48,5 +60,28 @@ mod tests {
storage.load().unwrap().as_deref(), storage.load().unwrap().as_deref(),
Some("{\"credential\":false}") Some("{\"credential\":false}")
); );
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt as _;
assert_eq!(
std::fs::metadata(&storage.path)
.unwrap()
.permissions()
.mode()
& 0o777,
0o600
);
}
}
#[test]
fn file_storage_reports_read_errors() {
let directory = tempfile::tempdir().unwrap();
let storage = FileCredentialStorage {
path: directory.path().to_path_buf(),
};
assert!(storage.load().is_err());
} }
} }
+42 -2
View File
@@ -1,4 +1,6 @@
use std::path::Path; use std::{io::Write as _, path::Path};
use atomic_write_file::{AtomicWriteFile, OpenOptions};
use easytier_core::management::{ConfigFileControl, ConfigFilePermission, ConfigFileStorage}; use easytier_core::management::{ConfigFileControl, ConfigFilePermission, ConfigFileStorage};
@@ -31,7 +33,20 @@ impl ConfigFileStorage for NativeConfigFileStorage {
} }
async fn write(&self, path: &Path, contents: &[u8]) -> anyhow::Result<()> { async fn write(&self, path: &Path, contents: &[u8]) -> anyhow::Result<()> {
tokio::fs::write(path, contents).await?; let path = path.to_owned();
let contents = contents.to_owned();
tokio::task::spawn_blocking(move || {
let mut options = OpenOptions::new();
#[cfg(unix)]
{
atomic_write_file::unix::OpenOptionsExt::preserve_mode(&mut options, false);
std::os::unix::fs::OpenOptionsExt::mode(&mut options, 0o600);
}
let mut file: AtomicWriteFile = options.open(path)?;
file.write_all(&contents)?;
file.commit()
})
.await??;
Ok(()) Ok(())
} }
@@ -40,3 +55,28 @@ impl ConfigFileStorage for NativeConfigFileStorage {
Ok(()) Ok(())
} }
} }
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn config_write_is_atomic_and_private() {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("instance.toml");
let storage = NativeConfigFileStorage;
storage.write(&path, b"first").await.unwrap();
storage.write(&path, b"second").await.unwrap();
assert_eq!(std::fs::read_to_string(&path).unwrap(), "second");
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt as _;
assert_eq!(
std::fs::metadata(path).unwrap().permissions().mode() & 0o777,
0o600
);
}
}
}
+1 -1
View File
@@ -95,7 +95,7 @@ impl TestConfigPatcher {
&self, &self,
patch: crate::proto::api::config::InstanceConfigPatch, patch: crate::proto::api::config::InstanceConfigPatch,
) -> anyhow::Result<()> { ) -> anyhow::Result<()> {
easytier_core::management::apply_config_patch(&self.core, patch).await easytier_core::management::apply_config_patch(&self.core, patch, None).await
} }
} }
+24 -14
View File
@@ -30,7 +30,7 @@ const PUBLIC_SERVER_NETWORK_NAME: &str = "__public_server__";
const PUBLIC_SERVER_SHARED_SECRET: &str = "public-server-shared-secret"; const PUBLIC_SERVER_SHARED_SECRET: &str = "public-server-shared-secret";
const NEED_P2P_ADMIN_NETWORK_NAME: &str = "need_p2p_credential_test_network"; const NEED_P2P_ADMIN_NETWORK_NAME: &str = "need_p2p_credential_test_network";
fn generate_credential( async fn generate_credential(
admin: &Instance, admin: &Instance,
groups: Vec<String>, groups: Vec<String>,
allow_relay: bool, allow_relay: bool,
@@ -46,9 +46,10 @@ fn generate_credential(
None, None,
true, true,
) )
.await
} }
fn generate_credential_with_options( async fn generate_credential_with_options(
admin: &Instance, admin: &Instance,
groups: Vec<String>, groups: Vec<String>,
allow_relay: bool, allow_relay: bool,
@@ -184,7 +185,7 @@ async fn create_credential_config(
ipv6: &str, ipv6: &str,
) -> TomlConfigLoader { ) -> TomlConfigLoader {
let (_cred_id, cred_secret) = let (_cred_id, cred_secret) =
generate_credential(admin_inst, vec![], false, vec![], Duration::from_secs(3600)); generate_credential(admin_inst, vec![], false, vec![], Duration::from_secs(3600)).await;
build_credential_config( build_credential_config(
admin_inst admin_inst
@@ -481,7 +482,8 @@ async fn credential_peers_p2p_to_need_p2p_admin_through_public_server(
Duration::from_secs(3600), Duration::from_secs(3600),
Some("credential-peer-a".to_string()), Some("credential-peer-a".to_string()),
false, false,
); )
.await;
let (_credential_b_id, credential_b_secret) = generate_credential_with_options( let (_credential_b_id, credential_b_secret) = generate_credential_with_options(
&admin_inst, &admin_inst,
vec![], vec![],
@@ -490,7 +492,8 @@ async fn credential_peers_p2p_to_need_p2p_admin_through_public_server(
Duration::from_secs(3600), Duration::from_secs(3600),
Some("credential-peer-b".to_string()), Some("credential-peer-b".to_string()),
false, false,
); )
.await;
admin_inst admin_inst
.get_global_ctx() .get_global_ctx()
.issue_event(GlobalCtxEvent::CredentialChanged); .issue_event(GlobalCtxEvent::CredentialChanged);
@@ -593,7 +596,7 @@ async fn credential_peers_p2p_to_need_p2p_admin_through_public_server(
.await; .await;
} }
fn create_generated_credential_config( async fn create_generated_credential_config(
admin_inst: &Instance, admin_inst: &Instance,
inst_name: &str, inst_name: &str,
ns: Option<&str>, ns: Option<&str>,
@@ -601,7 +604,7 @@ fn create_generated_credential_config(
ipv6: &str, ipv6: &str,
) -> (TomlConfigLoader, String) { ) -> (TomlConfigLoader, String) {
let (cred_id, cred_secret) = let (cred_id, cred_secret) =
generate_credential(admin_inst, vec![], false, vec![], Duration::from_secs(3600)); generate_credential(admin_inst, vec![], false, vec![], Duration::from_secs(3600)).await;
let config = build_credential_config( let config = build_credential_config(
admin_inst admin_inst
.get_global_ctx() .get_global_ctx()
@@ -881,7 +884,8 @@ async fn credential_relay_capability(#[case] allow_relay: bool) {
false, false,
vec![], vec![],
Duration::from_secs(3600), Duration::from_secs(3600),
); )
.await;
let (_cred_b_id, cred_b_secret) = generate_credential( let (_cred_b_id, cred_b_secret) = generate_credential(
&admin_inst, &admin_inst,
@@ -889,7 +893,8 @@ async fn credential_relay_capability(#[case] allow_relay: bool) {
false, false,
vec![], vec![],
Duration::from_secs(3600), Duration::from_secs(3600),
); )
.await;
let (_cred_c_id, cred_c_secret) = generate_credential( let (_cred_c_id, cred_c_secret) = generate_credential(
&admin_inst, &admin_inst,
@@ -897,7 +902,8 @@ async fn credential_relay_capability(#[case] allow_relay: bool) {
allow_relay, allow_relay,
vec![], vec![],
Duration::from_secs(3600), Duration::from_secs(3600),
); )
.await;
// Create credential A on ns_c1 // Create credential A on ns_c1
let cred_a_config = { let cred_a_config = {
@@ -1215,7 +1221,8 @@ async fn credential_revocation_propagates() {
false, false,
vec![], vec![],
Duration::from_secs(3600), Duration::from_secs(3600),
); )
.await;
// Create credential node // Create credential node
let cred_config = { let cred_config = {
@@ -1335,7 +1342,8 @@ async fn credential_non_reusable_allows_only_one_peer() {
Duration::from_secs(3600), Duration::from_secs(3600),
None, None,
false, false,
); )
.await;
let network_name = admin_inst let network_name = admin_inst
.get_global_ctx() .get_global_ctx()
@@ -1582,7 +1590,8 @@ async fn credential_unknown_via_shared_rejected(#[values(true, false)] test_revo
Some("ns_c2"), Some("ns_c2"),
"10.144.144.5", "10.144.144.5",
"fd00::5/64", "fd00::5/64",
); )
.await;
(config, Some(cred_id)) (config, Some(cred_id))
} else { } else {
( (
@@ -1841,7 +1850,8 @@ async fn credential_non_reusable_across_two_admins_allows_only_one_peer() {
Duration::from_secs(3600), Duration::from_secs(3600),
None, None,
false, false,
); )
.await;
admin_a_inst admin_a_inst
.get_global_ctx() .get_global_ctx()
.issue_event(GlobalCtxEvent::CredentialChanged); .issue_event(GlobalCtxEvent::CredentialChanged);
+3
View File
@@ -2210,6 +2210,7 @@ pub async fn wireguard_vpn_portal_dynamic_clients() {
}], }],
..Default::default() ..Default::default()
}, },
None,
) )
.await .await
.unwrap(); .unwrap();
@@ -2239,6 +2240,7 @@ pub async fn wireguard_vpn_portal_dynamic_clients() {
}], }],
..Default::default() ..Default::default()
}, },
None,
) )
.await .await
.unwrap_err(); .unwrap_err();
@@ -2301,6 +2303,7 @@ pub async fn wireguard_vpn_portal_dynamic_clients() {
}], }],
..Default::default() ..Default::default()
}, },
None,
) )
.await .await
.unwrap(); .unwrap();