diff --git a/Cargo.lock b/Cargo.lock index 2bc4819..c80908d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -224,6 +224,14 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "chat-proto" +version = "0.1.0" +source = "git+https://github.com/logos-messaging/chat_proto?rev=e05c21229a18bab655c9aec06409189447d24a6b#e05c21229a18bab655c9aec06409189447d24a6b" +dependencies = [ + "prost", +] + [[package]] name = "chat-store" version = "0.1.0" @@ -231,9 +239,12 @@ dependencies = [ "anyhow", "axum", "base64", + "chat-proto", "clap", "ed25519-dalek", "hex", + "logos-delivery", + "prost", "reqwest", "serde", "serde_json", @@ -328,6 +339,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" @@ -900,6 +920,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" @@ -982,6 +1011,19 @@ version = "0.4.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "953f07c43838f8e6f9758cab68bf5bed85465e7587ebe0b823f1bcd81978ad3a" +[[package]] +name = "logos-delivery" +version = "0.1.0" +source = "git+https://github.com/logos-messaging/libchat?rev=5c55c2ee76bebd1dbd5eb4bfa95d9e71acd0fa14#5c55c2ee76bebd1dbd5eb4bfa95d9e71acd0fa14" +dependencies = [ + "base64", + "crossbeam-channel", + "serde", + "serde_json", + "thiserror", + "tracing", +] + [[package]] name = "matchers" version = "0.2.0" @@ -1207,6 +1249,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 0dc0376..fa3af16 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,9 +15,12 @@ path = "src/main.rs" anyhow = "1.0" axum = "0.7" base64 = "0.22" +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" +logos-delivery = { git = "https://github.com/logos-messaging/libchat", rev = "5c55c2ee76bebd1dbd5eb4bfa95d9e71acd0fa14" } +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/Dockerfile b/Dockerfile index b0c6088..ad58cf2 100644 --- a/Dockerfile +++ b/Dockerfile @@ -3,24 +3,55 @@ ######################################## # Build stage ######################################## -FROM rust:1-bookworm AS builder +# Nix, not the `rust` image: the binary links the native liblogosdelivery, which +# only nix builds. That library comes from nixpkgs' glibc (2.40), newer than +# bookworm's, so cargo has to run against the same nixpkgs — mixing the two +# libcs in one process does not work. Everything below therefore builds inside +# the flake's dev shell. +FROM nixos/nix:2.32.0 AS builder -WORKDIR /app +# `filter-syscalls` off because the seccomp filter fails inside an emulated +# (qemu) container, which is how this image gets built on arm64 hosts. +RUN printf 'experimental-features = nix-command flakes\nfilter-syscalls = false\n' \ + >> /etc/nix/nix.conf -# Build dependencies first against a stub binary so the (slow) dependency -# compilation — bundled libsqlite3 and the async stack — is cached and only -# re-runs when Cargo.toml/Cargo.lock change. +WORKDIR /src + +# The native library first and on its own layer: it is by far the slowest part +# of the build and only changes when the flake pins do. +COPY flake.nix flake.lock ./ +RUN nix --accept-flake-config build .#logos-delivery --no-link + +# Then the Rust dependency graph, against a stub binary, so the (also slow) +# dependency compile is cached and only re-runs when Cargo.toml/Cargo.lock change. COPY Cargo.toml Cargo.lock ./ RUN mkdir src \ && echo "fn main() {}" > src/main.rs \ - && cargo build --release --locked \ + && nix --accept-flake-config develop --command cargo build --release --locked \ && rm -rf src target/release/deps/chat_store* target/release/chat-store -# Now build the real binary; dependency artifacts above are reused. `migrations/` -# is needed at build time too — `sqlx::migrate!()` embeds the .sql files. +# Now the real binary; dependency artifacts above are reused. `migrations/` is +# needed at build time too — `sqlx::migrate!()` embeds the .sql files. COPY src ./src COPY migrations ./migrations -RUN cargo build --release --locked --bin chat-store +RUN nix --accept-flake-config develop --command \ + cargo build --release --locked --bin chat-store + +# Collect what the binary loads at runtime. logos-delivery's build.rs stamps +# liblogosdelivery with an absolute store path as its soname, so the binary +# names it — along with its ELF interpreter and rpath — by absolute /nix/store +# path. Seeding the closure from those three is all the runtime image needs. +RUN nix --accept-flake-config develop --command sh -c '\ + bin=target/release/chat-store; \ + { patchelf --print-needed "$bin"; \ + patchelf --print-interpreter "$bin"; \ + patchelf --print-rpath "$bin" | tr ":" "\n"; } \ + | grep "^/nix/store/" \ + | sed "s|\(/nix/store/[^/]*\).*|\1|" | sort -u \ + | xargs nix-store --query --requisites | sort -u > /closure.txt' \ + && mkdir -p /output/nix/store \ + && cp target/release/chat-store /output/chat-store \ + && cp -a $(cat /closure.txt) /output/nix/store/ ######################################## # Runtime stage @@ -31,14 +62,18 @@ RUN apt-get update \ && apt-get install -y --no-install-recommends ca-certificates \ && rm -rf /var/lib/apt/lists/* -COPY --from=builder /app/target/release/chat-store /usr/local/bin/chat-store +# The nix closure the binary resolves by absolute path, then the binary itself. +COPY --from=builder /output/nix/store /nix/store +COPY --from=builder /output/chat-store /usr/local/bin/chat-store # Matches the default --bind 0.0.0.0:8080. EXPOSE 8080 +EXPOSE 60000/tcp +EXPOSE 60000/udp # Persist the SQLite database on a volume rather than the container layer. VOLUME ["/data"] ENV RUST_LOG=info ENTRYPOINT ["chat-store"] -CMD ["--bind", "0.0.0.0:8080", "--db", "/data/chat-store.db"] +CMD ["--bind", "0.0.0.0:8080", "--db", "/data/chat-store.db", "--p2p-port", "60000"] diff --git a/README.md b/README.md index 8460f5d..f3f8d4f 100644 --- a/README.md +++ b/README.md @@ -5,10 +5,21 @@ Persistence for group-chat users' key packages — the **chat-store** HTTP servi [libchat](https://github.com/logos-messaging/libchat) so it can be deployed on its own. -Standalone HTTP service that caches MLS KeyPackages keyed by **`device_id`**, so a +Standalone service that caches MLS KeyPackages keyed by **`device_id`**, so a client can fetch a contact's keypackage without an out-of-band exchange. -Throwaway by design: scheduled to be replaced by a λLEZ-based service in v0.3, so -it intentionally has no overlap with the rest of libchat (axum + rusqlite only). +Throwaway by design: scheduled to be replaced by a λLEZ-based service in v0.3, +with no libchat-core dependency (the embedded logos-delivery node comes from +libchat's transport crate, pulled in as a pinned git dependency). + +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 protobuf submissions 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. @@ -45,11 +56,17 @@ exactly one way — no delimiter, even though `key_package` is arbitrary bytes. ## Building & running +Building a runnable binary links the native `liblogosdelivery`, which this +repo's flake builds. The dev shell exports `LOGOS_DELIVERY_LIB_DIR` for you: + ```bash +nix develop # or set LOGOS_DELIVERY_LIB_DIR yourself cargo build --release ./target/release/chat-store # binds 0.0.0.0:8080, db ./chat-store.db ``` +Without the library, `cargo check` and `clippy` still pass — only linking fails. + | Flag | Default | Description | |------|---------|-------------| | `--bind ` | `0.0.0.0:8080` | HTTP bind address | @@ -57,27 +74,68 @@ cargo build --release | `--max-per-identity ` | `100` | Bundles retained per `device_id` | | `--retention-days ` | `30` | Drop bundles older than this | | `--prune-interval-secs ` | `3600` | How often the prune task runs | +| `--no-delivery` | off | Disable the logos-delivery subscriber (HTTP POST ingestion only) | +| `--preset ` | `logos.dev` | logos-delivery network preset the subscriber joins | +| `--p2p-port ` | `0` | TCP + discv5 UDP port for the embedded node (0 = OS-assigned) | Logs via `RUST_LOG` (default `info`). +## 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 a protobuf `KeyPackageSubmissionV1` or +`AccountSubmissionV1` — matching the `/proto` topic suffix — carrying the same +fields as the corresponding POST body and going through identical signature +verification and storage rules. The schemas live in +[chat_proto](https://github.com/logos-messaging/chat_proto) (`protos/store.proto`), +pinned by rev in `Cargo.toml` so the wire format only changes deliberately. + +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 `ContactRegistry` publishes on these +topics when constructed with `RegistryPublishMode::Delivery`. + ## Docker ```bash # Build the image docker build -t chat-store . -# Run it, persisting the SQLite db on a named volume and exposing port 8080 -docker run --rm -p 8080:8080 -v chat-store-data:/data chat-store +# Run it, persisting the SQLite db on a named volume. 8080 is the HTTP API; +# 60000 is the delivery node's libp2p (TCP) and discv5 (UDP) port. +docker run --rm -p 8080:8080 -p 60000:60000/tcp -p 60000:60000/udp \ + -v chat-store-data:/data chat-store ``` -The image runs the binary with `--bind 0.0.0.0:8080 --db /data/chat-store.db` -by default; override the `CMD` to change flags, e.g.: +The build runs inside nix, because the binary links `liblogosdelivery` and the +nixpkgs glibc it is built against is newer than bookworm's — so cargo has to run +against that same nixpkgs. The runtime image is still `debian:bookworm-slim`, +carrying only the nix store closure the binary resolves by absolute path. The +first build compiles the native library from source and takes a long time; +afterwards it is a cached layer that only changes with `flake.lock`. + +The image runs the binary with +`--bind 0.0.0.0:8080 --db /data/chat-store.db --p2p-port 60000` by default. The +p2p port is pinned rather than left at its default of `0`, because an +OS-assigned port cannot be published from a container and would leave the node +undialable. Override the `CMD` to change flags, e.g.: ```bash docker run --rm -p 9000:9000 -v chat-store-data:/data chat-store \ - --bind 0.0.0.0:9000 --db /data/registry.db --retention-days 14 + --bind 0.0.0.0:9000 --db /data/registry.db --retention-days 14 --no-delivery ``` +Note that overriding `CMD` replaces it wholesale, so a delivery-enabled run has +to repeat `--p2p-port`. Deployments that only serve the HTTP API can pass +`--no-delivery` and skip publishing the p2p ports entirely. + ## API ### `POST /v0/keypackage` @@ -191,6 +249,19 @@ GET /v0/account/ -> 200 OK (expect 200) {"payload":...,"signature":... POST /v0/account (replay) -> 409 Conflict (expect 409) ``` +The delivery write path has its own smoke test, +[`delivery_smoke_test`](examples/delivery_smoke_test.rs): it starts a publisher +node, publishes a signed keypackage and account bundle on the store's content +topics, and polls the query API until both appear: + +```bash +# Terminal 1 — start a server (delivery ingestion is on by default) +cargo run -- --bind 127.0.0.1:8080 --db tmp/chat-store.db + +# Terminal 2 — publish over the network and poll the query API +cargo run --example delivery_smoke_test +``` + You can also exercise it with the real `chat-cli` (which lives in the [libchat](https://github.com/logos-messaging/libchat) repo) against a running server: @@ -214,6 +285,11 @@ A non-zero exit from `chat-cli` means the server rejected the submission — e.g the signature failed verification. `GET /v0/keypackage/{device_id}` returns `200` for a registered device and `404` otherwise. +Add `--registry-publish delivery` to those `chat-cli` invocations to exercise +the delivery write path instead of the HTTP POST API. Publishing is then +fire-and-forget, so `chat-cli` exits `0` regardless and the bundles appear in +the database only once the server has received and accepted them. + ## Benchmark The bundled [`benchmark`](examples/benchmark.rs) performs an end-to-end load diff --git a/examples/delivery_smoke_test.rs b/examples/delivery_smoke_test.rs new file mode 100644 index 0000000..5fca756 --- /dev/null +++ b/examples/delivery_smoke_test.rs @@ -0,0 +1,141 @@ +//! 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 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 +//! 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 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; + +/// 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"; + +/// Domain-separation prefix the server expects on every account bundle payload. +const ACCOUNT_BUNDLE_DOMAIN: &[u8] = b"libchat:account-device-bundle\0"; + +/// How long to keep polling the query API for the published bundles. Covers +/// node startup, peer discovery, and gossip propagation. +const POLL_BUDGET: Duration = Duration::from_secs(90); + +fn main() -> Result<()> { + let base = std::env::args() + .nth(1) + .unwrap_or_else(|| "http://127.0.0.1:8080".to_string()); + let base = base.trim_end_matches('/').to_string(); + let client = Client::new(); + + println!("Starting logos-delivery publisher node (this takes a few seconds)..."); + let node = ThreadedDeliveryWrapper::>::start(P2pConfig::default(), |_| None) + .map_err(|e| anyhow::anyhow!("start logos-delivery node: {e}"))?; + + // Unique keys per run so re-runs don't hit the account lamport replay guard. + let seed_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64; + let device_key = derived_key(seed_ms, 1); + let account_key = derived_key(seed_ms, 2); + let device_id = hex::encode(device_key.verifying_key().as_bytes()); + let account_pub = hex::encode(account_key.verifying_key().as_bytes()); + + // Keypackage payload: opaque to the server; real clients use + // `timestamp_ms_le[8] || key_package`. + let mut kp_payload = seed_ms.to_le_bytes().to_vec(); + kp_payload.extend_from_slice(b"delivery-smoke-keypackage"); + publish( + &node, + KEYPACKAGE_SUBMIT_TOPIC, + &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}"); + + // 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( + &node, + ACCOUNT_SUBMIT_TOPIC, + &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}"); + + 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( + node: &ThreadedDeliveryWrapper>, + topic: &str, + submission: &M, +) -> Result<()> { + node.publish(topic, &submission.encode_to_vec()) + .map_err(|e| anyhow::anyhow!("publish on {topic}: {e}")) +} + +/// Poll `url` until it returns 200 (the subscriber stored the bundle) or the +/// budget runs out. +fn poll_until_stored(client: &Client, url: &str, what: &str) -> Result<()> { + let started = Instant::now(); + loop { + let status = client + .get(url) + .send() + .with_context(|| format!("GET {url}"))? + .status(); + if status.is_success() { + println!( + "GET {url} -> {status} ({what} stored, {:?})", + started.elapsed() + ); + return Ok(()); + } + if started.elapsed() > POLL_BUDGET { + bail!("{what} did not appear within {POLL_BUDGET:?} (last status {status})"); + } + std::thread::sleep(Duration::from_secs(2)); + } +} diff --git a/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/flake.lock b/flake.lock new file mode 100644 index 0000000..553450c --- /dev/null +++ b/flake.lock @@ -0,0 +1,115 @@ +{ + "nodes": { + "logos-delivery": { + "inputs": { + "nixpkgs": [ + "nixpkgs" + ], + "rust-overlay": "rust-overlay", + "zerokit": "zerokit" + }, + "locked": { + "lastModified": 1779915920, + "narHash": "sha256-rcIgP6MVyUoNEH6xpdLrZtfd4OcvIcMUloX4IhRq5AA=", + "owner": "logos-messaging", + "repo": "logos-delivery", + "rev": "74057c66224f43b4aa27b42033d4ed52eed5c7a7", + "type": "github" + }, + "original": { + "owner": "logos-messaging", + "repo": "logos-delivery", + "rev": "74057c66224f43b4aa27b42033d4ed52eed5c7a7", + "type": "github" + } + }, + "nixpkgs": { + "locked": { + "lastModified": 1782231800, + "narHash": "sha256-AyPvq+cx3JFicSd687dhhIfUMryk9PUFmHnofBjalMg=", + "owner": "NixOS", + "repo": "nixpkgs", + "rev": "6f5e2d2ea55618c5197ce40e346f205c35d2213e", + "type": "github" + }, + "original": { + "owner": "NixOS", + "repo": "nixpkgs", + "rev": "6f5e2d2ea55618c5197ce40e346f205c35d2213e", + "type": "github" + } + }, + "root": { + "inputs": { + "logos-delivery": "logos-delivery", + "nixpkgs": "nixpkgs" + } + }, + "rust-overlay": { + "inputs": { + "nixpkgs": [ + "logos-delivery", + "nixpkgs" + ] + }, + "locked": { + "lastModified": 1775099554, + "narHash": "sha256-3xBsGnGDLOFtnPZ1D3j2LU19wpAlYefRKTlkv648rU0=", + "owner": "oxalica", + "repo": "rust-overlay", + "rev": "8d6387ed6d8e6e6672fd3ed4b61b59d44b124d99", + "type": "github" + }, + "original": { + "owner": "oxalica", + "repo": "rust-overlay", + "type": "github" + } + }, + "rust-overlay_2": { + "inputs": { + "nixpkgs": [ + "logos-delivery", + "zerokit", + "nixpkgs" + ] + }, + "locked": { + "lastModified": 1771211437, + "narHash": "sha256-lcNK438i4DGtyA+bPXXyVLHVmJjYpVKmpux9WASa3ro=", + "owner": "oxalica", + "repo": "rust-overlay", + "rev": "c62195b3d6e1bb11e0c2fb2a494117d3b55d410f", + "type": "github" + }, + "original": { + "owner": "oxalica", + "repo": "rust-overlay", + "type": "github" + } + }, + "zerokit": { + "inputs": { + "nixpkgs": [ + "logos-delivery", + "nixpkgs" + ], + "rust-overlay": "rust-overlay_2" + }, + "locked": { + "owner": "vacp2p", + "repo": "zerokit", + "rev": "5e64cb8822bee65eed6cf459f95ae72b80c6ba63", + "type": "github" + }, + "original": { + "owner": "vacp2p", + "repo": "zerokit", + "rev": "5e64cb8822bee65eed6cf459f95ae72b80c6ba63", + "type": "github" + } + } + }, + "root": "root", + "version": 7 +} diff --git a/flake.nix b/flake.nix new file mode 100644 index 0000000..73348cb --- /dev/null +++ b/flake.nix @@ -0,0 +1,62 @@ +{ + description = "chat-store - keypackage and account registry for Logos Chat"; + + # Repeated from the logos-delivery flake: only the top-level flake's nixConfig + # is honoured, so an input's substituters do nothing for us. + nixConfig = { + extra-substituters = [ "https://nix-cache.status.im/" ]; + extra-trusted-public-keys = [ + "nix-cache.status.im-1:x/93lOfLU+duPplwMSBR+OlY4+mo+dCN7n0mr4oPwgY=" + ]; + }; + + inputs = { + # Both revs are exactly what libchat's flake.lock pins, and they only work + # as a pair: logos-delivery's derivation wants `nim-2_2`, which newer + # nixos-unstable-small no longer has. Bump the two together. + nixpkgs.url = "github:NixOS/nixpkgs/6f5e2d2ea55618c5197ce40e346f205c35d2213e"; + # `follows` keeps the library and the toolchain that links it on a single + # glibc — the Docker build depends on that. + logos-delivery = { + url = "github:logos-messaging/logos-delivery/74057c66224f43b4aa27b42033d4ed52eed5c7a7"; + inputs.nixpkgs.follows = "nixpkgs"; + }; + }; + + outputs = { self, nixpkgs, logos-delivery }: + let + systems = [ "aarch64-darwin" "x86_64-darwin" "aarch64-linux" "x86_64-linux" ]; + forAllSystems = f: nixpkgs.lib.genAttrs systems (system: f { + inherit system; + pkgs = import nixpkgs { inherit system; }; + }); + in + { + # The one thing this flake exists for: the native library the + # `logos-delivery` crate links. Same override as libchat's flake — no + # postgres archive driver, no Nim dlopen debug hooks, quiet logs. + packages = forAllSystems ({ system, ... }: { + logos-delivery = logos-delivery.packages.${system}.liblogosdelivery.override { + enablePostgres = false; + enableNimDebugDlOpen = false; + chroniclesLogLevel = "FATAL"; + }; + default = self.packages.${system}.logos-delivery; + }); + + devShells = forAllSystems ({ pkgs, system, ... }: + let logosDeliveryLib = self.packages.${system}.logos-delivery; + in { + default = pkgs.mkShell { + nativeBuildInputs = [ pkgs.cargo pkgs.rustc pkgs.pkg-config ] + # logos-delivery's build.rs rewrites the shipped library's soname + # with patchelf in its default (absolute-path) linking mode. + ++ pkgs.lib.optionals pkgs.stdenv.isLinux [ pkgs.patchelf ]; + buildInputs = [ logosDeliveryLib ]; + shellHook = '' + export LOGOS_DELIVERY_LIB_DIR="${logosDeliveryLib}/lib" + ''; + }; + }); + }; +} 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/delivery.rs b/src/delivery.rs new file mode 100644 index 0000000..0f9f1da --- /dev/null +++ b/src/delivery.rs @@ -0,0 +1,123 @@ +//! The logos-delivery write path. +//! +//! 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; + +use anyhow::{Context, Result}; +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::bundle::{Bundle, apply_account, apply_keypackage}; +use crate::store::Store; + +/// 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 bundles. +pub const ACCOUNT_SUBMIT_TOPIC: &str = "/logos-chat/1/store-account-v0/proto"; + +/// 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 { + KeyPackage(Vec), + Account(Vec), +} + +/// 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 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| { + let msg = event.into_received()?; + let wrap = match msg.content_topic() { + KEYPACKAGE_SUBMIT_TOPIC => Received::KeyPackage as fn(Vec) -> Received, + ACCOUNT_SUBMIT_TOPIC => Received::Account, + _ => return None, + }; + msg.into_payload().map(wrap) + }) + .context("start embedded logos-delivery node")?; + + node.subscribe(KEYPACKAGE_SUBMIT_TOPIC) + .context("subscribe keypackage bundles")?; + node.subscribe(ACCOUNT_SUBMIT_TOPIC) + .context("subscribe account bundles")?; + + 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(received) = inbound.recv() { + match received { + Received::KeyPackage(bytes) => ingest_keypackage(&store, &runtime, &bytes), + Received::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 wire = match KeyPackageSubmissionV1::decode(bytes) { + Ok(wire) => wire, + Err(e) => { + warn!("keypackage bundle: invalid protobuf: {e}"); + return; + } + }; + let bundle = match Bundle::from_bytes(&wire.device_id, &wire.payload, &wire.signature) { + Ok(bundle) => bundle, + Err(e) => { + warn!("keypackage bundle rejected: {e}"); + return; + } + }; + 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 bundle rejected: {e}"), + } +} + +fn ingest_account(store: &Store, runtime: &tokio::runtime::Handle, bytes: &[u8]) { + let wire = match AccountSubmissionV1::decode(bytes) { + Ok(wire) => wire, + Err(e) => { + warn!("account bundle: invalid protobuf: {e}"); + return; + } + }; + let bundle = match Bundle::from_bytes(&wire.account_pub, &wire.payload, &wire.signature) { + Ok(bundle) => bundle, + Err(e) => { + warn!("account bundle rejected: {e}"); + return; + } + }; + 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 bundle rejected: {e}"), + } +} diff --git a/src/handlers.rs b/src/handlers.rs index d2e400c..768f09a 100644 --- a/src/handlers.rs +++ b/src/handlers.rs @@ -7,13 +7,14 @@ 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 crate::store::{Store, StoredAccountBundle, StoredKeyPackageBundle}; +use crate::bundle::{self, Bundle, BundleError}; +use crate::store::Store; +/// A signed keypackage bundle. #[derive(Debug, Deserialize)] -pub struct SubmitRequest { +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, @@ -25,86 +26,14 @@ pub struct SubmitRequest { pub signature: String, } -#[derive(Debug, Serialize)] -pub struct FetchResponse { - /// base64 of the stored payload; consumers verify `signature` over it. - pub payload: String, - 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) + } } -#[derive(Debug, Serialize)] -struct ErrorBody { - error: String, -} - -pub fn router(store: Arc) -> Router { - Router::new() - .route("/v0/keypackage", post(submit)) - .route("/v0/keypackage/:device_id", get(fetch)) - .route("/v0/account", post(submit_account)) - .route("/v0/account/:account_pub", get(fetch_account)) - .with_state(store) -} - -async fn submit( - State(store): State>, - Json(req): Json, -) -> Result { - // Verify proof-of-possession before persisting. `payload` is opaque — the - // server only checks that `signature` over the received payload bytes is - // valid under `device_id`'s key. A valid signature means the submitter holds - // that key. This rejects junk early (DoS mitigation); consumers still verify - // on retrieve, the server is not a trusted authority. - let device_pubkey: [u8; 32] = hex::decode(&req.device_id) - .ok() - .and_then(|b| b.try_into().ok()) - .ok_or_else(|| ApiError::bad("device_id: must be hex of a 32-byte key"))?; - let payload = BASE64 - .decode(&req.payload) - .map_err(|_| ApiError::bad("payload: not valid base64"))?; - let signature: [u8; 64] = BASE64 - .decode(&req.signature) - .ok() - .and_then(|b| b.try_into().ok()) - .ok_or_else(|| ApiError::bad("signature: must be base64 of 64 bytes"))?; - - let verifying_key = VerifyingKey::from_bytes(&device_pubkey) - .map_err(|_| ApiError::bad("device_id: not a valid ed25519 key"))?; - verifying_key - .verify_strict(&payload, &Signature::from_bytes(&signature)) - .map_err(|_| ApiError::bad("signature: verification failed"))?; - - store - .insert( - &req.device_id, - &StoredKeyPackageBundle { - payload, - signature: signature.to_vec(), - }, - ) - .await - .map_err(ApiError::internal)?; - Ok(StatusCode::NO_CONTENT) -} - -async fn fetch( - State(store): State>, - Path(device_id): Path, -) -> 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 { - payload: BASE64.encode(&bundle.payload), - signature: BASE64.encode(&bundle.signature), - })) -} - -/// Request body for publishing a signed device-list bundle under an account. +/// 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 @@ -123,6 +52,76 @@ pub struct SubmitAccountRequest { 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 FetchKeyPackageResponse { + /// base64 of the stored payload; consumers verify `signature` over it. + pub payload: String, + pub signature: String, +} + +#[derive(Debug, Serialize)] +struct ErrorBody { + error: String, +} + +pub fn router(store: Arc) -> Router { + Router::new() + .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 bundle the logos-delivery subscriber +/// accepts, in JSON rather than protobuf; verification and storage live in +/// [`bundle::apply_keypackage`]. +async fn submit_key_package( + State(store): State>, + Json(req): Json, +) -> Result { + bundle::apply_keypackage(&store, &req.decode()?).await?; + Ok(StatusCode::NO_CONTENT) +} + +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 { + return Err(ApiError::not_found("no keypackage for device")); + }; + Ok(Json(FetchKeyPackageResponse { + payload: BASE64.encode(&bundle.payload), + signature: BASE64.encode(&bundle.signature), + })) +} + #[derive(Debug, Serialize)] pub struct FetchAccountResponse { /// base64 of the stored payload. @@ -137,53 +136,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. The same submission the +/// logos-delivery subscriber accepts, in JSON rather than protobuf; the shared +/// rules live in [`bundle::apply_account`]. async fn submit_account( State(store): State>, Json(req): Json, ) -> Result { - let account_pubkey: [u8; 32] = hex::decode(&req.account_pub) - .ok() - .and_then(|b| b.try_into().ok()) - .ok_or_else(|| ApiError::bad("account_pub: must be hex of a 32-byte key"))?; - let payload = BASE64 - .decode(&req.payload) - .map_err(|_| ApiError::bad("payload: not valid base64"))?; - let signature: [u8; 64] = BASE64 - .decode(&req.signature) - .ok() - .and_then(|b| b.try_into().ok()) - .ok_or_else(|| ApiError::bad("signature: must be base64 of 64 bytes"))?; - - let verifying_key = VerifyingKey::from_bytes(&account_pubkey) - .map_err(|_| ApiError::bad("account_pub: not a valid ed25519 key"))?; - verifying_key - .verify_strict(&payload, &Signature::from_bytes(&signature)) - .map_err(|_| ApiError::bad("signature: verification failed"))?; - - // Read the bundle's lamport so the store can reject replays. Safe to trust: - // the signature over `payload` was just verified, so the lamport can't be - // forged without the account key. - let lamport = crate::store::payload_lamport(&payload) - .ok_or_else(|| ApiError::bad("payload: too short to contain a lamport header"))?; - - let applied = store - .upsert_account( - &req.account_pub, - lamport, - &StoredAccountBundle { - payload, - signature: signature.to_vec(), - updated_at: 0, // filled in by store - }, - ) - .await - .map_err(ApiError::internal)?; - if !applied { - return Err(ApiError::conflict( - "stale bundle: lamport is not newer than the stored one", - )); - } + bundle::apply_account(&store, &req.decode()?).await?; Ok(StatusCode::NO_CONTENT) } @@ -215,24 +175,12 @@ struct ApiError { } impl ApiError { - fn bad(msg: impl Into) -> Self { - Self { - status: StatusCode::BAD_REQUEST, - message: msg.into(), - } - } fn not_found(msg: impl Into) -> Self { Self { status: StatusCode::NOT_FOUND, message: msg.into(), } } - fn conflict(msg: impl Into) -> Self { - Self { - status: StatusCode::CONFLICT, - message: msg.into(), - } - } fn internal(err: E) -> Self { tracing::error!("internal: {err}"); Self { @@ -242,6 +190,22 @@ impl ApiError { } } +impl From for ApiError { + fn from(err: BundleError) -> Self { + match err { + BundleError::Invalid(msg) => Self { + status: StatusCode::BAD_REQUEST, + message: msg.into(), + }, + BundleError::Stale => Self { + status: StatusCode::CONFLICT, + message: err.to_string(), + }, + BundleError::Internal(inner) => Self::internal(inner), + } + } +} + impl IntoResponse for ApiError { fn into_response(self) -> Response { ( diff --git a/src/main.rs b/src/main.rs index 79d23f6..da53754 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,14 +1,23 @@ -//! 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 libchat's transport +//! crate, pulled in as a pinned git dependency. //! -//! Wire: +//! Submissions arrive on either write path — both feed the same verification + +//! storage pipeline (`bundle`): +//! - HTTP POST (below), synchronous and acknowledged; +//! - 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 //! 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 bundle; +mod delivery; mod handlers; mod store; @@ -24,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")] @@ -45,6 +57,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] @@ -57,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; @@ -80,6 +102,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. + delivery::start( + store.clone(), + delivery::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