logos-messaging/chat-store
1
0
Fork 0
mirror of https://github.com/logos-messaging/chat-store.git synced 2026-07-30 06:53:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: publish via logos delivery

Browse Source
This commit is contained in:
kaichaosun 2026-07-20 15:44:40 +08:00
parent 673c49671a
commit 83c30ad17c
No known key found for this signature in database
GPG Key ID: 223E0F992F4F03BF
8 changed files with 641 additions and 131 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
Cargo.lock generated
View File

@ -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"

4
Cargo.toml
View File

@ -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.

54
README.md
View File

@ -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 <n>` | `100` | Bundles retained per `device_id` |
| `--retention-days <n>` | `30` | Drop bundles older than this |
| `--prune-interval-secs <n>` | `3600` | How often the prune task runs |
| `--no-delivery` | off | Disable the logos-delivery subscriber (HTTP POST ingestion only) |
| `--preset <name>` | `logos.dev` | logos-delivery network preset the subscriber joins |
| `--p2p-port <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/<id> -> 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:

134
examples/delivery_smoke_test.rs Normal file
View File

@ -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::<Vec<u8>>::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<Vec<u8>>,
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));
}
}

150
src/handlers.rs
View File

@ -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<Store>) -> 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<Arc<Store>>,
Json(req): Json<SubmitRequest>,
) -> Result<StatusCode, ApiError> {
// 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<Arc<Store>>,
Json(req): Json<SubmitAccountRequest>,
) -> Result<StatusCode, ApiError> {
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<String>) -> Self {
Self {
status: StatusCode::BAD_REQUEST,
message: msg.into(),
}
}
fn not_found(msg: impl Into<String>) -> Self {
Self {
status: StatusCode::NOT_FOUND,
message: msg.into(),
}
}
fn conflict(msg: impl Into<String>) -> Self {
Self {
status: StatusCode::CONFLICT,
message: msg.into(),
}
}
fn internal<E: std::fmt::Display>(err: E) -> Self {
tracing::error!("internal: {err}");
Self {
@ -242,6 +128,22 @@ impl ApiError {
}
}
impl From<SubmitError> 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 {
(

102
src/ingest.rs Normal file
View File

@ -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<u8>),
Account(Vec<u8>),
}
/// 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<Store>, 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<u8>) -> 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}"),
}
}

48
src/main.rs
View File

@ -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

258
src/submit.rs Normal file
View File

@ -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<u8>,
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<Verified, SubmitError> {
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<u8> {
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<u8> {
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));
}
}