feat: add delivery sequence numbers + major server/client refactor

Delivery sequence numbers (MLS epoch ordering fix): - schemas/node.capnp: add Envelope{seq,data} struct; enqueue returns seq:UInt64; fetch/fetchWait return List(Envelope) instead of List(Data) - storage.rs: Store trait enqueue returns u64; fetch/fetch_limited return Vec<(u64, Vec<u8>)>; FileBackedStore gains QueueMapV3 with per-inbox seq counters and V2→V3 on-disk migration - migrations/002_add_seq.sql: seq column, delivery_seq_counters table, index - sql_store.rs: atomic UPSERT counter via RETURNING, ORDER BY seq, SCHEMA_VERSION→3 - node_service/delivery.rs: builds Envelope list; returns seq from enqueue - client/rpc.rs: enqueue→u64, fetch_all/fetch_wait→Vec<(u64,Vec<u8>)> - client/commands.rs: sort-by-seq before MLS processing; retry loop in cmd_recv and receive_pending_plaintexts for correct epoch ordering Server refactor: - Split monolithic main.rs into node_service/{mod,delivery,auth_ops,key_ops,p2p_ops} - Add auth.rs (token validation, rate limiting), config.rs, metrics.rs, tls.rs - Add SQL migrations runner (001_initial.sql, 002_add_seq.sql) - OPAQUE PAKE login/registration, sealed-sender mode, queue depth limit (1000) Client refactor: - Split lib.rs into client/{commands,rpc,state,retry,hex,mod} - Add cmd_whoami, cmd_health, cmd_check_key, cmd_ping subcommands - Add cmd_register_user, cmd_login (OPAQUE), cmd_refresh_keypackage - Hybrid PQ envelope (X25519 + ML-KEM-768) on all send/recv paths - E2E test suite expanded Other: - quicnprotochat-gui: Tauri 2 desktop GUI skeleton (backend + HTML UI) - quicnprotochat-p2p: iroh-based P2P transport stub - quicnprotochat-core: app_message, hybrid_crypto modules; GroupMember API updates - .github/workflows/size-lint.yml: binary size regression check - docs: protocol comparison, roadmap updates, fully-operational checklist
2026-02-22 20:40:12 +01:00
parent b5b361e2ff
commit 6b8b61c6ae
56 changed files with 10693 additions and 3024 deletions
--- a/crates/quicnprotochat-server/src/node_service/delivery.rs
+++ b/crates/quicnprotochat-server/src/node_service/delivery.rs
@@ -0,0 +1,318 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use capnp::capability::Promise;
+use dashmap::DashMap;
+use quicnprotochat_proto::node_capnp::node_service;
+use tokio::sync::Notify;
+use tokio::time::timeout;
+
+use crate::auth::{
+    check_rate_limit, coded_error, fmt_hex, require_identity_or_request, validate_auth_context,
+};
+use crate::error_codes::*;
+use crate::metrics;
+use crate::storage::{StorageError, Store};
+
+use super::{NodeServiceImpl, CURRENT_WIRE_VERSION};
+
+// Audit events here must not include secrets: no payload content, no full recipient/token bytes (prefix only).
+
+const MAX_PAYLOAD_BYTES: usize = 5 * 1024 * 1024; // 5 MB cap per message
+const MAX_QUEUE_DEPTH: usize = 1000;
+
+fn storage_err(err: StorageError) -> capnp::Error {
+    coded_error(E009_STORAGE_ERROR, err)
+}
+
+pub fn fill_payloads_wait(
+    results: &mut node_service::FetchWaitResults,
+    messages: Vec<(u64, Vec<u8>)>,
+) {
+    let mut list = results.get().init_payloads(messages.len() as u32);
+    for (i, (seq, data)) in messages.iter().enumerate() {
+        let mut entry = list.reborrow().get(i as u32);
+        entry.set_seq(*seq);
+        entry.set_data(data);
+    }
+}
+
+impl NodeServiceImpl {
+    pub fn handle_enqueue(
+        &mut self,
+        params: node_service::EnqueueParams,
+        mut results: node_service::EnqueueResults,
+    ) -> Promise<(), capnp::Error> {
+        let p = match params.get() {
+            Ok(p) => p,
+            Err(e) => return Promise::err(coded_error(E020_BAD_PARAMS, e)),
+        };
+        let recipient_key = match p.get_recipient_key() {
+            Ok(v) => v.to_vec(),
+            Err(e) => return Promise::err(coded_error(E020_BAD_PARAMS, e)),
+        };
+        let payload = match p.get_payload() {
+            Ok(v) => v.to_vec(),
+            Err(e) => return Promise::err(coded_error(E020_BAD_PARAMS, e)),
+        };
+        let channel_id = p.get_channel_id().unwrap_or_default().to_vec();
+        let version = p.get_version();
+        let auth_ctx = match validate_auth_context(&self.auth_cfg, &self.sessions, p.get_auth()) {
+            Ok(ctx) => ctx,
+            Err(e) => return Promise::err(e),
+        };
+
+        if recipient_key.len() != 32 {
+            return Promise::err(coded_error(
+                E004_IDENTITY_KEY_LENGTH,
+                format!("recipientKey must be exactly 32 bytes, got {}", recipient_key.len()),
+            ));
+        }
+        if payload.is_empty() {
+            return Promise::err(coded_error(E005_PAYLOAD_EMPTY, "payload must not be empty"));
+        }
+        if payload.len() > MAX_PAYLOAD_BYTES {
+            return Promise::err(coded_error(
+                E006_PAYLOAD_TOO_LARGE,
+                format!("payload exceeds max size ({} bytes)", MAX_PAYLOAD_BYTES),
+            ));
+        }
+        if version != CURRENT_WIRE_VERSION {
+            return Promise::err(coded_error(
+                E012_WIRE_VERSION,
+                format!("unsupported wire version {} (expected {CURRENT_WIRE_VERSION})", version),
+            ));
+        }
+
+        if let Err(e) = check_rate_limit(&self.rate_limits, &auth_ctx.token) {
+            // Audit: rate limit hit — do not log token or identity.
+            tracing::warn!("rate_limit_hit");
+            metrics::record_rate_limit_hit_total();
+            return Promise::err(e);
+        }
+
+        // When sealed_sender is true, enqueue does not require identity; valid token only.
+        if !self.sealed_sender {
+            if let Err(e) = require_identity_or_request(
+                &auth_ctx,
+                &recipient_key,
+                self.auth_cfg.allow_insecure_identity_from_request,
+            ) {
+                return Promise::err(e);
+            }
+        }
+
+        match self.store.queue_depth(&recipient_key, &channel_id) {
+            Ok(depth) if depth >= MAX_QUEUE_DEPTH => {
+                return Promise::err(coded_error(
+                    E015_QUEUE_FULL,
+                    format!("queue depth {} exceeds limit {}", depth, MAX_QUEUE_DEPTH),
+                ));
+            }
+            Err(e) => return Promise::err(storage_err(e)),
+            _ => {}
+        }
+
+        let payload_len = payload.len();
+        let seq = match self
+            .store
+            .enqueue(&recipient_key, &channel_id, payload)
+            .map_err(storage_err)
+        {
+            Ok(seq) => seq,
+            Err(e) => return Promise::err(e),
+        };
+
+        results.get().set_seq(seq);
+
+        // Metrics and audit. Audit events must not include secrets (no payload, no full keys).
+        metrics::record_enqueue_total();
+        metrics::record_enqueue_bytes(payload_len as u64);
+        if let Ok(depth) = self.store.queue_depth(&recipient_key, &channel_id) {
+            metrics::record_delivery_queue_depth(depth);
+        }
+        tracing::info!(
+            recipient_prefix = %fmt_hex(&recipient_key[..4]),
+            payload_len = payload_len,
+            seq = seq,
+            "audit: enqueue"
+        );
+
+        crate::auth::waiter(&self.waiters, &recipient_key).notify_waiters();
+
+        Promise::ok(())
+    }
+
+    pub fn handle_fetch(
+        &mut self,
+        params: node_service::FetchParams,
+        mut results: node_service::FetchResults,
+    ) -> Promise<(), capnp::Error> {
+        let recipient_key = match params.get() {
+            Ok(p) => match p.get_recipient_key() {
+                Ok(v) => v.to_vec(),
+                Err(e) => return Promise::err(coded_error(E020_BAD_PARAMS, e)),
+            },
+            Err(e) => return Promise::err(coded_error(E020_BAD_PARAMS, e)),
+        };
+        let channel_id = params
+            .get()
+            .ok()
+            .and_then(|p| p.get_channel_id().ok())
+            .map(|c| c.to_vec())
+            .unwrap_or_default();
+        let version = params
+            .get()
+            .ok()
+            .map(|p| p.get_version())
+            .unwrap_or(CURRENT_WIRE_VERSION);
+        let limit = params.get().ok().map(|p| p.get_limit()).unwrap_or(0);
+        let auth_ctx = match params
+            .get()
+            .ok()
+            .map(|p| validate_auth_context(&self.auth_cfg, &self.sessions, p.get_auth()))
+            .transpose()
+        {
+            Ok(ctx) => ctx,
+            Err(e) => return Promise::err(e),
+        };
+
+        if recipient_key.len() != 32 {
+            return Promise::err(coded_error(
+                E004_IDENTITY_KEY_LENGTH,
+                format!("recipientKey must be exactly 32 bytes, got {}", recipient_key.len()),
+            ));
+        }
+        if version != CURRENT_WIRE_VERSION {
+            return Promise::err(coded_error(
+                E012_WIRE_VERSION,
+                format!("unsupported wire version {} (expected {CURRENT_WIRE_VERSION})", version),
+            ));
+        }
+
+        let auth_ctx = match auth_ctx {
+            Some(ctx) => ctx,
+            None => return Promise::err(coded_error(E003_INVALID_TOKEN, "auth required")),
+        };
+
+        if let Err(e) = require_identity_or_request(
+            &auth_ctx,
+            &recipient_key,
+            self.auth_cfg.allow_insecure_identity_from_request,
+        ) {
+            return Promise::err(e);
+        }
+
+        let messages = if limit > 0 {
+            match self
+                .store
+                .fetch_limited(&recipient_key, &channel_id, limit as usize)
+                .map_err(storage_err)
+            {
+                Ok(m) => m,
+                Err(e) => return Promise::err(e),
+            }
+        } else {
+            match self
+                .store
+                .fetch(&recipient_key, &channel_id)
+                .map_err(storage_err)
+            {
+                Ok(m) => m,
+                Err(e) => return Promise::err(e),
+            }
+        };
+
+        // Audit: fetch — do not log payload or full keys.
+        metrics::record_fetch_total();
+        tracing::info!(
+            recipient_prefix = %fmt_hex(&recipient_key[..4]),
+            count = messages.len(),
+            "audit: fetch"
+        );
+
+        let mut list = results.get().init_payloads(messages.len() as u32);
+        for (i, (seq, data)) in messages.iter().enumerate() {
+            let mut entry = list.reborrow().get(i as u32);
+            entry.set_seq(*seq);
+            entry.set_data(data);
+        }
+
+        Promise::ok(())
+    }
+
+    pub fn handle_fetch_wait(
+        &mut self,
+        params: node_service::FetchWaitParams,
+        mut results: node_service::FetchWaitResults,
+    ) -> Promise<(), capnp::Error> {
+        let p = match params.get() {
+            Ok(p) => p,
+            Err(e) => return Promise::err(coded_error(E020_BAD_PARAMS, e)),
+        };
+        let recipient_key = match p.get_recipient_key() {
+            Ok(v) => v.to_vec(),
+            Err(e) => return Promise::err(coded_error(E020_BAD_PARAMS, e)),
+        };
+        let channel_id = p.get_channel_id().unwrap_or_default().to_vec();
+        let version = p.get_version();
+        let timeout_ms = p.get_timeout_ms();
+        let limit = p.get_limit();
+        let auth_ctx = match validate_auth_context(&self.auth_cfg, &self.sessions, p.get_auth()) {
+            Ok(ctx) => ctx,
+            Err(e) => return Promise::err(e),
+        };
+
+        if recipient_key.len() != 32 {
+            return Promise::err(coded_error(
+                E004_IDENTITY_KEY_LENGTH,
+                format!("recipientKey must be exactly 32 bytes, got {}", recipient_key.len()),
+            ));
+        }
+        if version != CURRENT_WIRE_VERSION {
+            return Promise::err(coded_error(
+                E012_WIRE_VERSION,
+                format!("unsupported wire version {} (expected {CURRENT_WIRE_VERSION})", version),
+            ));
+        }
+
+        if let Err(e) = require_identity_or_request(
+            &auth_ctx,
+            &recipient_key,
+            self.auth_cfg.allow_insecure_identity_from_request,
+        ) {
+            return Promise::err(e);
+        }
+
+        let store = Arc::clone(&self.store);
+        let waiters: Arc<DashMap<Vec<u8>, Arc<Notify>>> = self.waiters.clone();
+
+        Promise::from_future(async move {
+            let fetch_fn = |s: &Arc<dyn Store>, rk: &[u8], ch: &[u8], lim: u32| -> Result<Vec<(u64, Vec<u8>)>, capnp::Error> {
+                if lim > 0 {
+                    s.fetch_limited(rk, ch, lim as usize).map_err(storage_err)
+                } else {
+                    s.fetch(rk, ch).map_err(storage_err)
+                }
+            };
+
+            let messages = fetch_fn(&store, &recipient_key, &channel_id, limit)?;
+
+            if messages.is_empty() && timeout_ms > 0 {
+                let waiter = waiters
+                    .entry(recipient_key.clone())
+                    .or_insert_with(|| Arc::new(Notify::new()))
+                    .clone();
+                let _ = timeout(Duration::from_millis(timeout_ms), waiter.notified()).await;
+                let msgs = fetch_fn(&store, &recipient_key, &channel_id, limit)?;
+                fill_payloads_wait(&mut results, msgs);
+                metrics::record_fetch_wait_total();
+                return Ok(());
+            }
+
+            fill_payloads_wait(&mut results, messages);
+            metrics::record_fetch_wait_total();
+            Ok(())
+        })
+    }
+}