From 83c30ad17c5bd1d3d55945c53725bf49767416a2 Mon Sep 17 00:00:00 2001 From: kaichaosun Date: Mon, 20 Jul 2026 15:44:40 +0800 Subject: [PATCH] feat: publish via logos delivery --- Cargo.lock | 22 +++ Cargo.toml | 4 + README.md | 54 ++++++- examples/delivery_smoke_test.rs | 134 +++++++++++++++++ src/handlers.rs | 150 ++++--------------- src/ingest.rs | 102 +++++++++++++ src/main.rs | 48 +++++- src/submit.rs | 258 ++++++++++++++++++++++++++++++++ 8 files changed, 641 insertions(+), 131 deletions(-) create mode 100644 examples/delivery_smoke_test.rs create mode 100644 src/ingest.rs create mode 100644 src/submit.rs diff --git a/Cargo.lock b/Cargo.lock index 2bc4819..a49d9d0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -234,6 +234,7 @@ dependencies = [ "clap", "ed25519-dalek", "hex", + "logos-delivery", "reqwest", "serde", "serde_json", @@ -328,6 +329,15 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853" +[[package]] +name = "crossbeam-channel" +version = "0.5.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-queue" version = "0.3.12" @@ -982,6 +992,18 @@ version = "0.4.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "953f07c43838f8e6f9758cab68bf5bed85465e7587ebe0b823f1bcd81978ad3a" +[[package]] +name = "logos-delivery" +version = "0.1.0" +dependencies = [ + "base64", + "crossbeam-channel", + "serde", + "serde_json", + "thiserror", + "tracing", +] + [[package]] name = "matchers" version = "0.2.0" diff --git a/Cargo.toml b/Cargo.toml index 0dc0376..6102daf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,10 @@ base64 = "0.22" clap = { version = "4", features = ["derive"] } ed25519-dalek = "2.2.0" hex = "0.4" +# Embedded logos-delivery node for subscription-based ingestion. Path dep on +# the sibling libchat checkout; links the native liblogosdelivery (enter the +# libchat dev shell or set LOGOS_DELIVERY_LIB_DIR to build a runnable binary). +logos-delivery = { path = "../libchat/extensions/logos-delivery-rust" } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" # SQLite via sqlx with a bundled libsqlite3 — no sqlcipher/OpenSSL, no system libs. diff --git a/README.md b/README.md index 8460f5d..cf9dd05 100644 --- a/README.md +++ b/README.md @@ -5,10 +5,21 @@ Persistence for group-chat users' key packages — the **chat-store** HTTP servi [libchat](https://github.com/logos-messaging/libchat) so it can be deployed on its own. -Standalone HTTP service that caches MLS KeyPackages keyed by **`device_id`**, so a +Standalone service that caches MLS KeyPackages keyed by **`device_id`**, so a client can fetch a contact's keypackage without an out-of-band exchange. -Throwaway by design: scheduled to be replaced by a λLEZ-based service in v0.3, so -it intentionally has no overlap with the rest of libchat (axum + rusqlite only). +Throwaway by design: scheduled to be replaced by a λLEZ-based service in v0.3, +with no libchat-core dependency (the embedded logos-delivery node comes from the +sibling libchat checkout's transport crate). + +Submissions arrive on either of two write paths feeding the same verification + +storage pipeline: + +- **HTTP POST** (`/v0/keypackage`, `/v0/account`) — synchronous and acknowledged; +- **logos-delivery subscription** — clients publish the same JSON bodies on the + store's content topics and the server picks them up from the network (see + [Delivery ingestion](#delivery-ingestion)). + +The query API is HTTP only. `device_id` is the hex-encoded 32-byte Ed25519 verifying key of a device. @@ -57,9 +68,33 @@ cargo build --release | `--max-per-identity ` | `100` | Bundles retained per `device_id` | | `--retention-days ` | `30` | Drop bundles older than this | | `--prune-interval-secs ` | `3600` | How often the prune task runs | +| `--no-delivery` | off | Disable the logos-delivery subscriber (HTTP POST ingestion only) | +| `--preset ` | `logos.dev` | logos-delivery network preset the subscriber joins | +| `--p2p-port ` | `0` | TCP + discv5 UDP port for the embedded node (0 = OS-assigned) | Logs via `RUST_LOG` (default `info`). +Building a runnable binary links the native `liblogosdelivery`; enter the +libchat dev shell (`nix develop` in the sibling checkout) or set +`LOGOS_DELIVERY_LIB_DIR` to the directory containing the library. + +## Delivery ingestion + +Unless `--no-delivery` is given, the server runs an embedded logos-delivery +node and subscribes to two content topics: + +```text +/logos-chat/1/store-keypackage-v0/proto keypackage submissions +/logos-chat/1/store-account-v0/proto account device-list submissions +``` + +Each received message is the same JSON body as the corresponding POST endpoint +(below) and goes through identical signature verification and storage rules. +Publishing is fire-and-forget on the client side: rejected submissions are only +logged by the server, which the trust model can afford because consumers verify +every bundle on retrieval anyway. libchat's `DeliveryRegistry` publishes on +these topics when its publish mode is set to delivery. + ## Docker ```bash @@ -191,6 +226,19 @@ GET /v0/account/ -> 200 OK (expect 200) {"payload":...,"signature":... POST /v0/account (replay) -> 409 Conflict (expect 409) ``` +The delivery write path has its own smoke test, +[`delivery_smoke_test`](examples/delivery_smoke_test.rs): it starts a publisher +node, publishes a signed keypackage and account bundle on the store's content +topics, and polls the query API until both appear: + +```bash +# Terminal 1 — start a server (delivery ingestion is on by default) +cargo run -- --bind 127.0.0.1:8080 --db tmp/chat-store.db + +# Terminal 2 — publish over the network and poll the query API +cargo run --example delivery_smoke_test +``` + You can also exercise it with the real `chat-cli` (which lives in the [libchat](https://github.com/logos-messaging/libchat) repo) against a running server: diff --git a/examples/delivery_smoke_test.rs b/examples/delivery_smoke_test.rs new file mode 100644 index 0000000..7e6e153 --- /dev/null +++ b/examples/delivery_smoke_test.rs @@ -0,0 +1,134 @@ +//! End-to-end smoke test for the logos-delivery write path. +//! +//! Publishes a signed keypackage bundle and an account device-list bundle over +//! the delivery network — the same JSON submissions the HTTP POST endpoints +//! take, on the store's subscription topics — then polls the HTTP query API +//! until both appear. Requires a chat-store running with delivery ingestion +//! enabled (the default) on the same network preset. +//! +//! ```text +//! cargo run -- --bind 127.0.0.1:8080 --db tmp/chat-store.db # terminal 1 +//! cargo run --example delivery_smoke_test # terminal 2 +//! cargo run --example delivery_smoke_test -- http://host:port # custom target +//! ``` + +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use anyhow::{Context, Result, bail}; +use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64; +use ed25519_dalek::{Signer, SigningKey}; +use logos_delivery::{P2pConfig, ThreadedDeliveryWrapper}; +use reqwest::blocking::Client; +use serde_json::json; + +/// Must match `ingest::KEYPACKAGE_SUBMIT_TOPIC` / `ingest::ACCOUNT_SUBMIT_TOPIC` +/// (examples cannot import from the binary crate). +const KEYPACKAGE_SUBMIT_TOPIC: &str = "/logos-chat/1/store-keypackage-v0/proto"; +const ACCOUNT_SUBMIT_TOPIC: &str = "/logos-chat/1/store-account-v0/proto"; + +/// Domain-separation prefix the server expects on every account bundle payload. +const ACCOUNT_BUNDLE_DOMAIN: &[u8] = b"libchat:account-device-bundle\0"; + +/// How long to keep polling the query API for the published bundles. Covers +/// node startup, peer discovery, and gossip propagation. +const POLL_BUDGET: Duration = Duration::from_secs(90); + +fn main() -> Result<()> { + let base = std::env::args() + .nth(1) + .unwrap_or_else(|| "http://127.0.0.1:8080".to_string()); + let base = base.trim_end_matches('/').to_string(); + let client = Client::new(); + + println!("Starting logos-delivery publisher node (this takes a few seconds)..."); + let node = ThreadedDeliveryWrapper::>::start(P2pConfig::default(), |_| None) + .map_err(|e| anyhow::anyhow!("start logos-delivery node: {e}"))?; + + // Unique keys per run so re-runs don't hit the account lamport replay guard. + let seed_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64; + let device_key = derived_key(seed_ms, 1); + let account_key = derived_key(seed_ms, 2); + let device_id = hex::encode(device_key.verifying_key().as_bytes()); + let account_pub = hex::encode(account_key.verifying_key().as_bytes()); + + // Keypackage payload: opaque to the server; real clients use + // `timestamp_ms_le[8] || key_package`. + let mut kp_payload = seed_ms.to_le_bytes().to_vec(); + kp_payload.extend_from_slice(b"delivery-smoke-keypackage"); + publish_json( + &node, + KEYPACKAGE_SUBMIT_TOPIC, + &json!({ + "device_id": device_id, + "payload": BASE64.encode(&kp_payload), + "signature": BASE64.encode(device_key.sign(&kp_payload).to_bytes()), + }), + )?; + println!("published keypackage submission on {KEYPACKAGE_SUBMIT_TOPIC}"); + + // Account bundle payload: domain || version || lamport || opaque rest. + let mut acct_payload = ACCOUNT_BUNDLE_DOMAIN.to_vec(); + acct_payload.push(1u8); + acct_payload.extend_from_slice(&1u64.to_le_bytes()); + acct_payload.extend_from_slice(b"delivery-smoke-devices"); + publish_json( + &node, + ACCOUNT_SUBMIT_TOPIC, + &json!({ + "account_pub": account_pub, + "payload": BASE64.encode(&acct_payload), + "signature": BASE64.encode(account_key.sign(&acct_payload).to_bytes()), + }), + )?; + println!("published account submission on {ACCOUNT_SUBMIT_TOPIC}"); + + poll_until_stored( + &client, + &format!("{base}/v0/keypackage/{device_id}"), + "keypackage", + )?; + poll_until_stored(&client, &format!("{base}/v0/account/{account_pub}"), "account")?; + + println!("delivery smoke test passed"); + Ok(()) +} + +fn derived_key(seed_ms: u64, salt: u8) -> SigningKey { + let mut bytes = [salt; 32]; + bytes[..8].copy_from_slice(&seed_ms.to_le_bytes()); + SigningKey::from_bytes(&bytes) +} + +fn publish_json( + node: &ThreadedDeliveryWrapper>, + topic: &str, + body: &serde_json::Value, +) -> Result<()> { + node.publish(topic, body.to_string().as_bytes()) + .map_err(|e| anyhow::anyhow!("publish on {topic}: {e}")) +} + +/// Poll `url` until it returns 200 (the subscriber stored the bundle) or the +/// budget runs out. +fn poll_until_stored(client: &Client, url: &str, what: &str) -> Result<()> { + let started = Instant::now(); + loop { + let status = client + .get(url) + .send() + .with_context(|| format!("GET {url}"))? + .status(); + if status.is_success() { + println!("GET {url} -> {status} ({what} stored, {:?})", started.elapsed()); + return Ok(()); + } + if started.elapsed() > POLL_BUDGET { + bail!("{what} did not appear within {POLL_BUDGET:?} (last status {status})"); + } + std::thread::sleep(Duration::from_secs(2)); + } +} diff --git a/src/handlers.rs b/src/handlers.rs index d2e400c..d7ed86b 100644 --- a/src/handlers.rs +++ b/src/handlers.rs @@ -7,23 +7,10 @@ use axum::routing::{get, post}; use axum::{Json, Router}; use base64::Engine; use base64::engine::general_purpose::STANDARD as BASE64; -use ed25519_dalek::{Signature, VerifyingKey}; -use serde::{Deserialize, Serialize}; +use serde::Serialize; -use crate::store::{Store, StoredAccountBundle, StoredKeyPackageBundle}; - -#[derive(Debug, Deserialize)] -pub struct SubmitRequest { - /// Hex of the 32-byte Ed25519 device verifying key. Used to verify the - /// signature and as the storage/lookup key. `payload` stays opaque. - pub device_id: String, - /// base64 of the signed payload. Opaque to the server — it never decodes it. - pub payload: String, - /// base64 of the 64-byte Ed25519 signature over `payload`. Verifying it - /// under `device_id`'s key is proof-of-possession: only the holder of that - /// key can publish under this `device_id`. - pub signature: String, -} +use crate::store::Store; +use crate::submit::{self, SubmitAccountRequest, SubmitError, SubmitRequest}; #[derive(Debug, Serialize)] pub struct FetchResponse { @@ -46,44 +33,13 @@ pub fn router(store: Arc) -> Router { .with_state(store) } +/// `POST /v0/keypackage` — same submission the logos-delivery subscriber +/// accepts; verification and storage live in [`submit::apply_keypackage`]. async fn submit( State(store): State>, Json(req): Json, ) -> Result { - // Verify proof-of-possession before persisting. `payload` is opaque — the - // server only checks that `signature` over the received payload bytes is - // valid under `device_id`'s key. A valid signature means the submitter holds - // that key. This rejects junk early (DoS mitigation); consumers still verify - // on retrieve, the server is not a trusted authority. - let device_pubkey: [u8; 32] = hex::decode(&req.device_id) - .ok() - .and_then(|b| b.try_into().ok()) - .ok_or_else(|| ApiError::bad("device_id: must be hex of a 32-byte key"))?; - let payload = BASE64 - .decode(&req.payload) - .map_err(|_| ApiError::bad("payload: not valid base64"))?; - let signature: [u8; 64] = BASE64 - .decode(&req.signature) - .ok() - .and_then(|b| b.try_into().ok()) - .ok_or_else(|| ApiError::bad("signature: must be base64 of 64 bytes"))?; - - let verifying_key = VerifyingKey::from_bytes(&device_pubkey) - .map_err(|_| ApiError::bad("device_id: not a valid ed25519 key"))?; - verifying_key - .verify_strict(&payload, &Signature::from_bytes(&signature)) - .map_err(|_| ApiError::bad("signature: verification failed"))?; - - store - .insert( - &req.device_id, - &StoredKeyPackageBundle { - payload, - signature: signature.to_vec(), - }, - ) - .await - .map_err(ApiError::internal)?; + submit::apply_keypackage(&store, &req).await?; Ok(StatusCode::NO_CONTENT) } @@ -104,25 +60,6 @@ async fn fetch( })) } -/// Request body for publishing a signed device-list bundle under an account. -/// -/// The `payload` is intentionally opaque to the server. Clients are expected -/// to encode a lamport-timestamped list of device (LocalIdentity) Ed25519 -/// public keys inside it so that consumers can detect stale bundles. The server -/// only verifies that `signature` is a valid Ed25519 signature over `payload` -/// made by the key identified by `account_pub`. -#[derive(Debug, Deserialize)] -pub struct SubmitAccountRequest { - /// Hex of the 32-byte Ed25519 account (AccountAddress) verifying key. - /// Acts as both the storage key and the verification key. - pub account_pub: String, - /// base64 of the opaque signed payload (lamport-ts + device pubkeys, etc.). - pub payload: String, - /// base64 of the 64-byte Ed25519 signature over `payload` made by the - /// account key. Proof-of-possession: only the account holder can publish. - pub signature: String, -} - #[derive(Debug, Serialize)] pub struct FetchAccountResponse { /// base64 of the stored payload. @@ -137,53 +74,14 @@ pub struct FetchAccountResponse { /// /// The server verifies the Ed25519 signature and then stores exactly one blob /// per `account_pub`, replacing any previous value. Clients should re-publish -/// whenever they add or rotate LocalIdentities. +/// whenever they add or rotate LocalIdentities. Same submission the +/// logos-delivery subscriber accepts; the shared rules live in +/// [`submit::apply_account`]. async fn submit_account( State(store): State>, Json(req): Json, ) -> Result { - let account_pubkey: [u8; 32] = hex::decode(&req.account_pub) - .ok() - .and_then(|b| b.try_into().ok()) - .ok_or_else(|| ApiError::bad("account_pub: must be hex of a 32-byte key"))?; - let payload = BASE64 - .decode(&req.payload) - .map_err(|_| ApiError::bad("payload: not valid base64"))?; - let signature: [u8; 64] = BASE64 - .decode(&req.signature) - .ok() - .and_then(|b| b.try_into().ok()) - .ok_or_else(|| ApiError::bad("signature: must be base64 of 64 bytes"))?; - - let verifying_key = VerifyingKey::from_bytes(&account_pubkey) - .map_err(|_| ApiError::bad("account_pub: not a valid ed25519 key"))?; - verifying_key - .verify_strict(&payload, &Signature::from_bytes(&signature)) - .map_err(|_| ApiError::bad("signature: verification failed"))?; - - // Read the bundle's lamport so the store can reject replays. Safe to trust: - // the signature over `payload` was just verified, so the lamport can't be - // forged without the account key. - let lamport = crate::store::payload_lamport(&payload) - .ok_or_else(|| ApiError::bad("payload: too short to contain a lamport header"))?; - - let applied = store - .upsert_account( - &req.account_pub, - lamport, - &StoredAccountBundle { - payload, - signature: signature.to_vec(), - updated_at: 0, // filled in by store - }, - ) - .await - .map_err(ApiError::internal)?; - if !applied { - return Err(ApiError::conflict( - "stale bundle: lamport is not newer than the stored one", - )); - } + submit::apply_account(&store, &req).await?; Ok(StatusCode::NO_CONTENT) } @@ -215,24 +113,12 @@ struct ApiError { } impl ApiError { - fn bad(msg: impl Into) -> Self { - Self { - status: StatusCode::BAD_REQUEST, - message: msg.into(), - } - } fn not_found(msg: impl Into) -> Self { Self { status: StatusCode::NOT_FOUND, message: msg.into(), } } - fn conflict(msg: impl Into) -> Self { - Self { - status: StatusCode::CONFLICT, - message: msg.into(), - } - } fn internal(err: E) -> Self { tracing::error!("internal: {err}"); Self { @@ -242,6 +128,22 @@ impl ApiError { } } +impl From for ApiError { + fn from(err: SubmitError) -> Self { + match err { + SubmitError::Invalid(msg) => Self { + status: StatusCode::BAD_REQUEST, + message: msg.into(), + }, + SubmitError::Stale => Self { + status: StatusCode::CONFLICT, + message: err.to_string(), + }, + SubmitError::Internal(inner) => Self::internal(inner), + } + } +} + impl IntoResponse for ApiError { fn into_response(self) -> Response { ( diff --git a/src/ingest.rs b/src/ingest.rs new file mode 100644 index 0000000..97a8cfa --- /dev/null +++ b/src/ingest.rs @@ -0,0 +1,102 @@ +//! logos-delivery ingestion. +//! +//! Runs an embedded logos-delivery node, subscribes to the store submission +//! content topics, and feeds every received submission through the same +//! verification + storage pipeline as the HTTP POST endpoints (`submit`). +//! Publishing is fire-and-forget on the client side, so rejected submissions +//! are only logged here — consumers verify every bundle on retrieval anyway. + +use std::sync::Arc; + +use anyhow::{Context, Result}; +use logos_delivery::ThreadedDeliveryWrapper; +pub use logos_delivery::P2pConfig; +use tracing::{debug, warn}; + +use crate::store::Store; +use crate::submit::{self, SubmitAccountRequest, SubmitRequest}; + +/// Content topic carrying keypackage submissions. Must match what libchat's +/// `DeliveryRegistry` publishes: delivery address `store-keypackage-v0` mapped +/// through the `/logos-chat/1/{address}/proto` content-topic scheme. +pub const KEYPACKAGE_SUBMIT_TOPIC: &str = "/logos-chat/1/store-keypackage-v0/proto"; + +/// Content topic carrying account device-list bundle submissions. +pub const ACCOUNT_SUBMIT_TOPIC: &str = "/logos-chat/1/store-account-v0/proto"; + +/// A raw submission taken off the wire; the JSON body is parsed (and its +/// signature verified) on the ingest thread, not the node callback. +#[derive(Clone)] +enum Submission { + KeyPackage(Vec), + Account(Vec), +} + +/// Start the embedded node, subscribe to the submission topics, and spawn the +/// ingest thread. The node lives as long as the returned thread does — i.e. +/// the whole process; there is no shutdown handshake for this testnet service. +/// +/// `runtime` is the server's tokio handle: the store is async, but the +/// delivery wrapper hands messages to a plain thread, so each submission is +/// bridged back with `block_on`. +pub fn start(store: Arc, cfg: P2pConfig, runtime: tokio::runtime::Handle) -> Result<()> { + let mut node = ThreadedDeliveryWrapper::start(cfg, |event| { + let msg = event.into_received()?; + let wrap = match msg.content_topic() { + KEYPACKAGE_SUBMIT_TOPIC => Submission::KeyPackage as fn(Vec) -> Submission, + ACCOUNT_SUBMIT_TOPIC => Submission::Account, + _ => return None, + }; + msg.into_payload().map(wrap) + }) + .context("start embedded logos-delivery node")?; + + node.subscribe(KEYPACKAGE_SUBMIT_TOPIC) + .context("subscribe keypackage submissions")?; + node.subscribe(ACCOUNT_SUBMIT_TOPIC) + .context("subscribe account submissions")?; + + let inbound = node.inbound_queue(); + std::thread::Builder::new() + .name("delivery-ingest".into()) + .spawn(move || { + // Keep the node alive: dropping the last wrapper clone stops it. + let _node = node; + while let Ok(submission) = inbound.recv() { + match submission { + Submission::KeyPackage(bytes) => ingest_keypackage(&store, &runtime, &bytes), + Submission::Account(bytes) => ingest_account(&store, &runtime, &bytes), + } + } + }) + .context("spawn delivery-ingest thread")?; + Ok(()) +} + +fn ingest_keypackage(store: &Store, runtime: &tokio::runtime::Handle, bytes: &[u8]) { + let req: SubmitRequest = match serde_json::from_slice(bytes) { + Ok(req) => req, + Err(e) => { + warn!("keypackage submission: invalid JSON: {e}"); + return; + } + }; + match runtime.block_on(submit::apply_keypackage(store, &req)) { + Ok(()) => debug!(device_id = %req.device_id, "stored keypackage from delivery"), + Err(e) => warn!(device_id = %req.device_id, "keypackage submission rejected: {e}"), + } +} + +fn ingest_account(store: &Store, runtime: &tokio::runtime::Handle, bytes: &[u8]) { + let req: SubmitAccountRequest = match serde_json::from_slice(bytes) { + Ok(req) => req, + Err(e) => { + warn!("account submission: invalid JSON: {e}"); + return; + } + }; + match runtime.block_on(submit::apply_account(store, &req)) { + Ok(()) => debug!(account_pub = %req.account_pub, "stored account bundle from delivery"), + Err(e) => warn!(account_pub = %req.account_pub, "account submission rejected: {e}"), + } +} diff --git a/src/main.rs b/src/main.rs index 79d23f6..1aa931b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,16 +1,25 @@ -//! Testnet KeyPackage Registry HTTP service. +//! Testnet KeyPackage Registry service. //! -//! Throwaway service for issue #110 — replaced by λLEZ in v0.3. Intentionally -//! self-contained: depends only on axum + sqlite + ed25519, no libchat core. +//! Throwaway service for issue #110 — replaced by λLEZ in v0.3. No libchat-core +//! dependency; the embedded logos-delivery node comes from the sibling libchat +//! checkout's transport crate. //! -//! Wire: +//! Submissions arrive on either write path — both feed the same verification + +//! storage pipeline (`submit`): +//! - HTTP POST (below), synchronous and acknowledged; +//! - logos-delivery subscription (`ingest`, disable with `--no-delivery`): +//! clients publish the same JSON on the store content topics. +//! +//! HTTP wire: //! POST /v0/keypackage — submit a signed keypackage bundle //! GET /v0/keypackage/{device_id} — fetch the latest stored keypackage bundle //! POST /v0/account — upsert a signed account device-list bundle //! GET /v0/account/{account_pub} — fetch the account device-list bundle mod handlers; +mod ingest; mod store; +mod submit; use std::net::SocketAddr; use std::path::PathBuf; @@ -45,6 +54,20 @@ struct Cli { /// How often the prune task runs. #[arg(long, default_value_t = 3600)] prune_interval_secs: u64, + + /// Disable the logos-delivery subscriber; submissions then arrive over + /// HTTP POST only. + #[arg(long)] + no_delivery: bool, + + /// logos-delivery network preset the subscriber joins. + #[arg(long, default_value = "logos.dev")] + preset: String, + + /// TCP + discv5 UDP port for the embedded logos-delivery node + /// (0 = OS-assigned). + #[arg(long, default_value_t = 0)] + p2p_port: u16, } #[tokio::main] @@ -80,6 +103,23 @@ async fn main() -> Result<()> { } }); + if !cli.no_delivery { + // Blocks for a few seconds while the node starts and finds peers; + // deliberately before serving HTTP so a subscriber failure is a + // startup error, not a silent half-running store. + ingest::start( + store.clone(), + ingest::P2pConfig { + preset: cli.preset.clone(), + port: cli.p2p_port, + log_level: "ERROR".into(), + }, + tokio::runtime::Handle::current(), + ) + .context("failed to start logos-delivery ingestion")?; + tracing::info!("logos-delivery ingestion running (preset={})", cli.preset); + } + let app = handlers::router(store); let listener = tokio::net::TcpListener::bind(cli.bind) .await diff --git a/src/submit.rs b/src/submit.rs new file mode 100644 index 0000000..3b2e711 --- /dev/null +++ b/src/submit.rs @@ -0,0 +1,258 @@ +//! Shared submission pipeline. +//! +//! Both write paths — the HTTP POST endpoints and the logos-delivery +//! subscriber (`ingest`) — carry the same JSON submissions and feed them +//! through these functions, so signature verification and storage rules are +//! identical no matter which wire delivered the request. + +use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64; +use ed25519_dalek::{Signature, VerifyingKey}; +use serde::Deserialize; + +use crate::store::{Store, StoredAccountBundle, StoredKeyPackageBundle}; + +/// A signed keypackage bundle submission. +#[derive(Debug, Deserialize)] +pub struct SubmitRequest { + /// Hex of the 32-byte Ed25519 device verifying key. Used to verify the + /// signature and as the storage/lookup key. `payload` stays opaque. + pub device_id: String, + /// base64 of the signed payload. Opaque to the server — it never decodes it. + pub payload: String, + /// base64 of the 64-byte Ed25519 signature over `payload`. Verifying it + /// under `device_id`'s key is proof-of-possession: only the holder of that + /// key can publish under this `device_id`. + pub signature: String, +} + +/// A signed account device-list bundle submission. +/// +/// The `payload` is intentionally opaque to the server. Clients are expected +/// to encode a lamport-timestamped list of device (LocalIdentity) Ed25519 +/// public keys inside it so that consumers can detect stale bundles. The server +/// only verifies that `signature` is a valid Ed25519 signature over `payload` +/// made by the key identified by `account_pub`. +#[derive(Debug, Deserialize)] +pub struct SubmitAccountRequest { + /// Hex of the 32-byte Ed25519 account (AccountAddress) verifying key. + /// Acts as both the storage key and the verification key. + pub account_pub: String, + /// base64 of the opaque signed payload (lamport-ts + device pubkeys, etc.). + pub payload: String, + /// base64 of the 64-byte Ed25519 signature over `payload` made by the + /// account key. Proof-of-possession: only the account holder can publish. + pub signature: String, +} + +#[derive(Debug)] +pub enum SubmitError { + /// Malformed submission or failed signature verification. + Invalid(&'static str), + /// Valid account bundle whose lamport is not newer than the stored one. + Stale, + /// Storage failure. + Internal(anyhow::Error), +} + +impl std::fmt::Display for SubmitError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SubmitError::Invalid(msg) => write!(f, "{msg}"), + SubmitError::Stale => { + write!(f, "stale bundle: lamport is not newer than the stored one") + } + SubmitError::Internal(err) => write!(f, "internal: {err}"), + } + } +} + +/// Decoded and signature-verified fields common to both submission kinds. +struct Verified { + payload: Vec, + signature: [u8; 64], +} + +/// Verify proof-of-possession before persisting. `payload` is opaque — the +/// server only checks that `signature` over the received payload bytes is +/// valid under `key_hex`'s key. A valid signature means the submitter holds +/// that key. This rejects junk early (DoS mitigation); consumers still verify +/// on retrieve, the server is not a trusted authority. +fn verify( + key_hex: &str, + payload_b64: &str, + signature_b64: &str, + key_field: KeyField, +) -> Result { + let pubkey: [u8; 32] = hex::decode(key_hex) + .ok() + .and_then(|b| b.try_into().ok()) + .ok_or(SubmitError::Invalid(key_field.not_hex))?; + let payload = BASE64 + .decode(payload_b64) + .map_err(|_| SubmitError::Invalid("payload: not valid base64"))?; + let signature: [u8; 64] = BASE64 + .decode(signature_b64) + .ok() + .and_then(|b| b.try_into().ok()) + .ok_or(SubmitError::Invalid("signature: must be base64 of 64 bytes"))?; + + let verifying_key = + VerifyingKey::from_bytes(&pubkey).map_err(|_| SubmitError::Invalid(key_field.not_key))?; + verifying_key + .verify_strict(&payload, &Signature::from_bytes(&signature)) + .map_err(|_| SubmitError::Invalid("signature: verification failed"))?; + Ok(Verified { payload, signature }) +} + +/// Error messages named after the submission's key field, so HTTP responses +/// and ingest logs point at the right field. +struct KeyField { + not_hex: &'static str, + not_key: &'static str, +} + +const DEVICE_ID: KeyField = KeyField { + not_hex: "device_id: must be hex of a 32-byte key", + not_key: "device_id: not a valid ed25519 key", +}; + +const ACCOUNT_PUB: KeyField = KeyField { + not_hex: "account_pub: must be hex of a 32-byte key", + not_key: "account_pub: not a valid ed25519 key", +}; + +/// Verify and store a keypackage bundle submission. +pub async fn apply_keypackage(store: &Store, req: &SubmitRequest) -> Result<(), SubmitError> { + let verified = verify(&req.device_id, &req.payload, &req.signature, DEVICE_ID)?; + store + .insert( + &req.device_id, + &StoredKeyPackageBundle { + payload: verified.payload, + signature: verified.signature.to_vec(), + }, + ) + .await + .map_err(SubmitError::Internal) +} + +/// Verify and upsert an account device-list bundle submission. +pub async fn apply_account(store: &Store, req: &SubmitAccountRequest) -> Result<(), SubmitError> { + let verified = verify(&req.account_pub, &req.payload, &req.signature, ACCOUNT_PUB)?; + + // Read the bundle's lamport so the store can reject replays. Safe to trust: + // the signature over `payload` was just verified, so the lamport can't be + // forged without the account key. + let lamport = crate::store::payload_lamport(&verified.payload).ok_or(SubmitError::Invalid( + "payload: too short to contain a lamport header", + ))?; + + let applied = store + .upsert_account( + &req.account_pub, + lamport, + &StoredAccountBundle { + payload: verified.payload, + signature: verified.signature.to_vec(), + updated_at: 0, // filled in by store + }, + ) + .await + .map_err(SubmitError::Internal)?; + if !applied { + return Err(SubmitError::Stale); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::path::Path; + + use ed25519_dalek::{Signer, SigningKey}; + use serde_json::json; + + use super::*; + + /// Must match `BUNDLE_DOMAIN` in `store.rs` (kept private there). + const ACCOUNT_BUNDLE_DOMAIN: &[u8] = b"libchat:account-device-bundle\0"; + + fn signing_key(seed: u8) -> SigningKey { + SigningKey::from_bytes(&[seed; 32]) + } + + /// A submission exactly as it arrives off the delivery wire: the JSON body + /// libchat's `DeliveryRegistry` publishes (same shape as the POST body). + fn keypackage_submission_json(key: &SigningKey, payload: &[u8]) -> Vec { + json!({ + "device_id": hex::encode(key.verifying_key().as_bytes()), + "payload": BASE64.encode(payload), + "signature": BASE64.encode(key.sign(payload).to_bytes()), + }) + .to_string() + .into_bytes() + } + + fn account_payload(lamport: u64) -> Vec { + let mut p = ACCOUNT_BUNDLE_DOMAIN.to_vec(); + p.push(1u8); // version + p.extend_from_slice(&lamport.to_le_bytes()); + p + } + + fn account_submission(key: &SigningKey, payload: &[u8]) -> SubmitAccountRequest { + SubmitAccountRequest { + account_pub: hex::encode(key.verifying_key().as_bytes()), + payload: BASE64.encode(payload), + signature: BASE64.encode(key.sign(payload).to_bytes()), + } + } + + #[tokio::test] + async fn wire_json_keypackage_is_parsed_verified_and_stored() { + let store = Store::open(Path::new(":memory:")).await.unwrap(); + let key = signing_key(1); + let payload = b"ts-and-keypackage-bytes".to_vec(); + + let wire = keypackage_submission_json(&key, &payload); + let req: SubmitRequest = serde_json::from_slice(&wire).unwrap(); + apply_keypackage(&store, &req).await.unwrap(); + + let stored = store.latest(&req.device_id).await.unwrap().unwrap(); + assert_eq!(stored.payload, payload); + } + + #[tokio::test] + async fn tampered_keypackage_submission_is_rejected() { + let store = Store::open(Path::new(":memory:")).await.unwrap(); + let key = signing_key(2); + + let mut req: SubmitRequest = + serde_json::from_slice(&keypackage_submission_json(&key, b"original")).unwrap(); + req.payload = BASE64.encode(b"tampered"); + + let err = apply_keypackage(&store, &req).await.unwrap_err(); + assert!(matches!(err, SubmitError::Invalid("signature: verification failed"))); + assert!(store.latest(&req.device_id).await.unwrap().is_none()); + } + + #[tokio::test] + async fn account_submission_upserts_and_rejects_stale_replay() { + let store = Store::open(Path::new(":memory:")).await.unwrap(); + let key = signing_key(3); + + apply_account(&store, &account_submission(&key, &account_payload(1))) + .await + .unwrap(); + apply_account(&store, &account_submission(&key, &account_payload(2))) + .await + .unwrap(); + + // Replaying the lamport-2 bundle (as a delivery duplicate would) is stale. + let err = apply_account(&store, &account_submission(&key, &account_payload(2))) + .await + .unwrap_err(); + assert!(matches!(err, SubmitError::Stale)); + } +}