feat(credentials): support managed credential synchronization (#2490)

* feat(credentials): support managed credential synchronization

Allow managed callers to upsert credentials with an exact ID, secret,
permissions, reuse policy, and expiry.

Return non-secret attributes plus a public-key fingerprint so callers can
verify relay credential consistency.

Persist imported credentials atomically and preserve identity and expiry
across restarts.

* fix(credentials): make managed upserts durable

Write the candidate credential snapshot before committing it to memory.
Propagate storage failures so controllers can retry instead of observing
false convergence.

Cover a transient storage failure to verify that memory stays unchanged
and the retry persists the credential.

* fix(credentials): atomically replace stored snapshots

Define CredentialStorage::store as an atomic replacement boundary and
use atomic-write-file in the management adapter. This keeps the last
committed credential JSON readable when a replacement fails.

Cover replacement of an existing credential snapshot and keep the
dependency scoped to the management feature.

Cargo.lock

@@ -486,6 +486,16 @@ version = "1.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
 
+[[package]]
+name = "atomic-write-file"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aeb1e2c1d58618bea806ccca5bbe65dc4e868be16f69ff118a39049389687548"
+dependencies = [
+ "nix 0.29.0",
+ "rand 0.8.5",
+]
+
 [[package]]
 name = "atomic_refcell"
 version = "0.1.13"
@@ -2288,6 +2298,7 @@ dependencies = [
  "async-recursion",
  "async-trait",
  "atomic-shim",
+ "atomic-write-file",
  "atomic_refcell",
  "auto_impl",
  "base64 0.22.1",

easytier-core/src/instance/management.rs

@@ -8,7 +8,9 @@ use crate::{
     foundation::stats::MetricSnapshot,
     peers::{
         conn::peer_conn::PeerConnId,
-        credential_manager::{CredentialCreateOptions, CredentialInfo, GeneratedCredential},
+        credential_manager::{
+            CredentialCreateOptions, CredentialInfo, CredentialUpsertOptions, GeneratedCredential,
+        },
         peer_manager::PeerSnapshot,
     },
 };
@@ -181,6 +183,21 @@ where
         Ok(revoked)
     }
 
+    pub fn upsert_credential(&self, options: CredentialUpsertOptions) -> anyhow::Result<bool> {
+        if !self.peer_manager.can_manage_credentials() {
+            anyhow::bail!("only admin nodes (with network_secret) can import credentials");
+        }
+        let changed = self
+            .peer_manager
+            .credential_manager()
+            .upsert_credential(options)
+            .map_err(anyhow::Error::msg)?;
+        if changed {
+            self.peer_manager.notify_credential_changed();
+        }
+        Ok(changed)
+    }
+
     pub fn credential_snapshots(&self) -> Vec<CredentialInfo> {
         self.peer_manager.credential_manager().list_credentials()
     }

easytier-core/src/management/instance_rpc/full.rs

@@ -12,8 +12,8 @@ use easytier_proto::{
             ListCredentialsResponse, ListMappedListenerRequest, ListMappedListenerResponse,
             ListPortForwardRequest, ListPortForwardResponse, MappedListener,
             MappedListenerManageRpc, MetricSnapshot, PeerManageRpc, PortForwardManageRpc,
-            RevokeCredentialRequest, RevokeCredentialResponse, StatsRpc, VpnPortalInfo,
+            RevokeCredentialRequest, RevokeCredentialResponse, StatsRpc, UpsertCredentialRequest,
-            VpnPortalRpc,
+            UpsertCredentialResponse, VpnPortalInfo, VpnPortalRpc,
         },
     },
     common::PortForwardConfigPb,
@@ -30,7 +30,9 @@ use crate::{
         CoreInstance, CoreInstanceHost,
         manager::{InstanceFactory, InstanceManager},
     },
-    peers::credential_manager::{CredentialCreateOptions, CredentialInfo as CoreCredentialInfo},
+    peers::credential_manager::{
+        CredentialCreateOptions, CredentialInfo as CoreCredentialInfo, CredentialUpsertOptions,
+    },
 };
 
 use super::InstanceManagementRpc;
@@ -105,6 +107,7 @@ fn credential_info_to_api(info: CoreCredentialInfo) -> CredentialInfo {
         expiry_unix: info.expiry_unix,
         allowed_proxy_cidrs: info.allowed_proxy_cidrs,
         reusable: info.reusable,
+        public_key_fingerprint: info.public_key_fingerprint,
     }
 }
 
@@ -298,9 +301,29 @@ where
         Ok(GenerateCredentialResponse {
             credential_id: generated.credential_id,
             credential_secret: generated.secret,
+            expiry_unix: generated.expiry_unix,
         })
     }
 
+    async fn upsert_credential(
+        &self,
+        _: BaseController,
+        request: UpsertCredentialRequest,
+    ) -> rpc_types::error::Result<UpsertCredentialResponse> {
+        let changed = self
+            .instance(request.instance.as_ref())?
+            .upsert_credential(CredentialUpsertOptions {
+                credential_id: request.credential_id,
+                credential_secret: request.credential_secret,
+                groups: request.groups,
+                allow_relay: request.allow_relay,
+                allowed_proxy_cidrs: request.allowed_proxy_cidrs,
+                expiry_unix: request.expiry_unix,
+                reusable: request.reusable.unwrap_or(true),
+            })?;
+        Ok(UpsertCredentialResponse { changed })
+    }
+
     async fn revoke_credential(
         &self,
         _: BaseController,

easytier-core/src/peers/credential_manager.rs

@@ -6,6 +6,7 @@ use std::{
 
 use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD};
 use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
 use x25519_dalek::{PublicKey, StaticSecret};
 
 use crate::proto::peer_rpc::{TrustedCredentialPubkey, TrustedCredentialPubkeyProof};
@@ -31,6 +32,17 @@ pub struct CredentialCreateOptions {
     pub reusable: bool,
 }
 
+#[derive(Debug, Clone)]
+pub struct CredentialUpsertOptions {
+    pub credential_id: String,
+    pub credential_secret: String,
+    pub groups: Vec<String>,
+    pub allow_relay: bool,
+    pub allowed_proxy_cidrs: Vec<String>,
+    pub expiry_unix: i64,
+    pub reusable: bool,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub(crate) struct CredentialEntry {
     pubkey: String,
@@ -69,6 +81,8 @@ impl CredentialEntry {
             expiry_unix: self.expiry_unix,
             allowed_proxy_cidrs: self.allowed_proxy_cidrs.clone(),
             reusable: Some(self.reusable),
+            public_key_fingerprint: CredentialManager::public_key_fingerprint(&self.pubkey)
+                .unwrap_or_default(),
         }
     }
 }
@@ -81,17 +95,22 @@ pub struct CredentialInfo {
     pub expiry_unix: i64,
     pub allowed_proxy_cidrs: Vec<String>,
     pub reusable: Option<bool>,
+    pub public_key_fingerprint: String,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct GeneratedCredential {
     pub credential_id: String,
     pub secret: String,
+    pub expiry_unix: i64,
     pub changed: bool,
 }
 
 pub trait CredentialStorage: Send + Sync + 'static {
     fn load(&self) -> anyhow::Result<Option<String>>;
+
+    /// Atomically replaces the previously committed credential snapshot.
+    /// Returning an error must leave that snapshot readable.
     fn store(&self, serialized_credentials: &str) -> anyhow::Result<()>;
 }
 
@@ -181,6 +200,7 @@ impl CredentialManager {
                     return GeneratedCredential {
                         credential_id: id,
                         secret: existing.secret.clone(),
+                        expiry_unix: existing.expiry_unix,
                         changed: false,
                     };
                 }
@@ -191,10 +211,12 @@ impl CredentialManager {
 
             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,
             }
         };
@@ -246,6 +268,76 @@ impl CredentialManager {
         removed
     }
 
+    pub fn upsert_credential(&self, options: CredentialUpsertOptions) -> Result<bool, String> {
+        let CredentialUpsertOptions {
+            credential_id,
+            credential_secret,
+            groups,
+            allow_relay,
+            allowed_proxy_cidrs,
+            expiry_unix,
+            reusable,
+        } = options;
+        let credential_id = credential_id.trim().to_string();
+        if credential_id.is_empty() {
+            return Err("credential_id must not be empty".to_string());
+        }
+        if expiry_unix <= current_unix_timestamp() {
+            return Err("expiry_unix must be in the future".to_string());
+        }
+
+        let private_bytes: [u8; 32] = BASE64_STANDARD
+            .decode(credential_secret.trim())
+            .map_err(|_| "credential_secret must be base64".to_string())?
+            .try_into()
+            .map_err(|_| "credential_secret must contain 32 bytes".to_string())?;
+        let private = StaticSecret::from(private_bytes);
+        let entry = CredentialEntry {
+            pubkey: BASE64_STANDARD.encode(PublicKey::from(&private).as_bytes()),
+            secret: BASE64_STANDARD.encode(private.as_bytes()),
+            groups,
+            allow_relay,
+            allowed_proxy_cidrs,
+            reusable,
+            expiry_unix,
+            created_at_unix: current_unix_timestamp(),
+        };
+
+        let _storage_write = self.storage_write.lock().unwrap();
+        let mut credentials = self.credentials.lock().unwrap();
+        if credentials.iter().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());
+        }
+        let changed = credentials.get(&credential_id).is_none_or(|existing| {
+            existing.secret != entry.secret
+                || existing.pubkey != entry.pubkey
+                || existing.groups != entry.groups
+                || existing.allow_relay != entry.allow_relay
+                || existing.allowed_proxy_cidrs != entry.allowed_proxy_cidrs
+                || existing.reusable != entry.reusable
+                || existing.expiry_unix != entry.expiry_unix
+        });
+        if !changed {
+            return Ok(false);
+        }
+
+        if let Some(storage) = &self.storage {
+            let mut updated = credentials.clone();
+            updated.insert(credential_id, entry);
+            let serialized = serde_json::to_string_pretty(&updated)
+                .map_err(|error| format!("failed to serialize credentials: {error}"))?;
+            storage
+                .store(&serialized)
+                .map_err(|error| format!("failed to store credentials: {error}"))?;
+            *credentials = updated;
+        } else {
+            credentials.insert(credential_id, entry);
+        }
+        Ok(true)
+    }
+
     pub fn remove_expired_credentials(&self) -> bool {
         self.remove_expired_credentials_at(current_unix_timestamp())
     }
@@ -309,6 +401,16 @@ impl CredentialManager {
         Some(decoded)
     }
 
+    fn public_key_fingerprint(pubkey: &str) -> Option<String> {
+        let decoded = Self::decode_pubkey_b64(pubkey)?;
+        Some(
+            Sha256::digest(decoded)
+                .iter()
+                .map(|byte| format!("{byte:02x}"))
+                .collect(),
+        )
+    }
+
     fn persist(&self) {
         let Some(storage) = &self.storage else {
             return;
@@ -384,6 +486,36 @@ mod tests {
         }
     }
 
+    struct FailOnceCredentialStorage {
+        serialized: Mutex<Option<String>>,
+        fail_next_store: Mutex<bool>,
+    }
+
+    impl Default for FailOnceCredentialStorage {
+        fn default() -> Self {
+            Self {
+                serialized: Mutex::new(None),
+                fail_next_store: Mutex::new(true),
+            }
+        }
+    }
+
+    impl CredentialStorage for FailOnceCredentialStorage {
+        fn load(&self) -> anyhow::Result<Option<String>> {
+            Ok(self.serialized.lock().unwrap().clone())
+        }
+
+        fn store(&self, serialized_credentials: &str) -> anyhow::Result<()> {
+            let mut fail_next_store = self.fail_next_store.lock().unwrap();
+            if *fail_next_store {
+                *fail_next_store = false;
+                anyhow::bail!("injected credential storage failure");
+            }
+            *self.serialized.lock().unwrap() = Some(serialized_credentials.to_owned());
+            Ok(())
+        }
+    }
+
     #[test]
     fn generate_and_revoke_credential() {
         let mgr = CredentialManager::new();
@@ -396,6 +528,7 @@ mod tests {
 
         assert!(!generated.credential_id.is_empty());
         assert!(!generated.secret.is_empty());
+        assert!(generated.expiry_unix > current_unix_timestamp());
         assert!(generated.changed);
         assert!(uuid::Uuid::parse_str(&generated.credential_id).is_ok());
 
@@ -453,6 +586,104 @@ mod tests {
         assert!(!list[0].allow_relay);
         assert_eq!(list[0].allowed_proxy_cidrs, vec!["10.0.0.0/24".to_string()]);
         assert_eq!(list[0].reusable, Some(true));
+        assert_eq!(list[0].public_key_fingerprint.len(), 64);
+    }
+
+    #[test]
+    fn upsert_credential_preserves_key_attributes_and_storage() {
+        let source = CredentialManager::new();
+        let generated = source.generate_credential_with_options(
+            vec!["users".to_string()],
+            false,
+            vec!["10.0.0.0/8".to_string()],
+            Duration::from_secs(3600),
+            Some("shared-id".to_string()),
+            false,
+        );
+        let source_info = source.list_credentials().remove(0);
+        let options = CredentialUpsertOptions {
+            credential_id: generated.credential_id,
+            credential_secret: generated.secret,
+            groups: source_info.groups.clone(),
+            allow_relay: source_info.allow_relay,
+            allowed_proxy_cidrs: source_info.allowed_proxy_cidrs.clone(),
+            expiry_unix: source_info.expiry_unix,
+            reusable: source_info.reusable.unwrap(),
+        };
+
+        let storage = Arc::new(MemoryCredentialStorage::default());
+        let target = CredentialManager::from_storage(storage.clone());
+        assert!(target.upsert_credential(options.clone()).unwrap());
+        assert!(!target.upsert_credential(options).unwrap());
+        assert_eq!(target.list_credentials(), vec![source_info.clone()]);
+        assert_eq!(
+            CredentialManager::from_storage(storage).list_credentials(),
+            vec![source_info]
+        );
+    }
+
+    #[test]
+    fn upsert_credential_can_retry_after_storage_failure() {
+        let source = CredentialManager::new();
+        let generated =
+            source.generate_credential(vec![], false, vec![], Duration::from_secs(3600));
+        let options = CredentialUpsertOptions {
+            credential_id: generated.credential_id,
+            credential_secret: generated.secret,
+            groups: vec!["users".to_string()],
+            allow_relay: false,
+            allowed_proxy_cidrs: vec![],
+            expiry_unix: generated.expiry_unix,
+            reusable: true,
+        };
+
+        let storage = Arc::new(FailOnceCredentialStorage::default());
+        let target = CredentialManager::from_storage(storage.clone());
+        assert!(target.upsert_credential(options.clone()).is_err());
+        assert!(target.list_credentials().is_empty());
+
+        assert!(target.upsert_credential(options).unwrap());
+        assert_eq!(target.list_credentials().len(), 1);
+        assert_eq!(
+            CredentialManager::from_storage(storage)
+                .list_credentials()
+                .len(),
+            1
+        );
+    }
+
+    #[test]
+    fn upsert_credential_rejects_public_key_assigned_to_another_id() {
+        let source = CredentialManager::new();
+        let generated =
+            source.generate_credential(vec![], false, vec![], Duration::from_secs(3600));
+        let options = CredentialUpsertOptions {
+            credential_id: "original-id".to_string(),
+            credential_secret: generated.secret,
+            groups: vec!["users".to_string()],
+            allow_relay: false,
+            allowed_proxy_cidrs: vec![],
+            expiry_unix: generated.expiry_unix,
+            reusable: true,
+        };
+
+        let target = CredentialManager::new();
+        assert!(target.upsert_credential(options.clone()).unwrap());
+
+        let duplicate = CredentialUpsertOptions {
+            credential_id: "duplicate-id".to_string(),
+            groups: vec!["admins".to_string()],
+            ..options
+        };
+        assert_eq!(
+            target.upsert_credential(duplicate).unwrap_err(),
+            "credential_secret is already used by another credential_id"
+        );
+
+        let credentials = target.list_credentials();
+        assert_eq!(credentials.len(), 1);
+        assert_eq!(credentials[0].credential_id, "original-id");
+        assert_eq!(credentials[0].groups, vec!["users".to_string()]);
     }
 
     #[test]

easytier-proto/proto/api_instance.proto

@@ -345,6 +345,22 @@ message GenerateCredentialRequest {
 message GenerateCredentialResponse {
   string credential_id = 1;       // UUID
   string credential_secret = 2;   // private key base64
+  int64 expiry_unix = 3;
+}
+
+message UpsertCredentialRequest {
+  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;
+  InstanceIdentifier instance = 8;
+}
+
+message UpsertCredentialResponse {
+  bool changed = 1;
 }
 
 message RevokeCredentialRequest {
@@ -367,6 +383,7 @@ message CredentialInfo {
   int64 expiry_unix = 4;
   repeated string allowed_proxy_cidrs = 5;
   optional bool reusable = 6;
+  string public_key_fingerprint = 7;
 }
 
 message ListCredentialsResponse {
@@ -377,4 +394,5 @@ service CredentialManageRpc {
   rpc GenerateCredential(GenerateCredentialRequest) returns (GenerateCredentialResponse);
   rpc RevokeCredential(RevokeCredentialRequest) returns (RevokeCredentialResponse);
   rpc ListCredentials(ListCredentialsRequest) returns (ListCredentialsResponse);
+  rpc UpsertCredential(UpsertCredentialRequest) returns (UpsertCredentialResponse);
 }

easytier/Cargo.toml

@@ -87,6 +87,7 @@ bytes = "1.5.0"
 pin-project-lite = "0.2.13"
 
 atomic_refcell = "0.1.13"
+atomic-write-file = { version = "0.2.3", optional = true }
 
 quinn = { version = "0.11.8", optional = true, features = ["ring"] }
 quinn-proto = { version = "0.11.12", optional = true }
@@ -413,6 +414,7 @@ extended-services = [
     "proxy-cidr-monitor",
 ]
 management = [
+    "dep:atomic-write-file",
     "web-client",
     "logging",
     "easytier-core/management",

easytier/src/common/credential_manager.rs

@@ -1,5 +1,6 @@
-use std::{path::PathBuf, sync::Arc};
+use std::{io::Write, path::PathBuf, sync::Arc};
 
+use atomic_write_file::AtomicWriteFile;
 use easytier_core::peers::credential_manager::CredentialStorage;
 
 struct FileCredentialStorage {
@@ -16,7 +17,9 @@ impl CredentialStorage for FileCredentialStorage {
     }
 
     fn store(&self, serialized_credentials: &str) -> anyhow::Result<()> {
-        std::fs::write(&self.path, serialized_credentials)?;
+        let mut file = AtomicWriteFile::open(&self.path)?;
+        file.write_all(serialized_credentials.as_bytes())?;
+        file.commit()?;
         Ok(())
     }
 }
@@ -40,9 +43,10 @@ mod tests {
 
         assert_eq!(storage.load().unwrap(), None);
         storage.store("{\"credential\":true}").unwrap();
+        storage.store("{\"credential\":false}").unwrap();
         assert_eq!(
             storage.load().unwrap().as_deref(),
-            Some("{\"credential\":true}")
+            Some("{\"credential\":false}")
         );
     }
 }