diff --git a/Cargo.lock b/Cargo.lock index a49d9d0..b773e5e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -224,6 +224,13 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "chat-proto" +version = "0.1.0" +dependencies = [ + "prost", +] + [[package]] name = "chat-store" version = "0.1.0" @@ -231,10 +238,12 @@ dependencies = [ "anyhow", "axum", "base64", + "chat-proto", "clap", "ed25519-dalek", "hex", "logos-delivery", + "prost", "reqwest", "serde", "serde_json", @@ -910,6 +919,15 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itertools" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.18" @@ -1229,6 +1247,29 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "prost" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-derive" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "quote" version = "1.0.45" diff --git a/Cargo.toml b/Cargo.toml index 6102daf..1e4be94 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,6 +15,11 @@ path = "src/main.rs" anyhow = "1.0" axum = "0.7" base64 = "0.22" +# Wire schema for the delivery submissions, shared with libchat. +# TODO: switch to `{ git = "https://github.com/logos-messaging/chat_proto", rev = "..." }` +# once the store submission schemas land on chat_proto's main. Path dep only so +# this builds ahead of that merge. +chat-proto = { path = "../chat_proto/gen/rust" } clap = { version = "4", features = ["derive"] } ed25519-dalek = "2.2.0" hex = "0.4" @@ -22,6 +27,7 @@ hex = "0.4" # 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" } +prost = "0.14" 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/examples/delivery_smoke_test.rs b/examples/delivery_smoke_test.rs index 7e6e153..924cb99 100644 --- a/examples/delivery_smoke_test.rs +++ b/examples/delivery_smoke_test.rs @@ -1,10 +1,10 @@ //! 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. +//! the delivery network — the protobuf submissions the store subscribes for, +//! carrying the same fields the HTTP POST endpoints take as JSON — 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 @@ -15,12 +15,12 @@ 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 chat_proto::logoschat::store::{AccountSubmissionV1, KeyPackageSubmissionV1}; use ed25519_dalek::{Signer, SigningKey}; use logos_delivery::{P2pConfig, ThreadedDeliveryWrapper}; +use prost::Message; +use prost::bytes::Bytes; 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). @@ -59,14 +59,14 @@ fn main() -> Result<()> { // `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( + publish( &node, KEYPACKAGE_SUBMIT_TOPIC, - &json!({ - "device_id": device_id, - "payload": BASE64.encode(&kp_payload), - "signature": BASE64.encode(device_key.sign(&kp_payload).to_bytes()), - }), + &KeyPackageSubmissionV1 { + device_id: Bytes::copy_from_slice(device_key.verifying_key().as_bytes()), + payload: Bytes::copy_from_slice(&kp_payload), + signature: Bytes::copy_from_slice(&device_key.sign(&kp_payload).to_bytes()), + }, )?; println!("published keypackage submission on {KEYPACKAGE_SUBMIT_TOPIC}"); @@ -75,14 +75,14 @@ fn main() -> Result<()> { acct_payload.push(1u8); acct_payload.extend_from_slice(&1u64.to_le_bytes()); acct_payload.extend_from_slice(b"delivery-smoke-devices"); - publish_json( + publish( &node, ACCOUNT_SUBMIT_TOPIC, - &json!({ - "account_pub": account_pub, - "payload": BASE64.encode(&acct_payload), - "signature": BASE64.encode(account_key.sign(&acct_payload).to_bytes()), - }), + &AccountSubmissionV1 { + account_pub: Bytes::copy_from_slice(account_key.verifying_key().as_bytes()), + payload: Bytes::copy_from_slice(&acct_payload), + signature: Bytes::copy_from_slice(&account_key.sign(&acct_payload).to_bytes()), + }, )?; println!("published account submission on {ACCOUNT_SUBMIT_TOPIC}"); @@ -103,12 +103,12 @@ fn derived_key(seed_ms: u64, salt: u8) -> SigningKey { SigningKey::from_bytes(&bytes) } -fn publish_json( +fn publish( node: &ThreadedDeliveryWrapper>, topic: &str, - body: &serde_json::Value, + submission: &M, ) -> Result<()> { - node.publish(topic, body.to_string().as_bytes()) + node.publish(topic, &submission.encode_to_vec()) .map_err(|e| anyhow::anyhow!("publish on {topic}: {e}")) } diff --git a/src/handlers.rs b/src/handlers.rs index d7ed86b..8017bdb 100644 --- a/src/handlers.rs +++ b/src/handlers.rs @@ -33,13 +33,14 @@ 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`]. +/// `POST /v0/keypackage` — the same submission the logos-delivery subscriber +/// accepts, in JSON rather than protobuf; verification and storage live in +/// [`submit::apply_keypackage`]. async fn submit( State(store): State>, Json(req): Json, ) -> Result { - submit::apply_keypackage(&store, &req).await?; + submit::apply_keypackage(&store, &req.decode()?).await?; Ok(StatusCode::NO_CONTENT) } @@ -74,14 +75,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. Same submission the -/// logos-delivery subscriber accepts; the shared rules live in -/// [`submit::apply_account`]. +/// 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`]. async fn submit_account( State(store): State>, Json(req): Json, ) -> Result { - submit::apply_account(&store, &req).await?; + submit::apply_account(&store, &req.decode()?).await?; Ok(StatusCode::NO_CONTENT) } diff --git a/src/ingest.rs b/src/ingest.rs index 97a8cfa..2fd90ed 100644 --- a/src/ingest.rs +++ b/src/ingest.rs @@ -3,18 +3,23 @@ //! 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. +//! 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. use std::sync::Arc; use anyhow::{Context, Result}; -use logos_delivery::ThreadedDeliveryWrapper; +use chat_proto::logoschat::store::{AccountSubmissionV1, KeyPackageSubmissionV1}; pub use logos_delivery::P2pConfig; +use logos_delivery::ThreadedDeliveryWrapper; +use prost::Message; use tracing::{debug, warn}; use crate::store::Store; -use crate::submit::{self, SubmitAccountRequest, SubmitRequest}; +use crate::submit::{self, Submission}; /// Content topic carrying keypackage submissions. Must match what libchat's /// `DeliveryRegistry` publishes: delivery address `store-keypackage-v0` mapped @@ -24,10 +29,10 @@ pub const KEYPACKAGE_SUBMIT_TOPIC: &str = "/logos-chat/1/store-keypackage-v0/pro /// 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 +/// A raw submission taken off the wire; the protobuf body is decoded (and its /// signature verified) on the ingest thread, not the node callback. #[derive(Clone)] -enum Submission { +enum Received { KeyPackage(Vec), Account(Vec), } @@ -43,8 +48,8 @@ pub fn start(store: Arc, cfg: P2pConfig, runtime: tokio::runtime::Handle) 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, + KEYPACKAGE_SUBMIT_TOPIC => Received::KeyPackage as fn(Vec) -> Received, + ACCOUNT_SUBMIT_TOPIC => Received::Account, _ => return None, }; msg.into_payload().map(wrap) @@ -62,10 +67,10 @@ pub fn start(store: Arc, cfg: P2pConfig, runtime: tokio::runtime::Handle) .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), + while let Ok(received) = inbound.recv() { + match received { + Received::KeyPackage(bytes) => ingest_keypackage(&store, &runtime, &bytes), + Received::Account(bytes) => ingest_account(&store, &runtime, &bytes), } } }) @@ -74,29 +79,45 @@ pub fn start(store: Arc, cfg: P2pConfig, runtime: tokio::runtime::Handle) } fn ingest_keypackage(store: &Store, runtime: &tokio::runtime::Handle, bytes: &[u8]) { - let req: SubmitRequest = match serde_json::from_slice(bytes) { - Ok(req) => req, + let wire = match KeyPackageSubmissionV1::decode(bytes) { + Ok(wire) => wire, Err(e) => { - warn!("keypackage submission: invalid JSON: {e}"); + warn!("keypackage submission: invalid protobuf: {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}"), + let submission = match Submission::keypackage(&wire.device_id, &wire.payload, &wire.signature) { + Ok(submission) => submission, + Err(e) => { + warn!("keypackage submission rejected: {e}"); + return; + } + }; + let device_id = submission.key_hex(); + match runtime.block_on(submit::apply_keypackage(store, &submission)) { + Ok(()) => debug!(%device_id, "stored keypackage from delivery"), + Err(e) => warn!(%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, + let wire = match AccountSubmissionV1::decode(bytes) { + Ok(wire) => wire, Err(e) => { - warn!("account submission: invalid JSON: {e}"); + warn!("account submission: invalid protobuf: {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}"), + let submission = match Submission::account(&wire.account_pub, &wire.payload, &wire.signature) { + Ok(submission) => submission, + Err(e) => { + warn!("account submission rejected: {e}"); + return; + } + }; + let account_pub = submission.key_hex(); + match runtime.block_on(submit::apply_account(store, &submission)) { + Ok(()) => debug!(%account_pub, "stored account bundle from delivery"), + Err(e) => warn!(%account_pub, "account submission rejected: {e}"), } } diff --git a/src/submit.rs b/src/submit.rs index 3b2e711..6c3c8ef 100644 --- a/src/submit.rs +++ b/src/submit.rs @@ -1,9 +1,11 @@ //! 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. +//! 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; @@ -26,6 +28,13 @@ pub struct SubmitRequest { 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 @@ -45,6 +54,18 @@ pub struct SubmitAccountRequest { 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. @@ -67,70 +88,134 @@ impl std::fmt::Display for SubmitError { } } -/// Decoded and signature-verified fields common to both submission kinds. -struct Verified { +/// 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 `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))?; +/// 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(&payload, &Signature::from_bytes(&signature)) + .verify_strict( + &submission.payload, + &Signature::from_bytes(&submission.signature), + ) .map_err(|_| SubmitError::Invalid("signature: verification failed"))?; - Ok(Verified { payload, signature }) + 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, req: &SubmitRequest) -> Result<(), SubmitError> { - let verified = verify(&req.device_id, &req.payload, &req.signature, DEVICE_ID)?; +pub async fn apply_keypackage(store: &Store, submission: &Submission) -> Result<(), SubmitError> { + verify(submission, DEVICE_ID)?; store .insert( - &req.device_id, + &submission.key_hex(), &StoredKeyPackageBundle { - payload: verified.payload, - signature: verified.signature.to_vec(), + payload: submission.payload.clone(), + signature: submission.signature.to_vec(), }, ) .await @@ -138,23 +223,23 @@ pub async fn apply_keypackage(store: &Store, req: &SubmitRequest) -> Result<(), } /// 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)?; +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(&verified.payload).ok_or(SubmitError::Invalid( + 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( - &req.account_pub, + &submission.key_hex(), lamport, &StoredAccountBundle { - payload: verified.payload, - signature: verified.signature.to_vec(), + payload: submission.payload.clone(), + signature: submission.signature.to_vec(), updated_at: 0, // filled in by store }, ) @@ -170,7 +255,10 @@ pub async fn apply_account(store: &Store, req: &SubmitAccountRequest) -> Result< 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::*; @@ -182,8 +270,18 @@ mod tests { 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). + /// 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()), @@ -201,25 +299,49 @@ mod tests { 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()), - } + 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_json_keypackage_is_parsed_verified_and_stored() { + 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 = keypackage_submission_json(&key, &payload); - let req: SubmitRequest = serde_json::from_slice(&wire).unwrap(); - apply_keypackage(&store, &req).await.unwrap(); + 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(&req.device_id).await.unwrap().unwrap(); + let stored = store.latest(&submission.key_hex()).await.unwrap().unwrap(); assert_eq!(stored.payload, payload); } @@ -228,13 +350,37 @@ mod tests { 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 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, &req).await.unwrap_err(); - assert!(matches!(err, SubmitError::Invalid("signature: verification failed"))); - assert!(store.latest(&req.device_id).await.unwrap().is_none()); + 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]