From b9a2081075b3f1c2f33ae191cf41bc24ef8a1fb5 Mon Sep 17 00:00:00 2001 From: kaichaosun Date: Mon, 27 Jul 2026 14:16:09 +0800 Subject: [PATCH] feat: refactor modules splition --- Cargo.toml | 5 - examples/delivery_smoke_test.rs | 13 +- examples/smoke_test.rs | 19 +- src/bundle.rs | 277 ++++++++++++++++++++++ src/{ingest.rs => delivery.rs} | 64 ++--- src/handlers.rs | 111 +++++++-- src/main.rs | 29 ++- src/submit.rs | 404 -------------------------------- 8 files changed, 434 insertions(+), 488 deletions(-) create mode 100644 src/bundle.rs rename src/{ingest.rs => delivery.rs} (59%) delete mode 100644 src/submit.rs diff --git a/Cargo.toml b/Cargo.toml index d84e850..fa3af16 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,15 +15,10 @@ path = "src/main.rs" anyhow = "1.0" axum = "0.7" base64 = "0.22" -# Wire schema for the delivery submissions, shared with libchat. Pinned to a rev -# rather than a branch so the wire format can only change deliberately. chat-proto = { git = "https://github.com/logos-messaging/chat_proto", rev = "e05c21229a18bab655c9aec06409189447d24a6b" } clap = { version = "4", features = ["derive"] } ed25519-dalek = "2.2.0" hex = "0.4" -# Embedded logos-delivery node for subscription-based ingestion. Links the -# native liblogosdelivery, which this repo's flake builds: `nix develop` exports -# LOGOS_DELIVERY_LIB_DIR, without which the binary fails at link. logos-delivery = { git = "https://github.com/logos-messaging/libchat", rev = "5c55c2ee76bebd1dbd5eb4bfa95d9e71acd0fa14" } prost = "0.14" serde = { version = "1.0", features = ["derive"] } diff --git a/examples/delivery_smoke_test.rs b/examples/delivery_smoke_test.rs index 924cb99..5fca756 100644 --- a/examples/delivery_smoke_test.rs +++ b/examples/delivery_smoke_test.rs @@ -22,7 +22,7 @@ use prost::Message; use prost::bytes::Bytes; use reqwest::blocking::Client; -/// Must match `ingest::KEYPACKAGE_SUBMIT_TOPIC` / `ingest::ACCOUNT_SUBMIT_TOPIC` +/// Must match `delivery::KEYPACKAGE_SUBMIT_TOPIC` / `delivery::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"; @@ -91,7 +91,11 @@ fn main() -> Result<()> { &format!("{base}/v0/keypackage/{device_id}"), "keypackage", )?; - poll_until_stored(&client, &format!("{base}/v0/account/{account_pub}"), "account")?; + poll_until_stored( + &client, + &format!("{base}/v0/account/{account_pub}"), + "account", + )?; println!("delivery smoke test passed"); Ok(()) @@ -123,7 +127,10 @@ fn poll_until_stored(client: &Client, url: &str, what: &str) -> Result<()> { .with_context(|| format!("GET {url}"))? .status(); if status.is_success() { - println!("GET {url} -> {status} ({what} stored, {:?})", started.elapsed()); + println!( + "GET {url} -> {status} ({what} stored, {:?})", + started.elapsed() + ); return Ok(()); } if started.elapsed() > POLL_BUDGET { diff --git a/examples/smoke_test.rs b/examples/smoke_test.rs index 92738e7..a9613a3 100644 --- a/examples/smoke_test.rs +++ b/examples/smoke_test.rs @@ -56,7 +56,10 @@ fn test_keypackage(client: &Client, base: &str) -> Result<()> { "signature": BASE64.encode(key.sign(&payload).to_bytes()), })) .send()?; - println!("POST /v0/keypackage -> {} (expect 204)", resp.status()); + println!( + "POST /v0/keypackage -> {} (expect 204)", + resp.status() + ); let resp = client .get(format!("{base}/v0/keypackage/{device_id}")) @@ -91,9 +94,14 @@ fn test_account(client: &Client, base: &str) -> Result<()> { .post(format!("{base}/v0/account")) .json(&body) .send()?; - println!("POST /v0/account -> {} (expect 204)", resp.status()); + println!( + "POST /v0/account -> {} (expect 204)", + resp.status() + ); - let resp = client.get(format!("{base}/v0/account/{account_pub}")).send()?; + let resp = client + .get(format!("{base}/v0/account/{account_pub}")) + .send()?; println!( "GET /v0/account/ -> {} (expect 200) {}", resp.status(), @@ -105,7 +113,10 @@ fn test_account(client: &Client, base: &str) -> Result<()> { .post(format!("{base}/v0/account")) .json(&body) .send()?; - println!("POST /v0/account (replay) -> {} (expect 409)", resp.status()); + println!( + "POST /v0/account (replay) -> {} (expect 409)", + resp.status() + ); Ok(()) } diff --git a/src/bundle.rs b/src/bundle.rs new file mode 100644 index 0000000..debc734 --- /dev/null +++ b/src/bundle.rs @@ -0,0 +1,277 @@ +//! Bundles: an opaque payload plus its signature, published under the key that +//! signed it. This is what the store holds and what both write paths carry. +//! +//! Each wire encodes a bundle in its own way — the HTTP POST endpoints take a +//! JSON body with hex and base64 strings, the logos-delivery subscriber +//! (`delivery`) takes a protobuf message with raw bytes — and both decode into +//! a [`Bundle`]. Verification and storage rules below are therefore identical +//! no matter which wire carried it. + +use ed25519_dalek::{Signature, VerifyingKey}; + +use crate::store::{Store, StoredAccountBundle, StoredKeyPackageBundle}; + +#[derive(Debug)] +pub enum BundleError { + /// Malformed bundle 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 BundleError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + BundleError::Invalid(msg) => write!(f, "{msg}"), + BundleError::Stale => { + write!(f, "stale bundle: lamport is not newer than the stored one") + } + BundleError::Internal(err) => write!(f, "internal: {err}"), + } + } +} + +/// A bundle decoded off either wire, before verification: the raw key, +/// payload and signature bytes that both encodings ultimately carry. +#[derive(Debug)] +pub struct Bundle { + key: [u8; 32], + payload: Vec, + signature: [u8; 64], +} + +impl Bundle { + /// From raw bytes, as the protobuf wire carries them. Keypackage and + /// account bundles are the same shape here — which kind it is comes from + /// the topic it arrived on and the `apply_*` it is passed to. + pub fn from_bytes(key: &[u8], payload: &[u8], signature: &[u8]) -> Result { + Ok(Self { + key: key + .try_into() + .map_err(|_| BundleError::Invalid("key: must be 32 bytes"))?, + payload: payload.to_vec(), + signature: signature + .try_into() + .map_err(|_| BundleError::Invalid("signature: must be 64 bytes"))?, + }) + } + + /// Hex of the bundle's key — the store's lookup key. Canonical + /// lowercase, so the same key always lands in the same row whichever wire + /// and whichever casing it arrived in. + pub fn key_hex(&self) -> String { + hex::encode(self.key) + } +} + +/// Verify proof-of-possession before persisting. `payload` is opaque — the +/// server only checks that `signature` over the received payload bytes is +/// valid under the bundle'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(bundle: &Bundle) -> Result<(), BundleError> { + let verifying_key = VerifyingKey::from_bytes(&bundle.key) + .map_err(|_| BundleError::Invalid("key: not a valid ed25519 key"))?; + verifying_key + .verify_strict(&bundle.payload, &Signature::from_bytes(&bundle.signature)) + .map_err(|_| BundleError::Invalid("signature: verification failed"))?; + Ok(()) +} + +/// Verify and store a keypackage bundle. +pub async fn apply_keypackage(store: &Store, bundle: &Bundle) -> Result<(), BundleError> { + verify(bundle)?; + store + .insert( + &bundle.key_hex(), + &StoredKeyPackageBundle { + payload: bundle.payload.clone(), + signature: bundle.signature.to_vec(), + }, + ) + .await + .map_err(BundleError::Internal) +} + +/// Verify and upsert an account device-list bundle. +pub async fn apply_account(store: &Store, bundle: &Bundle) -> Result<(), BundleError> { + verify(bundle)?; + + // 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(&bundle.payload).ok_or(BundleError::Invalid( + "payload: too short to contain a lamport header", + ))?; + + let applied = store + .upsert_account( + &bundle.key_hex(), + lamport, + &StoredAccountBundle { + payload: bundle.payload.clone(), + signature: bundle.signature.to_vec(), + updated_at: 0, // filled in by store + }, + ) + .await + .map_err(BundleError::Internal)?; + if !applied { + return Err(BundleError::Stale); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::path::Path; + + use base64::Engine; + use base64::engine::general_purpose::STANDARD as BASE64; + use chat_proto::logoschat::store::{AccountSubmissionV1, KeyPackageSubmissionV1}; + use ed25519_dalek::{Signer, SigningKey}; + use prost::Message; + use prost::bytes::Bytes; + use serde_json::json; + + use super::*; + // The JSON body lives with the HTTP wire that owns it; this test asserts + // that wire and the protobuf one decode to the same bundle. + use crate::handlers::SubmitKeyPackageRequest; + + /// 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 keypackage bundle exactly as it arrives off the delivery wire: + /// the protobuf message libchat's `DeliveryRegistry` publishes. + fn keypackage_submission_proto(key: &SigningKey, payload: &[u8]) -> Vec { + KeyPackageSubmissionV1 { + device_id: Bytes::copy_from_slice(key.verifying_key().as_bytes()), + payload: Bytes::copy_from_slice(payload), + signature: Bytes::copy_from_slice(&key.sign(payload).to_bytes()), + } + .encode_to_vec() + } + + /// The same bundle as the HTTP 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]) -> Bundle { + Bundle::from_bytes( + key.verifying_key().as_bytes(), + payload, + &key.sign(payload).to_bytes(), + ) + .unwrap() + } + + /// The protobuf wire and the JSON body must decode to the same bundle, + /// so the store applies identical rules whichever path delivered it. + #[tokio::test] + async fn proto_and_json_keypackage_wires_agree() { + let key = signing_key(1); + let payload = b"ts-and-keypackage-bytes".to_vec(); + + let wire = KeyPackageSubmissionV1::decode(&keypackage_submission_proto(&key, &payload)[..]) + .unwrap(); + let from_proto = + Bundle::from_bytes(&wire.device_id, &wire.payload, &wire.signature).unwrap(); + + let body: SubmitKeyPackageRequest = + serde_json::from_slice(&keypackage_submission_json(&key, &payload)).unwrap(); + let from_json = body.decode().unwrap(); + + assert_eq!(from_proto.key_hex(), from_json.key_hex()); + assert_eq!(from_proto.payload, from_json.payload); + assert_eq!(from_proto.signature, from_json.signature); + } + + #[tokio::test] + async fn wire_proto_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 = KeyPackageSubmissionV1::decode(&keypackage_submission_proto(&key, &payload)[..]) + .unwrap(); + let bundle = Bundle::from_bytes(&wire.device_id, &wire.payload, &wire.signature).unwrap(); + apply_keypackage(&store, &bundle).await.unwrap(); + + let stored = store.latest(&bundle.key_hex()).await.unwrap().unwrap(); + assert_eq!(stored.payload, payload); + } + + #[tokio::test] + async fn tampered_keypackage_bundle_is_rejected() { + let store = Store::open(Path::new(":memory:")).await.unwrap(); + let key = signing_key(2); + + let mut wire = + KeyPackageSubmissionV1::decode(&keypackage_submission_proto(&key, b"original")[..]) + .unwrap(); + wire.payload = Bytes::from_static(b"tampered"); + let bundle = Bundle::from_bytes(&wire.device_id, &wire.payload, &wire.signature).unwrap(); + + let err = apply_keypackage(&store, &bundle).await.unwrap_err(); + assert!(matches!( + err, + BundleError::Invalid("signature: verification failed") + )); + assert!(store.latest(&bundle.key_hex()).await.unwrap().is_none()); + } + + /// A truncated or foreign payload on the bundle topics must be dropped, + /// not panic the ingest thread. + #[test] + fn non_protobuf_bytes_are_rejected() { + assert!(KeyPackageSubmissionV1::decode(&b"{\"device_id\":\"xx\"}"[..]).is_err()); + assert!(AccountSubmissionV1::decode(&[0xff, 0xff, 0xff][..]).is_err()); + } + + /// A key that is not 32 bytes is rejected at decode, before verification. + #[test] + fn short_key_is_rejected() { + let err = Bundle::from_bytes(&[1u8; 31], b"payload", &[0u8; 64]).unwrap_err(); + assert!(matches!(err, BundleError::Invalid("key: must be 32 bytes"))); + } + + #[tokio::test] + async fn account_bundle_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, BundleError::Stale)); + } +} diff --git a/src/ingest.rs b/src/delivery.rs similarity index 59% rename from src/ingest.rs rename to src/delivery.rs index 2fd90ed..0f9f1da 100644 --- a/src/ingest.rs +++ b/src/delivery.rs @@ -1,13 +1,13 @@ -//! logos-delivery ingestion. +//! The logos-delivery write path. //! -//! 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`). -//! Submissions arrive as protobuf on this wire — matching the `/proto` content -//! topics they are published on — and carry the same fields the JSON POST -//! bodies do. Publishing is fire-and-forget on the client side, so rejected -//! submissions are only logged here; consumers verify every bundle on -//! retrieval anyway. +//! Runs an embedded logos-delivery node, subscribes to the store's content +//! topics, and feeds every bundle it receives through the same verification + +//! storage pipeline as the HTTP POST endpoints (`bundle`). Bundles arrive as +//! protobuf on this wire — matching the `/proto` content topics they are +//! published on — carrying the same fields the JSON POST bodies do. +//! +//! Publishing is fire-and-forget on the client side, so rejected bundles are +//! only logged here; consumers verify every bundle on retrieval anyway. use std::sync::Arc; @@ -18,18 +18,18 @@ use logos_delivery::ThreadedDeliveryWrapper; use prost::Message; use tracing::{debug, warn}; +use crate::bundle::{Bundle, apply_account, apply_keypackage}; use crate::store::Store; -use crate::submit::{self, Submission}; -/// Content topic carrying keypackage submissions. Must match what libchat's -/// `DeliveryRegistry` publishes: delivery address `store-keypackage-v0` mapped +/// Content topic carrying keypackage bundles. Must match what libchat's +/// `ContactRegistry` 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. +/// Content topic carrying account device-list bundles. pub const ACCOUNT_SUBMIT_TOPIC: &str = "/logos-chat/1/store-account-v0/proto"; -/// A raw submission taken off the wire; the protobuf body is decoded (and its +/// A raw bundle taken off the wire; the protobuf body is decoded (and its /// signature verified) on the ingest thread, not the node callback. #[derive(Clone)] enum Received { @@ -37,12 +37,12 @@ enum Received { Account(Vec), } -/// Start the embedded node, subscribe to the submission topics, and spawn the +/// Start the embedded node, subscribe to the bundle 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 +/// delivery wrapper hands messages to a plain thread, so each bundle 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| { @@ -57,9 +57,9 @@ pub fn start(store: Arc, cfg: P2pConfig, runtime: tokio::runtime::Handle) .context("start embedded logos-delivery node")?; node.subscribe(KEYPACKAGE_SUBMIT_TOPIC) - .context("subscribe keypackage submissions")?; + .context("subscribe keypackage bundles")?; node.subscribe(ACCOUNT_SUBMIT_TOPIC) - .context("subscribe account submissions")?; + .context("subscribe account bundles")?; let inbound = node.inbound_queue(); std::thread::Builder::new() @@ -82,21 +82,21 @@ fn ingest_keypackage(store: &Store, runtime: &tokio::runtime::Handle, bytes: &[u let wire = match KeyPackageSubmissionV1::decode(bytes) { Ok(wire) => wire, Err(e) => { - warn!("keypackage submission: invalid protobuf: {e}"); + warn!("keypackage bundle: invalid protobuf: {e}"); return; } }; - let submission = match Submission::keypackage(&wire.device_id, &wire.payload, &wire.signature) { - Ok(submission) => submission, + let bundle = match Bundle::from_bytes(&wire.device_id, &wire.payload, &wire.signature) { + Ok(bundle) => bundle, Err(e) => { - warn!("keypackage submission rejected: {e}"); + warn!("keypackage bundle rejected: {e}"); return; } }; - let device_id = submission.key_hex(); - match runtime.block_on(submit::apply_keypackage(store, &submission)) { + let device_id = bundle.key_hex(); + match runtime.block_on(apply_keypackage(store, &bundle)) { Ok(()) => debug!(%device_id, "stored keypackage from delivery"), - Err(e) => warn!(%device_id, "keypackage submission rejected: {e}"), + Err(e) => warn!(%device_id, "keypackage bundle rejected: {e}"), } } @@ -104,20 +104,20 @@ fn ingest_account(store: &Store, runtime: &tokio::runtime::Handle, bytes: &[u8]) let wire = match AccountSubmissionV1::decode(bytes) { Ok(wire) => wire, Err(e) => { - warn!("account submission: invalid protobuf: {e}"); + warn!("account bundle: invalid protobuf: {e}"); return; } }; - let submission = match Submission::account(&wire.account_pub, &wire.payload, &wire.signature) { - Ok(submission) => submission, + let bundle = match Bundle::from_bytes(&wire.account_pub, &wire.payload, &wire.signature) { + Ok(bundle) => bundle, Err(e) => { - warn!("account submission rejected: {e}"); + warn!("account bundle rejected: {e}"); return; } }; - let account_pub = submission.key_hex(); - match runtime.block_on(submit::apply_account(store, &submission)) { + let account_pub = bundle.key_hex(); + match runtime.block_on(apply_account(store, &bundle)) { Ok(()) => debug!(%account_pub, "stored account bundle from delivery"), - Err(e) => warn!(%account_pub, "account submission rejected: {e}"), + Err(e) => warn!(%account_pub, "account bundle rejected: {e}"), } } diff --git a/src/handlers.rs b/src/handlers.rs index 8017bdb..768f09a 100644 --- a/src/handlers.rs +++ b/src/handlers.rs @@ -7,13 +7,78 @@ use axum::routing::{get, post}; use axum::{Json, Router}; use base64::Engine; use base64::engine::general_purpose::STANDARD as BASE64; -use serde::Serialize; +use serde::{Deserialize, Serialize}; +use crate::bundle::{self, Bundle, BundleError}; use crate::store::Store; -use crate::submit::{self, SubmitAccountRequest, SubmitError, SubmitRequest}; + +/// A signed keypackage bundle. +#[derive(Debug, Deserialize)] +pub struct SubmitKeyPackageRequest { + /// 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, +} + +impl SubmitKeyPackageRequest { + /// Decode the JSON body's hex + base64 fields into a [`Bundle`]. + pub fn decode(&self) -> Result { + decode_bundle(&self.device_id, &self.payload, &self.signature) + } +} + +/// A signed account device-list bundle. +/// +/// 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, +} + +impl SubmitAccountRequest { + /// Decode the JSON body's hex + base64 fields into a [`Bundle`]. + pub fn decode(&self) -> Result { + decode_bundle(&self.account_pub, &self.payload, &self.signature) + } +} + +/// This wire spells a bundle out in text: the key as hex, payload and signature +/// as base64. Undo that here — the delivery wire hands over raw bytes already — +/// and let [`Bundle::from_bytes`] enforce the lengths. +fn decode_bundle( + key_hex: &str, + payload_b64: &str, + signature_b64: &str, +) -> Result { + let key = hex::decode(key_hex).map_err(|_| BundleError::Invalid("key: must be hex"))?; + let payload = BASE64 + .decode(payload_b64) + .map_err(|_| BundleError::Invalid("payload: not valid base64"))?; + let signature = BASE64 + .decode(signature_b64) + .map_err(|_| BundleError::Invalid("signature: not valid base64"))?; + Bundle::from_bytes(&key, &payload, &signature) +} #[derive(Debug, Serialize)] -pub struct FetchResponse { +pub struct FetchKeyPackageResponse { /// base64 of the stored payload; consumers verify `signature` over it. pub payload: String, pub signature: String, @@ -26,36 +91,32 @@ struct ErrorBody { pub fn router(store: Arc) -> Router { Router::new() - .route("/v0/keypackage", post(submit)) - .route("/v0/keypackage/:device_id", get(fetch)) + .route("/v0/keypackage", post(submit_key_package)) + .route("/v0/keypackage/:device_id", get(fetch_key_package)) .route("/v0/account", post(submit_account)) .route("/v0/account/:account_pub", get(fetch_account)) .with_state(store) } -/// `POST /v0/keypackage` — the same submission the logos-delivery subscriber +/// `POST /v0/keypackage` — the same bundle the logos-delivery subscriber /// accepts, in JSON rather than protobuf; verification and storage live in -/// [`submit::apply_keypackage`]. -async fn submit( +/// [`bundle::apply_keypackage`]. +async fn submit_key_package( State(store): State>, - Json(req): Json, + Json(req): Json, ) -> Result { - submit::apply_keypackage(&store, &req.decode()?).await?; + bundle::apply_keypackage(&store, &req.decode()?).await?; Ok(StatusCode::NO_CONTENT) } -async fn fetch( +async fn fetch_key_package( State(store): State>, Path(device_id): Path, -) -> Result, ApiError> { - let Some(bundle) = store - .latest(&device_id) - .await - .map_err(ApiError::internal)? - else { +) -> Result, ApiError> { + let Some(bundle) = store.latest(&device_id).await.map_err(ApiError::internal)? else { return Err(ApiError::not_found("no keypackage for device")); }; - Ok(Json(FetchResponse { + Ok(Json(FetchKeyPackageResponse { payload: BASE64.encode(&bundle.payload), signature: BASE64.encode(&bundle.signature), })) @@ -77,12 +138,12 @@ pub struct FetchAccountResponse { /// per `account_pub`, replacing any previous value. Clients should re-publish /// whenever they add or rotate LocalIdentities. The same submission the /// logos-delivery subscriber accepts, in JSON rather than protobuf; the shared -/// rules live in [`submit::apply_account`]. +/// rules live in [`bundle::apply_account`]. async fn submit_account( State(store): State>, Json(req): Json, ) -> Result { - submit::apply_account(&store, &req.decode()?).await?; + bundle::apply_account(&store, &req.decode()?).await?; Ok(StatusCode::NO_CONTENT) } @@ -129,18 +190,18 @@ impl ApiError { } } -impl From for ApiError { - fn from(err: SubmitError) -> Self { +impl From for ApiError { + fn from(err: BundleError) -> Self { match err { - SubmitError::Invalid(msg) => Self { + BundleError::Invalid(msg) => Self { status: StatusCode::BAD_REQUEST, message: msg.into(), }, - SubmitError::Stale => Self { + BundleError::Stale => Self { status: StatusCode::CONFLICT, message: err.to_string(), }, - SubmitError::Internal(inner) => Self::internal(inner), + BundleError::Internal(inner) => Self::internal(inner), } } } diff --git a/src/main.rs b/src/main.rs index 1aa931b..da53754 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,14 +1,14 @@ //! Testnet KeyPackage Registry service. //! //! 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. +//! dependency; the embedded logos-delivery node comes from libchat's transport +//! crate, pulled in as a pinned git dependency. //! //! Submissions arrive on either write path — both feed the same verification + -//! storage pipeline (`submit`): +//! storage pipeline (`bundle`): //! - HTTP POST (below), synchronous and acknowledged; -//! - logos-delivery subscription (`ingest`, disable with `--no-delivery`): -//! clients publish the same JSON on the store content topics. +//! - logos-delivery subscription (`delivery`, disable with `--no-delivery`): +//! clients publish protobuf submissions on the store content topics. //! //! HTTP wire: //! POST /v0/keypackage — submit a signed keypackage bundle @@ -16,10 +16,10 @@ //! POST /v0/account — upsert a signed account device-list bundle //! GET /v0/account/{account_pub} — fetch the account device-list bundle +mod bundle; +mod delivery; mod handlers; -mod ingest; mod store; -mod submit; use std::net::SocketAddr; use std::path::PathBuf; @@ -33,7 +33,10 @@ use tracing_subscriber::EnvFilter; use store::Store; #[derive(Parser, Debug)] -#[command(name = "chat-store", about = "Testnet Chat Store (KeyPackage + account directory)")] +#[command( + name = "chat-store", + about = "Testnet Chat Store (KeyPackage + account directory)" +)] struct Cli { /// Address to bind the HTTP server. #[arg(long, default_value = "0.0.0.0:8080")] @@ -80,11 +83,7 @@ async fn main() -> Result<()> { let cli = Cli::parse(); - let store = Arc::new( - Store::open(&cli.db) - .await - .context("failed to open store")?, - ); + let store = Arc::new(Store::open(&cli.db).await.context("failed to open store")?); let prune_store = store.clone(); let max_per_id = cli.max_per_identity; @@ -107,9 +106,9 @@ async fn main() -> Result<()> { // 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( + delivery::start( store.clone(), - ingest::P2pConfig { + delivery::P2pConfig { preset: cli.preset.clone(), port: cli.p2p_port, log_level: "ERROR".into(), diff --git a/src/submit.rs b/src/submit.rs deleted file mode 100644 index 6c3c8ef..0000000 --- a/src/submit.rs +++ /dev/null @@ -1,404 +0,0 @@ -//! Shared submission pipeline. -//! -//! The two write paths carry the same fields in the encoding native to each -//! wire: the HTTP POST endpoints take a JSON body with hex and base64 strings, -//! while the logos-delivery subscriber (`ingest`) takes a protobuf message -//! with raw bytes. Both decode into a [`Submission`] and feed it 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, -} - -impl SubmitRequest { - /// Decode the JSON body's hex + base64 fields into a [`Submission`]. - pub fn decode(&self) -> Result { - Submission::from_encoded(&self.device_id, &self.payload, &self.signature, DEVICE_ID) - } -} - -/// 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, -} - -impl SubmitAccountRequest { - /// Decode the JSON body's hex + base64 fields into a [`Submission`]. - pub fn decode(&self) -> Result { - Submission::from_encoded( - &self.account_pub, - &self.payload, - &self.signature, - ACCOUNT_PUB, - ) - } -} - -#[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}"), - } - } -} - -/// A submission decoded off either wire, before verification: the raw key, -/// payload and signature bytes that both encodings ultimately carry. -#[derive(Debug)] -pub struct Submission { - key: [u8; 32], - payload: Vec, - signature: [u8; 64], -} - -impl Submission { - /// A keypackage submission from raw bytes, as the protobuf wire carries them. - pub fn keypackage( - device_id: &[u8], - payload: &[u8], - signature: &[u8], - ) -> Result { - Self::from_bytes(device_id, payload, signature, DEVICE_ID) - } - - /// An account device-list submission from raw bytes. - pub fn account( - account_pub: &[u8], - payload: &[u8], - signature: &[u8], - ) -> Result { - Self::from_bytes(account_pub, payload, signature, ACCOUNT_PUB) - } - - fn from_bytes( - key: &[u8], - payload: &[u8], - signature: &[u8], - key_field: KeyField, - ) -> Result { - Ok(Self { - key: key - .try_into() - .map_err(|_| SubmitError::Invalid(key_field.not_32_bytes))?, - payload: payload.to_vec(), - signature: signature - .try_into() - .map_err(|_| SubmitError::Invalid("signature: must be 64 bytes"))?, - }) - } - - /// From the HTTP body's hex key and base64 payload/signature. - fn from_encoded( - key_hex: &str, - payload_b64: &str, - signature_b64: &str, - key_field: KeyField, - ) -> Result { - let key: [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", - ))?; - Ok(Self { - key, - payload, - signature, - }) - } - - /// Hex of the submission's key — the store's lookup key. Canonical - /// lowercase, so the same key always lands in the same row whichever wire - /// and whichever casing it arrived in. - pub fn key_hex(&self) -> String { - hex::encode(self.key) - } -} - -/// Verify proof-of-possession before persisting. `payload` is opaque — the -/// server only checks that `signature` over the received payload bytes is -/// valid under the submission'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(submission: &Submission, key_field: KeyField) -> Result<(), SubmitError> { - let verifying_key = VerifyingKey::from_bytes(&submission.key) - .map_err(|_| SubmitError::Invalid(key_field.not_key))?; - verifying_key - .verify_strict( - &submission.payload, - &Signature::from_bytes(&submission.signature), - ) - .map_err(|_| SubmitError::Invalid("signature: verification failed"))?; - Ok(()) -} - -/// Error messages named after the submission's key field, so HTTP responses -/// and ingest logs point at the right field. -#[derive(Clone, Copy)] -struct KeyField { - not_hex: &'static str, - not_32_bytes: &'static str, - not_key: &'static str, -} - -const DEVICE_ID: KeyField = KeyField { - not_hex: "device_id: must be hex of a 32-byte key", - not_32_bytes: "device_id: must be 32 bytes", - 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_32_bytes: "account_pub: must be 32 bytes", - not_key: "account_pub: not a valid ed25519 key", -}; - -/// Verify and store a keypackage bundle submission. -pub async fn apply_keypackage(store: &Store, submission: &Submission) -> Result<(), SubmitError> { - verify(submission, DEVICE_ID)?; - store - .insert( - &submission.key_hex(), - &StoredKeyPackageBundle { - payload: submission.payload.clone(), - signature: submission.signature.to_vec(), - }, - ) - .await - .map_err(SubmitError::Internal) -} - -/// Verify and upsert an account device-list bundle submission. -pub async fn apply_account(store: &Store, submission: &Submission) -> Result<(), SubmitError> { - verify(submission, 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(&submission.payload).ok_or(SubmitError::Invalid( - "payload: too short to contain a lamport header", - ))?; - - let applied = store - .upsert_account( - &submission.key_hex(), - lamport, - &StoredAccountBundle { - payload: submission.payload.clone(), - signature: submission.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 chat_proto::logoschat::store::{AccountSubmissionV1, KeyPackageSubmissionV1}; - use ed25519_dalek::{Signer, SigningKey}; - use prost::Message; - use prost::bytes::Bytes; - 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 keypackage submission exactly as it arrives off the delivery wire: - /// the protobuf message libchat's `DeliveryRegistry` publishes. - fn keypackage_submission_proto(key: &SigningKey, payload: &[u8]) -> Vec { - KeyPackageSubmissionV1 { - device_id: Bytes::copy_from_slice(key.verifying_key().as_bytes()), - payload: Bytes::copy_from_slice(payload), - signature: Bytes::copy_from_slice(&key.sign(payload).to_bytes()), - } - .encode_to_vec() - } - - /// The same submission as the HTTP 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]) -> Submission { - Submission::account( - key.verifying_key().as_bytes(), - payload, - &key.sign(payload).to_bytes(), - ) - .unwrap() - } - - /// The protobuf wire and the JSON body must decode to the same submission, - /// so the store applies identical rules whichever path delivered it. - #[tokio::test] - async fn proto_and_json_keypackage_wires_agree() { - let key = signing_key(1); - let payload = b"ts-and-keypackage-bytes".to_vec(); - - let wire = KeyPackageSubmissionV1::decode(&keypackage_submission_proto(&key, &payload)[..]) - .unwrap(); - let from_proto = - Submission::keypackage(&wire.device_id, &wire.payload, &wire.signature).unwrap(); - - let body: SubmitRequest = - serde_json::from_slice(&keypackage_submission_json(&key, &payload)).unwrap(); - let from_json = body.decode().unwrap(); - - assert_eq!(from_proto.key_hex(), from_json.key_hex()); - assert_eq!(from_proto.payload, from_json.payload); - assert_eq!(from_proto.signature, from_json.signature); - } - - #[tokio::test] - async fn wire_proto_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 = KeyPackageSubmissionV1::decode(&keypackage_submission_proto(&key, &payload)[..]) - .unwrap(); - let submission = - Submission::keypackage(&wire.device_id, &wire.payload, &wire.signature).unwrap(); - apply_keypackage(&store, &submission).await.unwrap(); - - let stored = store.latest(&submission.key_hex()).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 wire = - KeyPackageSubmissionV1::decode(&keypackage_submission_proto(&key, b"original")[..]) - .unwrap(); - wire.payload = Bytes::from_static(b"tampered"); - let submission = - Submission::keypackage(&wire.device_id, &wire.payload, &wire.signature).unwrap(); - - let err = apply_keypackage(&store, &submission).await.unwrap_err(); - assert!(matches!( - err, - SubmitError::Invalid("signature: verification failed") - )); - assert!(store.latest(&submission.key_hex()).await.unwrap().is_none()); - } - - /// A truncated or foreign payload on the submission topics must be dropped, - /// not panic the ingest thread. - #[test] - fn non_protobuf_bytes_are_rejected() { - assert!(KeyPackageSubmissionV1::decode(&b"{\"device_id\":\"xx\"}"[..]).is_err()); - assert!(AccountSubmissionV1::decode(&[0xff, 0xff, 0xff][..]).is_err()); - } - - /// A key that is not 32 bytes is rejected at decode, before verification. - #[test] - fn short_key_is_rejected() { - let err = Submission::keypackage(&[1u8; 31], b"payload", &[0u8; 64]).unwrap_err(); - assert!(matches!( - err, - SubmitError::Invalid("device_id: must be 32 bytes") - )); - } - - #[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)); - } -}