swarm-nats-auth: open sealed callout requests and seal the replies

When nats-server has an auth_callout xkey it seals each request to that
key and names its own public xkey in the Nats-Server-Xkey header. The
responder now loads the callout xkey seed from the secret store before
it connects, opens a sealed request with it, and seals its reply (grant
or deny) back to the header's key. A request with no header is answered
in plaintext, as before.

A new `ensure-xkey` verb creates the pair when the store holds no seed:
the seed at swarm/services/swarm-nats/callout/xkey-seed, the public half
at swarm/public/swarm-nats/callout-xkey. Each write is a create-only
(cas=0) write, and a run whose create fails reads the stored value back,
so two concurrent runs end up with one key. The responder itself only
reads the seed and refuses to start without it.

swarm-secret-client gains SecretStore::create, the cas=0 write.

Cargo.lock

@@ -848,6 +848,35 @@ dependencies = [
  "hybrid-array",
 ]
 
+[[package]]
+name = "crypto_box"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "16182b4f39a82ec8a6851155cc4c0cda3065bb1db33651726a29e1951de0f009"
+dependencies = [
+ "aead",
+ "crypto_secretbox",
+ "curve25519-dalek",
+ "salsa20",
+ "subtle",
+ "zeroize",
+]
+
+[[package]]
+name = "crypto_secretbox"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9d6cf87adf719ddf43a805e92c6870a531aedda35ff640442cbaf8674e141e1"
+dependencies = [
+ "aead",
+ "cipher 0.4.4",
+ "generic-array",
+ "poly1305",
+ "salsa20",
+ "subtle",
+ "zeroize",
+]
+
 [[package]]
 name = "ctr"
 version = "0.9.2"
@@ -1527,6 +1556,7 @@ checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
 dependencies = [
  "typenum",
  "version_check",
+ "zeroize",
 ]
 
 [[package]]
@@ -3138,6 +3168,7 @@ version = "0.4.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "879011babc47a1c7fdf5a935ae3cfe94f34645ca0cac1c7f6424b36fc743d1bf"
 dependencies = [
+ "crypto_box",
  "data-encoding",
  "ed25519",
  "ed25519-dalek",
@@ -4295,6 +4326,15 @@ version = "1.0.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
 
+[[package]]
+name = "salsa20"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "97a22f5af31f73a954c10289c93e8a50cc23d971e80ee446f1f6f7137a088213"
+dependencies = [
+ "cipher 0.4.4",
+]
+
 [[package]]
 name = "same-file"
 version = "1.0.6"

swarm-nats-auth/Cargo.toml

@@ -16,7 +16,8 @@ clap.workspace = true
 data-encoding.workspace = true
 # StreamExt::next on the subscription: async-nats returns a Stream.
 futures-util.workspace = true
-nkeys.workspace = true
+# `xkeys` for the callout xkey: opening sealed requests and sealing replies.
+nkeys = { workspace = true, features = ["xkeys"] }
 reqwest.workspace = true
 serde.workspace = true
 serde_json.workspace = true

swarm-nats-auth/README.md

@@ -33,6 +33,21 @@ reintroduces the bug. `nats-jwt` is a **dev-dependency**: it cannot express
 `aud` on either token, and its role here is as the encoder's _test oracle_,
 not part of the path that runs.
 
+## The sealed exchange
+
+The server seals each auth request to the callout xkey and names its own
+public xkey in the `Nats-Server-Xkey` header; the responder opens the request
+with the xkey's seed and seals its reply to that header's key. It answers a
+request with no header in plaintext, which is what a server with no xkey
+configured reads.
+
+The seed is in the swarm secret store, and the responder reads it before it
+connects to the queue: a responder without it doesn't start. The
+`ensure-xkey` verb creates the pair when the store holds none. It runs from
+its own unit before the server, because the server needs the public half to
+start and the responder needs the server. `docs/swarm/secrets.md` has the
+paths and the boot order.
+
 ## Rules the code follows
 
 - **A denial is a signed response carrying an error, never silence.** The

swarm-nats-auth/src/main.rs

@@ -14,18 +14,23 @@
 //! indistinguishable from the responder being down, and the queue is the
 //! swarm's control path.
 //!
+//! A request that arrives sealed to the callout xkey is opened with it, and
+//! its reply is sealed back to the server; see `xkey`. The `ensure-xkey` verb
+//! creates that key in the secret store, and runs before the server does.
+//!
 //! # Secrets
 //!
 //! Every credential is taken as a **path**, never a value. Two reasons, both
 //! previously learned the hard way here: a value in nix config is rendered into
 //! the world-readable store, and a value in `argv` is readable by anyone via
 //! `/proc/<pid>/cmdline`, which is `0444`. Paths are not secrets, so passing
-//! them as flags is fine.
+//! them as flags is fine. The xkey seed is the exception to files: it is read
+//! from the secret store, at the path `xkey` names.
 
 use std::path::PathBuf;
 
 use anyhow::Context;
-use clap::Parser;
+use clap::{Parser, Subcommand};
 use futures_util::StreamExt;
 
 mod agent_token;
@@ -33,15 +38,39 @@ mod introspect;
 mod policy;
 mod request;
 mod respond;
+mod xkey;
 
 /// Subject the NATS server publishes authorization requests on.
 const AUTH_SUBJECT: &str = "$SYS.REQ.USER.AUTH";
 
+/// With no verb this is the responder, and its flags are required.
 #[derive(Debug, Parser)]
 #[command(
     name = "swarm-nats-auth",
-    about = "Auth-callout responder for the swarm NATS queue"
+    about = "Auth-callout responder for the swarm NATS queue",
+    args_conflicts_with_subcommands = true,
+    subcommand_negates_reqs = true
 )]
+struct Cli {
+    #[command(subcommand)]
+    command: Option<Command>,
+
+    #[command(flatten)]
+    responder: Option<Args>,
+}
+
+#[derive(Debug, Subcommand)]
+enum Command {
+    /// Create the callout xkey in the secret store when it holds none, and
+    /// check that its public half is stored beside it. Needs no NATS server.
+    EnsureXkey {
+        /// Role on the store's `cert` auth mount to log in as.
+        #[arg(long, default_value = "swarm-nats-auth")]
+        store_cert_role: String,
+    },
+}
+
+#[derive(Debug, clap::Args)]
 struct Args {
     /// NATS server to connect to.
     #[arg(long, default_value = "nats://127.0.0.1:4222")]
@@ -161,8 +190,17 @@ async fn main() -> anyhow::Result<()> {
         // they land in victorialogs as raw byte-array spam otherwise.
         .with_ansi(false)
         .init();
-    let args = Args::parse();
+    let cli = Cli::parse();
+    match (cli.command, cli.responder) {
+        (Some(Command::EnsureXkey { store_cert_role }), _) => ensure_xkey(&store_cert_role).await,
+        (None, Some(args)) => serve(&args).await,
+        (None, None) => unreachable!("clap requires the responder's flags when no verb is given"),
+    }
+}
 
+/// The responder: answers every authorization request until its
+/// subscription ends.
+async fn serve(args: &Args) -> anyhow::Result<()> {
     let client_secret = read_secret(&args.client_secret_file)?;
     // The bucket is NOT a flag. `swarm_queue_client::status`'s own docs say
     // why: reader and writer must name the same bucket, and an option is a
@@ -185,6 +223,15 @@ async fn main() -> anyhow::Result<()> {
     let http = reqwest::Client::new();
     let issuer = nkeys::KeyPair::from_seed(&read_secret(&args.issuer_seed_file)?)
         .context("parse the account signing seed")?;
+    // Before the connect: a responder that cannot open a sealed request must
+    // not start answering. Read, never created here; see `xkey`.
+    let callout_xkey = xkey::load(
+        &swarm_secret_client::SecretStore::from_env(&args.store_cert_role)
+            .await
+            .context("logging in to the secret store for the callout xkey")?,
+    )
+    .await?;
+    tracing::info!(public_key = %callout_xkey.public_key(), "loaded the callout xkey");
     let user_seed = read_secret(&args.user_seed_file)?;
     let client = async_nats::ConnectOptions::with_nkey(user_seed)
         .name("swarm-nats-auth")
@@ -202,15 +249,8 @@ async fn main() -> anyhow::Result<()> {
     );
 
     while let Some(msg) = requests.next().await {
-        // Decode failures are logged and dropped, never propagated: this loop
+        let Some((framing, req)) = open_request(&callout_xkey, &msg) else {
-        // is the swarm's login path, and exiting on one malformed payload
+            continue;
-        // would let any client take authentication down for everyone.
-        let req = match request::decode(&msg.payload) {
-            Ok(req) => req,
-            Err(e) => {
-                tracing::warn!(error = ?e, "undecodable auth request, ignoring");
-                continue;
-            }
         };
         // No token is a denial, not an error: an anonymous connect is a
         // normal thing for a client to attempt and an abnormal thing to
@@ -233,7 +273,7 @@ async fn main() -> anyhow::Result<()> {
                 (None, None)
             }
             Some(agent_token::Presented::Bearer(token)) => {
-                introspected(&policy, &http, &args, &client_secret, token).await
+                introspected(&policy, &http, args, &client_secret, token).await
             }
             None => (None, None),
         };
@@ -245,6 +285,7 @@ async fn main() -> anyhow::Result<()> {
             user_nkey = %req.user_nkey,
             server_id = %req.server_id.id,
             granted = permissions.is_some(),
+            sealed = framing.is_sealed(),
             caller = caller.as_deref().unwrap_or("-"),
             "auth request"
         );
@@ -267,13 +308,65 @@ async fn main() -> anyhow::Result<()> {
             ),
             None => respond::deny(&issuer, &req.server_id.id, &req.user_nkey),
         };
-        if let Err(e) = client.publish(reply_to, token.into()).await {
+        let reply = match framing.seal(&callout_xkey, token.into_bytes()) {
+            Ok(reply) => reply,
+            Err(e) => {
+                tracing::warn!(error = %e, "could not seal the auth response; dropping");
+                continue;
+            }
+        };
+        if let Err(e) = client.publish(reply_to, reply.into()).await {
             tracing::warn!(error = ?e, "failed to publish auth response");
         }
     }
     anyhow::bail!("subscription to {AUTH_SUBJECT} ended")
 }
 
+/// The request in `msg`, opened when the server sealed it, and the framing
+/// its reply takes. `None`, logged, for anything else.
+///
+/// Logged and dropped, never propagated: the loop calling this is the
+/// swarm's login path, and exiting on one malformed payload would let any
+/// client take authentication down for everyone.
+fn open_request(
+    callout_xkey: &nkeys::XKey,
+    msg: &async_nats::Message,
+) -> Option<(xkey::Framing, request::AuthRequest)> {
+    let server_xkey = msg
+        .headers
+        .as_ref()
+        .and_then(|h| h.get(xkey::SERVER_XKEY_HEADER))
+        .map(async_nats::HeaderValue::as_str);
+    let (framing, payload) = match xkey::Framing::open(callout_xkey, server_xkey, &msg.payload) {
+        Ok(opened) => opened,
+        Err(e) => {
+            tracing::warn!(error = %e, "unopenable auth request, ignoring");
+            return None;
+        }
+    };
+    match request::decode(&payload) {
+        Ok(req) => Some((framing, req)),
+        Err(e) => {
+            tracing::warn!(error = ?e, "undecodable auth request, ignoring");
+            None
+        }
+    }
+}
+
+/// The `ensure-xkey` verb: see [`xkey::ensure`].
+async fn ensure_xkey(store_cert_role: &str) -> anyhow::Result<()> {
+    let store = swarm_secret_client::SecretStore::from_env(store_cert_role)
+        .await
+        .context("logging in to the secret store")?;
+    let key = xkey::ensure(&store).await?;
+    tracing::info!(
+        public_key = %key.public_key(),
+        path = xkey::PUBLIC_PATH,
+        "the callout xkey is in the secret store"
+    );
+    Ok(())
+}
+
 /// The OIDC path: who the `IdP` says presented `token`, and what the policy
 /// grants that client. `None` permissions is a denial.
 ///
@@ -322,6 +415,14 @@ async fn introspected(
 mod tests {
     use super::*;
 
+    /// The responder's flags, parsed the way `main` parses them.
+    fn responder<const N: usize>(argv: [&str; N]) -> Result<Args, clap::Error> {
+        Cli::try_parse_from(argv).map(|cli| {
+            assert!(cli.command.is_none(), "no verb was given");
+            cli.responder.expect("with no verb, the responder's flags")
+        })
+    }
+
     /// The argument vector `swarm-nats.nix` actually builds, in the form it
     /// builds it: every flag space-separated, and the suffix's value opening
     /// with `-`. Read as a short-flag cluster it aborts before `main` runs,
@@ -329,7 +430,7 @@ mod tests {
     /// the operator as `authorization violation` on an unrelated process.
     #[test]
     fn the_units_own_argument_vector_parses() {
-        let args = Args::try_parse_from([
+        let args = responder([
             "swarm-nats-auth",
             "--user-seed-file",
             "/run/credentials/callout-user.seed",
@@ -366,7 +467,7 @@ mod tests {
     /// the argument being unknown.
     #[test]
     fn an_agent_suffix_without_a_leading_dash_parses_too() {
-        let args = Args::try_parse_from([
+        let args = responder([
             "swarm-nats-auth",
             "--user-seed-file",
             "/run/credentials/callout-user.seed",
@@ -387,7 +488,7 @@ mod tests {
     /// drift into disagreeing about what an agent client is called.
     #[test]
     fn the_default_suffix_is_what_the_module_passes() {
-        let args = Args::try_parse_from([
+        let args = responder([
             "swarm-nats-auth",
             "--user-seed-file",
             "/run/credentials/callout-user.seed",
@@ -401,4 +502,40 @@ mod tests {
         .expect("the required flags alone must parse");
         assert_eq!(args.agent_client_suffix, "-agent");
     }
+
+    /// The verb as `swarm-bao-nats-callout-xkey` runs it, with none of the
+    /// responder's required flags.
+    #[test]
+    fn the_ensure_verb_parses_as_the_unit_runs_it() {
+        let cli = Cli::try_parse_from([
+            "swarm-nats-auth",
+            "ensure-xkey",
+            "--store-cert-role",
+            "swarm-nats-auth",
+        ])
+        .expect("the verb needs no responder flags");
+        assert!(cli.responder.is_none());
+        assert!(matches!(
+            cli.command,
+            Some(Command::EnsureXkey { ref store_cert_role }) if store_cert_role == "swarm-nats-auth"
+        ));
+    }
+
+    /// A bare invocation is neither: the responder's flags stay required
+    /// when no verb is given.
+    #[test]
+    fn no_verb_and_no_responder_flags_is_refused() {
+        Cli::try_parse_from(["swarm-nats-auth"]).expect_err("the responder's flags are required");
+    }
+
+    #[test]
+    fn the_verb_does_not_take_the_responders_flags() {
+        Cli::try_parse_from([
+            "swarm-nats-auth",
+            "--user-seed-file",
+            "/run/credentials/callout-user.seed",
+            "ensure-xkey",
+        ])
+        .expect_err("a verb and the responder's flags conflict");
+    }
 }

swarm-nats-auth/src/xkey.rs

@@ -0,0 +1,410 @@
+//! The callout xkey: the curve keypair `nats-server` seals every auth request
+//! to, and that this responder seals every reply back with.
+//!
+//! Both halves live in the secret store. The seed is at [`SEED_PATH`], which
+//! only the `swarm-nats-auth` role reads. The public half is at
+//! [`PUBLIC_PATH`], under `swarm/public/`, which every token the store issues
+//! may read; the queue's host renders it into `auth_callout.xkey` from there.
+//!
+//! [`ensure`] creates the pair when the seed is absent and otherwise only
+//! reads it. [`load`] only ever reads, and is what the responder uses: a
+//! responder that minted a new pair on its own start would hold a key the
+//! running server does not know, and every request would fail to open.
+
+use std::future::Future;
+
+use anyhow::{Context, Result, anyhow};
+use nkeys::XKey;
+use swarm_secret_client::SecretStore;
+
+/// The seed's path under the store's KV mount. The nix half is
+/// `xkeySeedLeaf` in `nix/host-modules/swarm-bao.nix`.
+pub const SEED_PATH: &str = "swarm/services/swarm-nats/callout/xkey-seed";
+
+/// The public half's path under the store's KV mount. The nix half is
+/// `xkeyPublicLeaf` in `nix/host-modules/swarm-bao.nix`, and the fetch in
+/// `nix/host-modules/swarm-nats.nix` reads it.
+pub const PUBLIC_PATH: &str = "swarm/public/swarm-nats/callout-xkey";
+
+/// The header `nats-server` sets on a request it sealed. Its value is the
+/// server's own public xkey, which the reply is sealed to.
+pub const SERVER_XKEY_HEADER: &str = "Nats-Server-Xkey";
+
+/// The two operations [`ensure`] and [`load`] need from the store, so a test
+/// can stand in for it.
+pub trait KeyStore {
+    /// The value at `path`, or `None` when nothing is stored there.
+    fn read(&self, path: &str) -> impl Future<Output = Result<Option<String>>> + Send;
+    /// Store `value` at `path`, failing when anything is already there.
+    fn create(&self, path: &str, value: &str) -> impl Future<Output = Result<()>> + Send;
+}
+
+/// One field, `value`, as every single-value secret in the store is kept.
+/// No `Debug`: one of the values held here is the seed.
+#[derive(serde::Serialize, serde::Deserialize)]
+struct Stored {
+    value: String,
+}
+
+impl KeyStore for SecretStore {
+    async fn read(&self, path: &str) -> Result<Option<String>> {
+        let stored: Option<Stored> = self
+            .read_optional(path)
+            .await
+            .with_context(|| format!("reading {path}"))?;
+        Ok(stored.map(|s| s.value))
+    }
+
+    async fn create(&self, path: &str, value: &str) -> Result<()> {
+        SecretStore::create(
+            self,
+            path,
+            &Stored {
+                value: value.to_owned(),
+            },
+        )
+        .await
+        .with_context(|| format!("creating {path}"))
+    }
+}
+
+/// The stored pair, created first when no seed is stored.
+///
+/// Safe to run from two places at once. Each writes with a create that fails
+/// once the path holds anything, so one seed lands, and the run whose write
+/// failed reads that one back. The public half is written the same way, and
+/// is written even when the seed already exists, so a run that stopped
+/// between the two writes is completed by the next.
+///
+/// # Errors
+/// When the store cannot be read or written, when the stored seed is not a
+/// curve seed, and when the stored public half is not the seed's.
+pub async fn ensure(store: &impl KeyStore) -> Result<XKey> {
+    let key = match store.read(SEED_PATH).await? {
+        Some(seed) => from_seed(&seed)?,
+        None => create_seed(store).await?,
+    };
+    publish(store, &key.public_key()).await?;
+    Ok(key)
+}
+
+/// The stored pair. Never writes.
+///
+/// # Errors
+/// When the store cannot be read, holds no seed, or holds one that is not a
+/// curve seed.
+pub async fn load(store: &impl KeyStore) -> Result<XKey> {
+    let seed = store.read(SEED_PATH).await?.with_context(|| {
+        format!("nothing is stored at {SEED_PATH}; swarm-bao-nats-callout-xkey creates it")
+    })?;
+    from_seed(&seed)
+}
+
+async fn create_seed(store: &impl KeyStore) -> Result<XKey> {
+    let fresh = XKey::new();
+    let seed = fresh
+        .seed()
+        .map_err(|_| anyhow!("a freshly generated xkey has no seed"))?;
+    match store.create(SEED_PATH, &seed).await {
+        Ok(()) => {
+            tracing::info!(public_key = %fresh.public_key(), "created the callout xkey");
+            Ok(fresh)
+        }
+        Err(e) => match store.read(SEED_PATH).await? {
+            Some(theirs) => from_seed(&theirs),
+            None => Err(e),
+        },
+    }
+}
+
+async fn publish(store: &impl KeyStore, public: &str) -> Result<()> {
+    let stored = match store.read(PUBLIC_PATH).await? {
+        Some(stored) => stored,
+        None => match store.create(PUBLIC_PATH, public).await {
+            Ok(()) => return Ok(()),
+            Err(e) => store.read(PUBLIC_PATH).await?.ok_or(e)?,
+        },
+    };
+    anyhow::ensure!(
+        stored == public,
+        "{PUBLIC_PATH} holds a public key that is not the one for the seed at {SEED_PATH}"
+    );
+    Ok(())
+}
+
+/// Never carries the parse error: it is about the bytes of the seed.
+fn from_seed(seed: &str) -> Result<XKey> {
+    XKey::from_seed(seed.trim()).map_err(|_| anyhow!("{SEED_PATH} does not hold a curve seed"))
+}
+
+/// How one request arrived, which is how its reply has to leave.
+pub enum Framing {
+    /// No [`SERVER_XKEY_HEADER`]: the server reads the reply as it is.
+    Plain,
+    /// Sealed to this responder by the server key the header named.
+    Sealed(XKey),
+}
+
+impl Framing {
+    /// `payload` as the server sent it, opened when `server_key` says it was
+    /// sealed.
+    ///
+    /// # Errors
+    /// When the header is not a public xkey, or the payload does not open
+    /// with `own`.
+    pub fn open(own: &XKey, server_key: Option<&str>, payload: &[u8]) -> Result<(Self, Vec<u8>)> {
+        let Some(server_key) = server_key else {
+            return Ok((Self::Plain, payload.to_vec()));
+        };
+        let server = XKey::from_public_key(server_key)
+            .map_err(|_| anyhow!("{SERVER_XKEY_HEADER} is not a public xkey"))?;
+        let opened = own
+            .open(payload, &server)
+            .map_err(|_| anyhow!("the request does not open with this responder's xkey"))?;
+        Ok((Self::Sealed(server), opened))
+    }
+
+    /// `reply` framed the way the request was.
+    ///
+    /// # Errors
+    /// When sealing fails, which it does only for a key with no seed.
+    pub fn seal(&self, own: &XKey, reply: Vec<u8>) -> Result<Vec<u8>> {
+        match self {
+            Self::Plain => Ok(reply),
+            Self::Sealed(server) => own
+                .seal(&reply, server)
+                .map_err(|_| anyhow!("sealing the reply")),
+        }
+    }
+
+    /// Whether the reply leaves sealed.
+    pub fn is_sealed(&self) -> bool {
+        matches!(self, Self::Sealed(_))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+    use std::sync::Mutex;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    use super::*;
+
+    /// A store that keeps values in memory and counts writes. `rival` is a
+    /// value another start lands at a path between this one's read and its
+    /// create, which the create then loses to.
+    #[derive(Default)]
+    struct Fake {
+        values: Mutex<HashMap<String, String>>,
+        rival: Mutex<HashMap<String, String>>,
+        writes: AtomicUsize,
+    }
+
+    impl Fake {
+        fn holding(entries: &[(&str, &str)]) -> Self {
+            let fake = Self::default();
+            for (path, value) in entries {
+                fake.put(path, value);
+            }
+            fake
+        }
+
+        fn put(&self, path: &str, value: &str) {
+            self.values
+                .lock()
+                .expect("unpoisoned")
+                .insert(path.to_owned(), value.to_owned());
+        }
+
+        fn get(&self, path: &str) -> Option<String> {
+            self.values.lock().expect("unpoisoned").get(path).cloned()
+        }
+    }
+
+    impl KeyStore for Fake {
+        async fn read(&self, path: &str) -> Result<Option<String>> {
+            Ok(self.get(path))
+        }
+
+        async fn create(&self, path: &str, value: &str) -> Result<()> {
+            if let Some(theirs) = self.rival.lock().expect("unpoisoned").remove(path) {
+                self.put(path, &theirs);
+            }
+            if self.get(path).is_some() {
+                anyhow::bail!("403 permission denied");
+            }
+            self.writes.fetch_add(1, Ordering::SeqCst);
+            self.put(path, value);
+            Ok(())
+        }
+    }
+
+    fn seed_of(key: &XKey) -> String {
+        key.seed().expect("a generated key has a seed")
+    }
+
+    #[tokio::test]
+    async fn an_empty_store_gets_a_seed_and_its_public_half() {
+        let store = Fake::default();
+        let key = ensure(&store).await.expect("ensured");
+        assert_eq!(store.get(SEED_PATH), Some(seed_of(&key)));
+        assert_eq!(store.get(PUBLIC_PATH), Some(key.public_key()));
+        assert!(key.public_key().starts_with('X'), "a curve public key");
+        assert_eq!(store.writes.load(Ordering::SeqCst), 2);
+    }
+
+    #[tokio::test]
+    async fn an_existing_pair_is_only_read() {
+        let existing = XKey::new();
+        let store = Fake::holding(&[
+            (SEED_PATH, &seed_of(&existing)),
+            (PUBLIC_PATH, &existing.public_key()),
+        ]);
+        let key = ensure(&store).await.expect("ensured");
+        assert_eq!(key.public_key(), existing.public_key());
+        assert_eq!(store.writes.load(Ordering::SeqCst), 0);
+    }
+
+    /// A run that stopped between its two writes left a seed and no public
+    /// half; the next one writes the public half for that seed.
+    #[tokio::test]
+    async fn an_existing_seed_without_its_public_half_gets_one() {
+        let existing = XKey::new();
+        let store = Fake::holding(&[(SEED_PATH, &seed_of(&existing))]);
+        let key = ensure(&store).await.expect("ensured");
+        assert_eq!(key.public_key(), existing.public_key());
+        assert_eq!(store.get(PUBLIC_PATH), Some(existing.public_key()));
+        assert_eq!(store.writes.load(Ordering::SeqCst), 1);
+    }
+
+    /// Two starts read an empty store at once and both generate; the one
+    /// whose create fails takes the seed that landed.
+    #[tokio::test]
+    async fn a_start_that_loses_the_create_race_takes_the_winners_seed() {
+        let winner = XKey::new();
+        let store = Fake::default();
+        store
+            .rival
+            .lock()
+            .expect("unpoisoned")
+            .insert(SEED_PATH.to_owned(), seed_of(&winner));
+        store
+            .rival
+            .lock()
+            .expect("unpoisoned")
+            .insert(PUBLIC_PATH.to_owned(), winner.public_key());
+        let key = ensure(&store).await.expect("ensured");
+        assert_eq!(key.public_key(), winner.public_key());
+        assert_eq!(store.get(SEED_PATH), Some(seed_of(&winner)));
+        assert_eq!(store.get(PUBLIC_PATH), Some(winner.public_key()));
+        assert_eq!(store.writes.load(Ordering::SeqCst), 0);
+    }
+
+    #[tokio::test]
+    async fn a_public_half_that_is_not_the_seeds_is_refused() {
+        let existing = XKey::new();
+        let store = Fake::holding(&[
+            (SEED_PATH, &seed_of(&existing)),
+            (PUBLIC_PATH, &XKey::new().public_key()),
+        ]);
+        assert!(ensure(&store).await.is_err());
+    }
+
+    #[tokio::test]
+    async fn a_seed_that_is_not_a_curve_seed_is_refused_without_quoting_it() {
+        let account = nkeys::KeyPair::new_account();
+        let not_curve = account.seed().expect("a seed");
+        let store = Fake::holding(&[(SEED_PATH, &not_curve)]);
+        let e = ensure(&store).await.expect_err("an account seed");
+        assert!(!format!("{e:#}").contains(&not_curve), "{e:#}");
+        let e = load(&store).await.expect_err("an account seed");
+        assert!(!format!("{e:#}").contains(&not_curve), "{e:#}");
+    }
+
+    #[tokio::test]
+    async fn loading_reads_the_stored_seed_and_never_creates_one() {
+        let store = Fake::default();
+        assert!(load(&store).await.is_err());
+        assert_eq!(store.writes.load(Ordering::SeqCst), 0);
+        assert_eq!(store.get(SEED_PATH), None);
+
+        let existing = XKey::new();
+        store.put(SEED_PATH, &seed_of(&existing));
+        let key = load(&store).await.expect("loaded");
+        assert_eq!(key.public_key(), existing.public_key());
+    }
+
+    fn request_jwt() -> Vec<u8> {
+        let b = |s: &str| data_encoding::BASE64URL_NOPAD.encode(s.as_bytes());
+        format!(
+            "{}.{}.{}",
+            b("{}"),
+            b(r#"{"nats":{"connect_opts":{"auth_token":"t0k"},"user_nkey":"UCLIENT","server_id":{"id":"NSERVER"}}}"#),
+            b("sig"),
+        )
+        .into_bytes()
+    }
+
+    /// The exchange as the server runs it: it seals the request to our public
+    /// key under a key of its own, which it names in the header, and opens
+    /// our reply with that same key.
+    #[test]
+    fn a_sealed_request_opens_and_its_grant_is_sealed_back_to_the_server() {
+        let own = XKey::new();
+        let server = XKey::new();
+        let own_public = XKey::from_public_key(&own.public_key()).expect("public");
+        let sealed = server.seal(&request_jwt(), &own_public).expect("sealed");
+
+        let (framing, opened) =
+            Framing::open(&own, Some(&server.public_key()), &sealed).expect("opens");
+        assert!(framing.is_sealed());
+        let req = crate::request::decode(&opened).expect("a request");
+        assert_eq!(req.user_nkey, "UCLIENT");
+
+        let issuer = nkeys::KeyPair::new_account();
+        let permissions = crate::policy::Permissions {
+            publish: vec!["$KV.hive-status.alpha".to_owned()],
+            subscribe: None,
+        };
+        let token = crate::respond::grant(
+            &issuer,
+            "APP",
+            &req.server_id.id,
+            &req.user_nkey,
+            &permissions,
+        );
+        let reply = framing
+            .seal(&own, token.clone().into_bytes())
+            .expect("sealed");
+        assert_ne!(reply, token.as_bytes(), "the reply leaves sealed");
+        let read_by_server = server
+            .open(&reply, &own_public)
+            .expect("the server opens it");
+        assert_eq!(read_by_server, token.as_bytes());
+    }
+
+    /// A server with no xkey configured sends no header and reads the reply
+    /// as it is.
+    #[test]
+    fn a_request_without_the_header_is_answered_in_plaintext() {
+        let own = XKey::new();
+        let (framing, opened) = Framing::open(&own, None, &request_jwt()).expect("plain");
+        assert!(!framing.is_sealed());
+        assert_eq!(opened, request_jwt());
+        let reply = framing.seal(&own, b"a.b.c".to_vec()).expect("plain");
+        assert_eq!(reply, b"a.b.c");
+    }
+
+    #[test]
+    fn a_request_sealed_to_another_key_does_not_open() {
+        let own = XKey::new();
+        let server = XKey::new();
+        let someone_else = XKey::from_public_key(&XKey::new().public_key()).expect("public");
+        let sealed = server.seal(&request_jwt(), &someone_else).expect("sealed");
+        assert!(Framing::open(&own, Some(&server.public_key()), &sealed).is_err());
+        // The header present over a payload that was never sealed.
+        assert!(Framing::open(&own, Some(&server.public_key()), &request_jwt()).is_err());
+        assert!(Framing::open(&own, Some("not-a-key"), &sealed).is_err());
+    }
+}

swarm-secret-client/src/client.rs

@@ -237,6 +237,23 @@ impl SecretStore {
         Ok(())
     }
 
+    /// Write `value` at `path` only if nothing is stored there yet.
+    ///
+    /// KV v2's check-and-set with `cas=0`, so of two writers racing on an
+    /// empty path exactly one lands. The other gets an error, and a caller
+    /// that only needs *a* value there reads the winner's back.
+    ///
+    /// # Errors
+    /// [`Error::Vault`] when something is already stored at `path`, or the
+    /// token's policy does not cover it. Under a grant of `create` without
+    /// `update` the first of those is a 403, not a check-and-set mismatch:
+    /// the store checks the capability for an existing path before it gets
+    /// to `cas`.
+    pub async fn create<T: Serialize + Sync>(&self, path: &str, value: &T) -> Result<(), Error> {
+        vaultrs::kv2::set_with_options(&self.inner, MOUNT, path, value, create_only()).await?;
+        Ok(())
+    }
+
     /// Delete `path` outright: every version of it, and the metadata that
     /// lists them.
     ///
@@ -395,6 +412,12 @@ struct WriteAclPolicy {
     policy: String,
 }
 
+/// The options [`SecretStore::create`] writes with: version 0, which only a
+/// path holding nothing is at.
+fn create_only() -> vaultrs::api::kv2::requests::SetSecretRequestOptions {
+    vaultrs::api::kv2::requests::SetSecretRequestOptions { cas: 0 }
+}
+
 /// The request [`SecretStore::issue_client_certificate`] sends.
 ///
 /// The name stays out of the SANs so the certificate carries exactly one name,
@@ -627,6 +650,30 @@ mod tests {
         }
     }
 
+    /// A create is a write with `cas=0` in the body: without it the store
+    /// takes the write as an overwrite of whatever is there, and the two
+    /// racing writers [`SecretStore::create`] exists for would both land.
+    #[test]
+    fn a_create_writes_the_data_path_only_while_it_holds_nothing() {
+        use rustify::endpoint::Endpoint as _;
+
+        let request = vaultrs::api::kv2::requests::SetSecretRequest::builder()
+            .mount(MOUNT)
+            .path("swarm/public/x")
+            .data(serde_json::json!({ "value": "v" }))
+            .options(create_only())
+            .build()
+            .expect("the request builds");
+        assert_eq!(request.path(), "secret/data/swarm/public/x");
+        let body = request
+            .body()
+            .expect("the body serialises")
+            .expect("a write sends one");
+        let sent: serde_json::Value = serde_json::from_slice(&body).expect("JSON");
+        assert_eq!(sent["options"]["cas"], 0);
+        assert_eq!(sent["data"]["value"], "v");
+    }
+
     /// What makes [`SecretStore::delete_all_versions`] safe to run twice: the
     /// second run finds nothing and must still succeed. A teardown that
     /// hard-errors on an already-revoked credential is a worse fault than the