fix(crypto): make key rotation actually rotate, and prove reseal is interruption-safe

bin/migrate-keys/Cargo.toml

@@ -8,6 +8,9 @@ license.workspace = true
 repository.workspace = true
 authors.workspace = true
 
+[lib]
+path = "src/lib.rs"
+
 [[bin]]
 name = "octo-migrate-keys"
 path = "src/main.rs"
@@ -24,3 +27,6 @@ uuid.workspace = true
 sha2.workspace = true
 hex.workspace = true
 dotenvy = "0.15"
+
+[dev-dependencies]
+sqlx.workspace = true

bin/migrate-keys/src/lib.rs

@@ -0,0 +1,107 @@
+//! Core of `octo-migrate-keys`: re-seal every sealed wallet seed from `old_key` to `new_key`.
+//!
+//! Split from `main.rs` so the interruption-safety tests can drive the exact production loop.
+//!
+//! ## Which rows need work
+//!
+//! Every record is tagged `SCHEME_V1` whichever key sealed it, so the scheme tag cannot say
+//! whether a row is already rotated. AES-GCM authentication can: a row that opens under
+//! `new_key` is done and is skipped; anything else is opened under `old_key` and re-sealed.
+//! A row that opens under neither aborts the run — it needs an operator, not a silent skip.
+//!
+//! ## Interruption safety
+//!
+//! Each row is written by one `UPDATE` of all four sealed fields (`Store::reseal_wallet`), so an
+//! interrupted run leaves every row wholly old or wholly new. Re-running resumes: rotated rows
+//! open under `new_key` and are skipped.
+
+#![forbid(unsafe_code)]
+
+use anyhow::{Context, Result};
+use octo_crypto::{open, reseal, SealedSeed, MASTER_KEY_LEN, SCHEME_V1};
+use octo_store::Store;
+use uuid::Uuid;
+
+/// Totals from one run.
+#[derive(Debug, Default, PartialEq, Eq)]
+pub struct Summary {
+    /// Rows re-sealed under the new key by this run.
+    pub migrated: usize,
+    /// Rows already on the new key (or changed concurrently) and left untouched.
+    pub skipped: usize,
+}
+
+/// Re-seal every sealed wallet from `old_key` to `new_key`, `batch_size` rows per page.
+///
+/// `before_write` runs after a row is re-sealed in memory and before it is persisted — the
+/// window a crash would hit. Production passes a no-op; tests use it to inject a failure there.
+pub async fn migrate(
+    store: &Store,
+    old_key: &[u8; MASTER_KEY_LEN],
+    new_key: &[u8; MASTER_KEY_LEN],
+    batch_size: i64,
+    mut before_write: impl FnMut(Uuid) -> Result<()>,
+) -> Result<Summary> {
+    let mut summary = Summary::default();
+    let mut after_id: Option<Uuid> = None;
+
+    loop {
+        let batch = store
+            .list_sealed_wallets(batch_size, after_id)
+            .await
+            .context("list_sealed_wallets")?;
+        let Some(last) = batch.last() else { break };
+        after_id = Some(last.id);
+        tracing::info!(batch_len = batch.len(), "processing batch");
+
+        for wallet in &batch {
+            // Client-custody wallets hold no server-side seed; only sealed rows are rotated.
+            let (Some(ciphertext), Some(nonce), Some(salt)) = (
+                wallet.sealed_ciphertext.as_ref(),
+                wallet.sealed_nonce.as_ref(),
+                wallet.sealed_salt.as_ref(),
+            ) else {
+                continue;
+            };
+            let scheme = wallet.sealed_scheme.unwrap_or(SCHEME_V1 as i16);
+            let sealed = SealedSeed::from_parts_with_scheme(
+                ciphertext.clone(),
+                nonce,
+                salt,
+                u8::try_from(scheme).context("sealed_scheme out of range")?,
+            )
+            .with_context(|| format!("from_parts wallet {}", wallet.id))?;
+            // Context is the network string bound into the AEAD AAD (e.g. "octo:mainnet").
+            let context = format!("octo:{}", wallet.network);
+
+            // Already rotated (or old == new on a current-scheme row): nothing to do.
+            if open(new_key, &sealed, context.as_bytes()).is_ok() {
+                summary.skipped += 1;
+                continue;
+            }
+
+            let new_sealed = reseal(old_key, new_key, &sealed, context.as_bytes())
+                .with_context(|| format!("wallet {} opens under neither key", wallet.id))?;
+
+            before_write(wallet.id)?;
+
+            let updated = store
+                .reseal_wallet(
+                    wallet.id,
+                    &new_sealed.ciphertext,
+                    &new_sealed.nonce,
+                    &new_sealed.salt,
+                    i16::from(new_sealed.scheme),
+                    ciphertext,
+                )
+                .await
+                .with_context(|| format!("reseal_wallet DB update for {}", wallet.id))?;
+            if updated {
+                summary.migrated += 1;
+            } else {
+                summary.skipped += 1;
+            }
+        }
+    }
+    Ok(summary)
+}

bin/migrate-keys/src/main.rs

@@ -25,17 +25,16 @@
 //!
 //! ## Idempotency
 //!
-//! The store method `reseal_wallet` only updates a row when its current `sealed_scheme` matches
-//! the expected "old" scheme. Re-running the tool against a fully-migrated database is safe and
-//! produces 0 updates.
+//! A row that already opens under the new key is skipped, and `reseal_wallet` only writes when
+//! the row still holds the ciphertext that was read (compare-and-swap). Re-running the tool
+//! against a fully-migrated database is safe and produces 0 updates. See `lib.rs`.
 //!
 //! ## Rollback
 //!
-//! Old-scheme and new-scheme records can coexist in the database indefinitely because every open
-//! call reads the scheme tag from the row and picks the correct key. To abort a rotation, simply
-//! stop the tool; already-migrated rows remain openable with the new key, un-migrated rows remain
-//! openable with the old key. Rolling back a completed rotation requires running the tool again
-//! with the old and new keys swapped.
+//! Old-key and new-key records can coexist indefinitely: during the rotation window the server
+//! tries `MASTER_KEY_NEXT` first and falls back to `MASTER_KEY` (AES-GCM authentication tells
+//! them apart). To abort, stop the tool. Rolling back a completed rotation means running the tool
+//! again with the old and new keys swapped.
 //!
 //! ## Usage
 //!
@@ -69,7 +68,7 @@
 
 use anyhow::{Context, Result};
 use base64::Engine;
-use octo_crypto::{master_key_from_slice, reseal, MASTER_KEY_LEN, SCHEME_V1};
+use octo_crypto::{master_key_from_slice, MASTER_KEY_LEN};
 use octo_store::Store;
 use sha2::{Digest, Sha256};
 use std::path::{Path, PathBuf};

bin/migrate-keys/tests/interruption_tests.rs

@@ -0,0 +1,165 @@
+//! Interruption-safety tests for the key-rotation reseal path (`octo_migrate_keys::migrate`).
+//!
+//! Requires Postgres via `DATABASE_URL` (loaded from `.env`); skipped with a message otherwise.
+//! Each test runs in its own throwaway database so the whole wallets table is under its control.
+
+use octo_crypto::{open, seal, SealedSeed, MASTER_KEY_LEN};
+use octo_migrate_keys::{migrate, Summary};
+use octo_store::{NewWallet, Store};
+use sqlx::{Connection, Executor, PgConnection};
+use uuid::Uuid;
+
+const OLD_KEY: [u8; MASTER_KEY_LEN] = [1u8; MASTER_KEY_LEN];
+const NEW_KEY: [u8; MASTER_KEY_LEN] = [2u8; MASTER_KEY_LEN];
+const CONTEXT: &[u8] = b"octo:testnet";
+const WALLETS: usize = 12;
+/// The injected crash hits the write of the Nth wallet (0-based), so N rows are already rotated.
+const CRASH_AT: usize = 5;
+
+/// A fresh, migrated database; returns its store and name (for dropping).
+async fn fresh_store() -> Option<(Store, String, String)> {
+    let _ = dotenvy::dotenv();
+    let Ok(url) = std::env::var("DATABASE_URL") else {
+        eprintln!("SKIPPED: DATABASE_URL is not set");
+        return None;
+    };
+    let name = format!("octo_mk_{}", Uuid::new_v4().simple());
+    let mut admin = PgConnection::connect(&url).await.expect("connect admin");
+    admin
+        .execute(format!(r#"CREATE DATABASE "{name}""#).as_str())
+        .await
+        .expect("create db");
+    let base = url.split('?').next().unwrap();
+    let db_url = format!("{}/{name}", base.rsplit_once('/').unwrap().0);
+    let store = Store::connect(&db_url).await.expect("connect test db");
+    store.migrate().await.expect("migrate");
+    Some((store, url, name))
+}
+
+async fn drop_db(store: Store, admin_url: &str, name: &str) {
+    drop(store);
+    let mut admin = PgConnection::connect(admin_url)
+        .await
+        .expect("connect admin");
+    let _ = admin
+        .execute(format!(r#"DROP DATABASE IF EXISTS "{name}" WITH (FORCE)"#).as_str())
+        .await;
+}
+
+/// Seed `WALLETS` wallets whose seeds are sealed under `OLD_KEY`; returns (id, plaintext).
+async fn seed_wallets(store: &Store) -> Vec<(Uuid, Vec<u8>)> {
+    let mut out = Vec::new();
+    for i in 0..WALLETS {
+        let secret = format!("seed-{i}-{}", Uuid::new_v4()).into_bytes();
+        let sealed = seal(&OLD_KEY, &secret, CONTEXT).unwrap();
+        let account = format!("G{}", Uuid::new_v4().simple());
+        let w = store
+            .create_wallet(NewWallet {
+                network: "testnet",
+                stellar_account_g: &account,
+                sealed_ciphertext: &sealed.ciphertext,
+                sealed_nonce: &sealed.nonce,
+                sealed_salt: &sealed.salt,
+                sealed_scheme: i16::from(sealed.scheme),
+                label: None,
+                user_id: None,
+                description: None,
+            })
+            .await
+            .unwrap();
+        out.push((w.id, secret));
+    }
+    out
+}
+
+/// Which key a row's four sealed fields open under, asserting it is exactly one and that the
+/// plaintext is intact — i.e. the row is wholly pre- or post-migration, never a mix.
+async fn row_key(store: &Store, id: Uuid, secret: &[u8]) -> &'static str {
+    let w = store.get_wallet(id).await.unwrap();
+    let sealed = SealedSeed::from_parts_with_scheme(
+        w.sealed_ciphertext.unwrap(),
+        &w.sealed_nonce.unwrap(),
+        &w.sealed_salt.unwrap(),
+        u8::try_from(w.sealed_scheme.unwrap()).unwrap(),
+    )
+    .unwrap();
+    let old = open(&OLD_KEY, &sealed, CONTEXT).ok();
+    let new = open(&NEW_KEY, &sealed, CONTEXT).ok();
+    match (old, new) {
+        (Some(p), None) if p.as_slice() == secret => "old",
+        (None, Some(p)) if p.as_slice() == secret => "new",
+        _ => panic!("wallet {id} has mismatched sealed fields (opens under neither/both keys)"),
+    }
+}
+
+/// Run `migrate` in its own task with a hook that panics on the `CRASH_AT`-th write — the moment
+/// between re-sealing in memory and persisting, i.e. a process killed mid-batch.
+async fn run_and_crash(store: &Store) {
+    let store = store.clone();
+    let handle = tokio::spawn(async move {
+        let mut writes = 0;
+        // Small batches so the crash lands mid-run, across batch boundaries.
+        migrate(&store, &OLD_KEY, &NEW_KEY, 4, |_| {
+            if writes == CRASH_AT {
+                panic!("simulated crash mid-batch");
+            }
+            writes += 1;
+            Ok(())
+        })
+        .await
+    });
+    assert!(handle.await.unwrap_err().is_panic(), "the run must crash");
+}
+
+#[tokio::test]
+async fn interrupting_migrate_keys_mid_batch_never_leaves_a_wallet_row_with_mismatched_scheme_and_ciphertext(
+) {
+    let Some((store, admin_url, name)) = fresh_store().await else {
+        return;
+    };
+    let wallets = seed_wallets(&store).await;
+
+    run_and_crash(&store).await;
+
+    let mut new = 0;
+    for (id, secret) in &wallets {
+        if row_key(&store, *id, secret).await == "new" {
+            new += 1;
+        }
+    }
+    assert_eq!(
+        new, CRASH_AT,
+        "exactly the rows written before the crash are rotated"
+    );
+    drop_db(store, &admin_url, &name).await;
+}
+
+#[tokio::test]
+async fn a_second_clean_run_after_interruption_completes_the_remaining_wallets_correctly() {
+    let Some((store, admin_url, name)) = fresh_store().await else {
+        return;
+    };
+    let wallets = seed_wallets(&store).await;
+    run_and_crash(&store).await;
+
+    let summary = migrate(&store, &OLD_KEY, &NEW_KEY, 4, |_| Ok(()))
+        .await
+        .unwrap();
+    assert_eq!(
+        summary,
+        Summary {
+            migrated: WALLETS - CRASH_AT,
+            skipped: CRASH_AT
+        }
+    );
+    for (id, secret) in &wallets {
+        assert_eq!(row_key(&store, *id, secret).await, "new");
+    }
+
+    // A third run is a no-op.
+    let summary = migrate(&store, &OLD_KEY, &NEW_KEY, 4, |_| Ok(()))
+        .await
+        .unwrap();
+    assert_eq!(summary.migrated, 0);
+    drop_db(store, &admin_url, &name).await;
+}

crates/api/src/routes/sponsor.rs

@@ -133,8 +133,7 @@ pub async fn sponsor(
                 .into(),
         ));
     };
-    // Keep the versioned-scheme path (PR #158) so master-key rotation keeps working. Rows
-    // written before the scheme tag existed fall back to V1.
+    // Rows written before the scheme tag existed fall back to V1.
     let scheme = wallet
         .sealed_scheme
         .unwrap_or(octo_crypto::SCHEME_V1 as i16);
@@ -144,15 +143,13 @@ pub async fn sponsor(
         inner_xdr: &inner_xdr,
         max_base_fee_stroops: max_fee,
     };
-    let signed = match sign_fee_bump(
-        state.master_key_for_scheme(scheme),
-        &sealed,
-        state.network(),
-        0,
-        &fb,
-    ) {
-        Ok(s) => s,
-        Err(_) => {
+    // During a rotation the row may be sealed under either key; AES-GCM tells us which.
+    let signed = match state
+        .opening_keys()
+        .find_map(|key| sign_fee_bump(key, &sealed, state.network(), 0, &fb).ok())
+    {
+        Some(s) => s,
+        None => {
             let _ = state
                 .store()
                 .finalize_sponsored_transaction(reserved.id, "failed", None, Some("signing failed"))

crates/api/src/routes/wallets.rs

@@ -323,7 +323,7 @@ pub async fn create_gas_tank(
 
     // Provision a fresh keypair inside wallet-core. The mnemonic is deliberately dropped: the
     // tank is a disposable fee account, recoverable only by re-provisioning.
-    let provisioned = octo_wallet_core::provision_wallet(state.master_key(), state.network())?;
+    let provisioned = octo_wallet_core::provision_wallet(state.sealing_key(), state.network())?;
     let wallet = state
         .store()
         .set_gas_tank(