Draft 0.5: identity/registry spec; Phase A signed MA registry, mailbox challenge auth, key rotation, skew rejection

src/node.rs

@@ -17,12 +17,13 @@ use tokio::net::TcpListener;
 use tokio::task::JoinHandle;
 
 use crate::config::{CLASS_ENRICHMENT, Config, Member, load_members};
-use crate::crypto::Keypair;
+use crate::crypto::{Keypair, now_ts, poll_signing_bytes};
 use crate::index::{LocalIndex, SearchHit, response_items};
 use crate::message::{
     AggregateBody, Envelope, QueryBody, ResponseBody, ResponseItem, TYPE_AGGREGATE, TYPE_QUERY,
-    TYPE_RESPONSE, build_response,
+    TYPE_RESPONSE, build_response, timestamp_is_fresh,
 };
+use crate::registry::{SignedRegistry, authorized_keys, load_registry, verify_registry};
 
 #[derive(Default)]
 struct Aggregates {
@@ -36,6 +37,8 @@ pub struct Node {
     index: LocalIndex,
     members: RwLock<Vec<Member>>,
     members_mtime: Mutex<Option<SystemTime>>,
+    registry: Mutex<Option<SignedRegistry>>,
+    registry_mtime: Mutex<Option<SystemTime>>,
     aggregates: Mutex<Aggregates>,
     seen: Mutex<HashSet<String>>,
     pending: Mutex<HashMap<String, Vec<(String, ResponseBody)>>>,
@@ -119,12 +122,14 @@ impl Node {
             .timeout(Duration::from_secs(15))
             .build()
             .context("building http client")?;
-        Ok(Arc::new(Self {
+        let node = Arc::new(Self {
             config,
             key,
             index,
             members: RwLock::new(members),
             members_mtime: Mutex::new(members_mtime),
+            registry: Mutex::new(None),
+            registry_mtime: Mutex::new(None),
             aggregates: Mutex::new(Aggregates::default()),
             seen: Mutex::new(HashSet::new()),
             pending: Mutex::new(HashMap::new()),
@@ -132,7 +137,90 @@ impl Node {
             client,
             sent: AtomicU64::new(0),
             received: AtomicU64::new(0),
-        }))
+        });
+        node.load_initial_registry();
+        Ok(node)
+    }
+
+    fn load_initial_registry(&self) {
+        let Some(source) = &self.config.node.registry else {
+            return;
+        };
+        let result = if source.starts_with("http://") || source.starts_with("https://") {
+            load_registry(&self.config.registry_cache_path())
+        } else {
+            let path = std::path::PathBuf::from(source);
+            let loaded = load_registry(&path);
+            if loaded.is_ok() {
+                let mtime = fs::metadata(&path)
+                    .and_then(|metadata| metadata.modified())
+                    .ok();
+                *self.registry_mtime.lock().expect("registry mtime lock") = mtime;
+            }
+            loaded
+        };
+        if let Ok(signed) = result {
+            if let Err(error) = self.verify_and_apply_registry(signed) {
+                eprintln!("cached registry rejected: {error}");
+            }
+        }
+    }
+
+    fn verify_and_apply_registry(&self, signed: SignedRegistry) -> Result<()> {
+        let ma_key = self
+            .config
+            .node
+            .ma_key
+            .as_deref()
+            .ok_or_else(|| anyhow!("registry configured without ma_key"))?;
+        verify_registry(&signed, ma_key)?;
+        {
+            let current = self.registry.lock().expect("registry lock");
+            if let Some(existing) = current.as_ref() {
+                if signed.doc.version <= existing.doc.version {
+                    return Err(anyhow!("registry version rollback rejected"));
+                }
+            }
+        }
+        *self.registry.lock().expect("registry lock") = Some(signed);
+        Ok(())
+    }
+
+    fn refresh_registry(&self) {
+        let Some(source) = &self.config.node.registry else {
+            return;
+        };
+        if source.starts_with("http://") || source.starts_with("https://") {
+            return;
+        }
+        let path = std::path::PathBuf::from(source);
+        let mtime = fs::metadata(&path)
+            .and_then(|metadata| metadata.modified())
+            .ok();
+        {
+            let last = self.registry_mtime.lock().expect("registry mtime lock");
+            if *last == mtime {
+                return;
+            }
+        }
+        let Ok(signed) = load_registry(&path) else {
+            return;
+        };
+        *self.registry_mtime.lock().expect("registry mtime lock") = mtime;
+        if let Err(error) = self.verify_and_apply_registry(signed) {
+            eprintln!("registry update rejected: {error}");
+        }
+    }
+
+    async fn fetch_registry(&self) -> Result<()> {
+        let Some(url) = &self.config.node.registry else {
+            return Ok(());
+        };
+        let response = self.client.get(url).send().await?;
+        let signed: SignedRegistry = response.json().await?;
+        self.verify_and_apply_registry(signed.clone())?;
+        crate::registry::save_registry(&self.config.registry_cache_path(), &signed)?;
+        Ok(())
     }
 
     fn refresh_members(&self) {
@@ -156,7 +244,9 @@ impl Node {
     fn member_class(&self, members: &[Member], pubkey: &str) -> Option<String> {
         members
             .iter()
-            .find(|member| member.pubkey == pubkey)
+            .find(|member| {
+                member.pubkey == pubkey || member.previous.iter().any(|key| key == pubkey)
+            })
             .map(|member| member.class.clone())
     }
 
@@ -213,6 +303,22 @@ impl Node {
             .with_context(|| format!("binding {}", node.config.node.listen))?;
         let addr = listener.local_addr()?;
         let mut tasks = Vec::new();
+        if let Some(source) = &node.config.node.registry {
+            if source.starts_with("http://") || source.starts_with("https://") {
+                if let Err(error) = node.fetch_registry().await {
+                    eprintln!("initial registry fetch failed: {error}");
+                }
+                let node_for_registry = node.clone();
+                tasks.push(tokio::spawn(async move {
+                    loop {
+                        if let Err(error) = node_for_registry.fetch_registry().await {
+                            eprintln!("registry fetch failed: {error}");
+                        }
+                        tokio::time::sleep(Duration::from_secs(60)).await;
+                    }
+                }));
+            }
+        }
         for relay in node.config.node.relays.clone() {
             tasks.push(tokio::spawn(poll_relay(node.clone(), relay)));
         }
@@ -333,11 +439,31 @@ impl Node {
         if envelope.verify().is_err() {
             return;
         }
+        if !timestamp_is_fresh(envelope.ts, crate::crypto::now_ts()) {
+            return;
+        }
         self.refresh_members();
-        let (class, listed) = {
+        self.refresh_registry();
+        let (class, listed) = if self.config.node.registry.is_some() {
+            let registry = self.registry.lock().expect("registry lock");
+            let authorized = registry
+                .as_ref()
+                .map(|signed| authorized_keys(signed, now_ts()));
+            match authorized {
+                Some(map) => {
+                    let entry = map.get(&envelope.from);
+                    (entry.map(|(_, class)| class.clone()), entry.is_some())
+                }
+                None => (None, false),
+            }
+        } else {
             let members = self.members.read().expect("members lock");
             let class = self.member_class(&members, &envelope.from);
-            let listed = members.is_empty() || class.is_some();
+            let listed = if members.is_empty() {
+                self.config.node.dev_bootstrap
+            } else {
+                class.is_some()
+            };
             (class, listed)
         };
         if !listed {
@@ -504,13 +630,35 @@ impl Node {
     }
 }
 
+async fn fetch_challenge(node: &Node, base: &str, member: &str) -> Option<String> {
+    let response = node
+        .client
+        .get(format!("{base}/v1/challenge?member={member}"))
+        .send()
+        .await
+        .ok()?;
+    if !response.status().is_success() {
+        return None;
+    }
+    let payload: Value = response.json().await.ok()?;
+    payload
+        .get("nonce")
+        .and_then(Value::as_str)
+        .map(str::to_string)
+}
+
 async fn poll_relay(node: Arc<Node>, relay: String) {
     let base = relay.trim_end_matches('/').to_string();
+    let member = node.key.public_hex();
     loop {
+        let Some(nonce) = fetch_challenge(&node, &base, &member).await else {
+            tokio::time::sleep(Duration::from_secs(1)).await;
+            continue;
+        };
+        let signature = node.key.sign(&poll_signing_bytes(&member, &nonce));
         let url = format!(
-            "{}/v1/poll?member={}&timeout_ms=20000",
-            base,
-            node.key.public_hex()
+            "{}/v1/poll?member={}&nonce={}&sig={}&timeout_ms=20000",
+            base, member, nonce, signature
         );
         match node.client.get(&url).send().await {
             Ok(response) if response.status() == reqwest::StatusCode::NO_CONTENT => continue,
@@ -582,8 +730,16 @@ async fn local_query(
 
 async fn local_status(State(node): State<Arc<Node>>) -> Response {
     let aggregates = node.aggregate_for(&current_period(), None);
+    let registry_version = node
+        .registry
+        .lock()
+        .expect("registry lock")
+        .as_ref()
+        .map(|signed| signed.doc.version);
     Json(json!({
         "name": node.config.node.name,
+        "id": node.config.node.id,
+        "registry_version": registry_version,
         "pubkey": node.key.public_hex(),
         "listen": node.config.node.listen,
         "relays": node.config.node.relays,