kestrove-core/crates/kv-sync/tests/sync.rs

//! End-to-end sync tests against the in-memory fake server (wiremock).

use std::sync::Arc;

use kv_crypto::{
    aad::{Aad, ObjectType},
    aead::SymKey,
    keys::AccountKeyset,
};
use kv_proto::{DeviceEnroll, ItemRecord, Platform};
use kv_store::{MemoryStore, Store, SyncState, VaultRow};
use kv_sync::{
    engine::{open_record, seal_record},
    signing::SoftwareSigner,
    test_server::{TestServer, SETUP_TOKEN},
    ApiClient, DeviceSigner, LoginOutcome, ServerConfig, SyncContext, SyncError, SyncReport,
};
use serde_json::{json, Value};
use uuid::Uuid;

const AUTH: [u8; 32] = [7; 32];

fn test_kdf() -> kv_proto::KdfParams {
    kv_crypto::kdf::default_params()
}

struct Device {
    store: MemoryStore,
    api: ApiClient,
    keyset: AccountKeyset,
    signer: Arc<SoftwareSigner>,
}

fn copy_keyset(ks: &AccountKeyset) -> AccountKeyset {
    let k = SymKey::generate();
    AccountKeyset::unwrap(ks.account_id, &k, &ks.wrap(&k).unwrap()).unwrap()
}

fn enroll(signer: &SoftwareSigner) -> DeviceEnroll {
    DeviceEnroll {
        device_id: signer.device_id(),
        name: "test".into(),
        platform: Platform::Linux,
        p256_pk: signer.public_key(),
    }
}

impl Device {
    fn new(server: &TestServer, keyset: AccountKeyset) -> Self {
        let signer = Arc::new(SoftwareSigner::generate(Uuid::new_v4()));
        let api = ApiClient::new(&ServerConfig::new(server.url(), vec![]), signer.clone()).unwrap();
        Self { store: MemoryStore::new(), api, keyset, signer }
    }

    /// First device: registers the account and creates a vault locally.
    async fn first(server: &TestServer) -> (Self, Uuid) {
        let keyset = AccountKeyset::generate(Uuid::new_v4());
        let d = Self::new(server, keyset);
        let wrapped = d.keyset.wrap(&SymKey::generate()).unwrap();
        d.api
            .register(
                SETUP_TOKEN,
                d.keyset.account_id,
                &test_kdf(),
                &AUTH,
                d.keyset.enc.public_bytes(),
                d.keyset.sig_public().to_vec(),
                wrapped,
                enroll(&d.signer),
            )
            .await
            .unwrap();
        assert!(matches!(d.login().await, LoginOutcome::Ok { .. }));
        let vid = Uuid::new_v4();
        let (vk, sealed) = d.keyset.new_vault_key(vid).unwrap();
        let meta = vk
            .seal(br#"{"name":"Personal"}"#, &Aad::build(ObjectType::VaultMeta, d.keyset.account_id, Some(vid), None))
            .unwrap();
        d.store.put_vault(&VaultRow::new(vid, sealed, Some(meta))).unwrap();
        (d, vid)
    }

    async fn second(server: &TestServer, first: &Device) -> Self {
        let d = Self::new(server, copy_keyset(&first.keyset));
        assert!(matches!(d.login().await, LoginOutcome::Ok { .. }));
        d
    }

    async fn login(&self) -> LoginOutcome {
        self.api.login(self.keyset.account_id, &AUTH, enroll(&self.signer), None).await.unwrap()
    }

    fn ctx(&self) -> SyncContext<'_> {
        SyncContext { api: &self.api, store: &self.store, keyset: &self.keyset, device_id: self.signer.device_id() }
    }

    async fn sync(&self) -> Result<SyncReport, SyncError> {
        self.ctx().sync_all().await
    }

    fn vk(&self, vid: Uuid) -> SymKey {
        self.keyset.open_vault_key(vid, &self.store.vault(vid).unwrap().unwrap().sealed_key).unwrap()
    }

    fn save(&self, vid: Uuid, id: Uuid, item: Value) {
        let rec = seal_record(self.keyset.account_id, vid, &self.vk(vid), id, &item, self.signer.device_id()).unwrap();
        self.store.put_local(vid, &rec).unwrap();
    }

    fn delete(&self, vid: Uuid, id: Uuid) {
        let mut cur = self.store.get(vid, id).unwrap().unwrap().record;
        cur.deleted = true;
        cur.wrapped_item_key = None;
        cur.payload = None;
        self.store.put_local(vid, &cur).unwrap();
    }

    fn read(&self, vid: Uuid, id: Uuid) -> Option<Value> {
        let item = self.store.get(vid, id).unwrap()?;
        open_record(self.keyset.account_id, vid, &self.vk(vid), &item.record).unwrap()
    }
}

fn item(id: Uuid, title: &str, password: &str, modified_at: i64) -> Value {
    json!({"id": id, "kind": "login", "title": title, "password": password, "created_at": 1, "modified_at": modified_at})
}

#[tokio::test]
async fn two_devices_roundtrip_and_manifest_chain() {
    let server = TestServer::start().await;
    let (a, vid) = Device::first(&server).await;
    let id = Uuid::new_v4();
    a.save(vid, id, item(id, "Bank", "pw1", 10));

    let r = a.sync().await.unwrap();
    assert_eq!(r.pushed, 1);
    assert_eq!(a.store.get(vid, id).unwrap().unwrap().state, SyncState::Clean);
    let m1 = server.state().vaults[&vid].manifest.clone().unwrap();
    // First sync publishes v1 (new vault) …
    assert_eq!(m1.version, 1);

    let b = Device::second(&server, &a).await;
    let r = b.sync().await.unwrap();
    assert_eq!(r.vaults_synced, 1);
    assert_eq!(b.read(vid, id).unwrap()["password"], "pw1");
    assert_eq!(b.store.vault(vid).unwrap().unwrap().manifest_version, 1);

    b.save(vid, id, item(id, "Bank", "pw2", 20));
    assert_eq!(b.sync().await.unwrap().pushed, 1);
    assert_eq!(server.state().vaults[&vid].manifest.as_ref().unwrap().version, 2);

    a.sync().await.unwrap();
    assert_eq!(a.read(vid, id).unwrap()["password"], "pw2");
    assert_eq!(a.store.vault(vid).unwrap().unwrap().manifest_version, 2);
    // Nothing to do: no new manifest is published.
    a.sync().await.unwrap();
    assert_eq!(server.state().vaults[&vid].manifest.as_ref().unwrap().version, 2);
    // The server only ever saw ciphertext.
    let raw = serde_json::to_string(&server.state().vaults[&vid].items).unwrap();
    assert!(!raw.contains("pw2") && !raw.contains("Bank"));
}

#[tokio::test]
async fn concurrent_edits_merge_field_by_field() {
    let server = TestServer::start().await;
    let (a, vid) = Device::first(&server).await;
    let id = Uuid::new_v4();
    a.save(vid, id, json!({"id": id, "kind": "login", "title": "Mail", "username": "me", "password": "p0", "notes": "n0", "created_at": 1, "modified_at": 10}));
    a.sync().await.unwrap();
    let b = Device::second(&server, &a).await;
    b.sync().await.unwrap();

    // A changes username + password (later), B changes notes + password (earlier).
    a.save(vid, id, json!({"id": id, "kind": "login", "title": "Mail", "username": "me2", "password": "pA", "notes": "n0", "created_at": 1, "modified_at": 30}));
    b.save(vid, id, json!({"id": id, "kind": "login", "title": "Mail", "username": "me", "password": "pB", "notes": "nB", "created_at": 1, "modified_at": 20}));
    a.sync().await.unwrap();

    let r = b.sync().await.unwrap();
    assert_eq!(r.merged, 1, "B hit 412 and merged");
    assert_eq!(r.pushed, 1);
    let merged = b.read(vid, id).unwrap();
    assert_eq!(merged["username"], "me2");
    assert_eq!(merged["notes"], "nB");
    assert_eq!(merged["password"], "pA", "later modified_at wins the same field");
    assert_eq!(merged["modified_at"], 30);
    // B's losing version is kept in history.
    let hist = b.store.history(vid, id).unwrap();
    assert_eq!(hist.len(), 1);
    let lost = open_record(
        b.keyset.account_id,
        vid,
        &b.vk(vid),
        &ItemRecord {
            item_id: id,
            revision: hist[0].revision,
            deleted: false,
            wrapped_item_key: Some(hist[0].wrapped_item_key.clone()),
            payload: Some(hist[0].payload.clone()),
            updated_by: Uuid::nil(),
            updated_at: String::new(),
        },
    )
    .unwrap()
    .unwrap();
    assert_eq!(lost["password"], "pB");

    a.sync().await.unwrap();
    assert_eq!(a.read(vid, id).unwrap(), merged);
    assert!(a.store.pending(vid).unwrap().is_empty() && b.store.pending(vid).unwrap().is_empty());
}

#[tokio::test]
async fn edit_beats_delete_both_ways() {
    let server = TestServer::start().await;
    let (a, vid) = Device::first(&server).await;
    let (x, y) = (Uuid::new_v4(), Uuid::new_v4());
    a.save(vid, x, item(x, "X", "x", 1));
    a.save(vid, y, item(y, "Y", "y", 1));
    a.sync().await.unwrap();
    let b = Device::second(&server, &a).await;
    b.sync().await.unwrap();

    // x: A deletes, B edits (B syncs second → B's edit is pushed over the tombstone).
    // y: A edits, B deletes (B syncs second → A's edit wins, B restores y).
    a.delete(vid, x);
    a.save(vid, y, item(y, "Y", "y-edited", 5));
    a.sync().await.unwrap();
    b.save(vid, x, item(x, "X", "x-edited", 5));
    b.delete(vid, y);
    let r = b.sync().await.unwrap();
    assert_eq!(r.restored.len(), 2);
    assert_eq!(b.read(vid, x).unwrap()["password"], "x-edited");
    assert_eq!(b.read(vid, y).unwrap()["password"], "y-edited");

    a.sync().await.unwrap();
    assert_eq!(a.read(vid, x).unwrap()["password"], "x-edited");
    assert!(!server.state().vaults[&vid].items[&x].deleted);
}

#[tokio::test]
async fn deleting_and_never_synced_items() {
    let server = TestServer::start().await;
    let (a, vid) = Device::first(&server).await;
    let (x, local_only) = (Uuid::new_v4(), Uuid::new_v4());
    a.save(vid, x, item(x, "X", "x", 1));
    a.sync().await.unwrap();
    a.delete(vid, x);
    a.save(vid, local_only, item(local_only, "L", "l", 1));
    a.delete(vid, local_only);
    a.sync().await.unwrap();
    assert!(server.state().vaults[&vid].items[&x].deleted);
    assert!(!server.state().vaults[&vid].items.contains_key(&local_only));
    assert!(a.store.get(vid, local_only).unwrap().is_none());
}

#[tokio::test]
async fn manifest_rollback_is_an_integrity_alarm() {
    let server = TestServer::start().await;
    let (a, vid) = Device::first(&server).await;
    let id = Uuid::new_v4();
    for (i, pw) in ["1", "2", "3"].iter().enumerate() {
        a.save(vid, id, item(id, "T", pw, i as i64));
        a.sync().await.unwrap();
    }
    let history = server.state().vaults[&vid].manifest_history.clone();
    assert_eq!(history.len(), 3);
    server.state().serve_manifest = Some(history[0].clone());

    let err = a.sync().await.unwrap_err();
    assert!(
        matches!(err, SyncError::Integrity { vault_id, ref reason } if vault_id == vid && reason.contains("rollback")),
        "{err}"
    );
    assert!(a.store.vault(vid).unwrap().unwrap().integrity_alarm);

    // Syncing that vault stays stopped even after the server "fixes" itself.
    server.state().serve_manifest = None;
    a.save(vid, id, item(id, "T", "4", 9));
    assert!(matches!(a.sync().await, Err(SyncError::Integrity { .. })));
    assert!(server.state().vaults[&vid].items[&id].revision < 4, "nothing was pushed");
}

#[tokio::test]
async fn forged_or_foreign_manifest_is_rejected() {
    let server = TestServer::start().await;
    let (a, vid) = Device::first(&server).await;
    a.sync().await.unwrap();

    let mut forged = server.state().vaults[&vid].manifest.clone().unwrap();
    forged.signature[0] ^= 1;
    server.state().serve_manifest = Some(forged);
    let err = a.sync().await.unwrap_err();
    assert!(matches!(err, SyncError::Integrity { ref reason, .. } if reason.contains("signature")), "{err}");
}

#[tokio::test]
async fn hidden_item_breaks_merkle_root() {
    let server = TestServer::start().await;
    let (a, vid) = Device::first(&server).await;
    let (x, y) = (Uuid::new_v4(), Uuid::new_v4());
    a.save(vid, x, item(x, "X", "x", 1));
    a.save(vid, y, item(y, "Y", "y", 1));
    a.sync().await.unwrap();

    let b = Device::new(&server, copy_keyset(&a.keyset));
    b.login().await;
    server.state().hide_items = vec![y];
    let err = b.sync().await.unwrap_err();
    assert!(matches!(err, SyncError::Integrity { ref reason, .. } if reason.contains("Merkle")), "{err}");
}

#[tokio::test]
async fn unsigned_remote_vault_is_ignored() {
    let server = TestServer::start().await;
    let (a, vid) = Device::first(&server).await;
    a.sync().await.unwrap();
    // The server injects a vault whose key it sealed to our public key itself.
    let evil = Uuid::new_v4();
    let sealed = kv_crypto::seal::seal_to(
        a.keyset.enc.id,
        &a.keyset.enc.public_bytes(),
        &[0u8; 48],
        &Aad::build(ObjectType::VaultKey, a.keyset.account_id, Some(evil), None),
    )
    .unwrap();
    {
        let mut st = server.state();
        let v = st.vaults.entry(evil).or_default();
        v.sealed_key = Some(sealed.clone());
        v.meta = Some(sealed);
    }
    let r = a.sync().await.unwrap();
    assert_eq!(r.vaults_skipped, vec![evil]);
    assert!(a.store.vault(evil).unwrap().is_none());
    assert!(a.store.vault(vid).unwrap().is_some());
}

#[tokio::test]
async fn pagination_and_writes_from_elsewhere() {
    let server = TestServer::start().await;
    let (a, vid) = Device::first(&server).await;
    a.sync().await.unwrap();
    // 520 items written by "another device" directly into the server.
    let vk = a.vk(vid);
    let ids: Vec<Uuid> = (0..520).map(|_| Uuid::new_v4()).collect();
    for id in &ids {
        let rec = seal_record(a.keyset.account_id, vid, &vk, *id, &item(*id, "bulk", "p", 1), Uuid::nil()).unwrap();
        server.state().write_item(vid, rec);
    }
    let r = a.sync().await.unwrap();
    assert_eq!(r.pulled, 520);
    assert_eq!(a.store.list(vid).unwrap().len(), 520);
    assert_eq!(a.store.cursor(vid).unwrap(), 520);
    let changes_calls = server.state().log.iter().filter(|l| l.contains("/changes")).count();
    assert!(changes_calls >= 2, "paged");
}

#[tokio::test]
async fn repeated_412_is_bounded() {
    let server = TestServer::start().await;
    let (a, vid) = Device::first(&server).await;
    let id = Uuid::new_v4();
    a.save(vid, id, item(id, "T", "p", 1));
    server.state().force_412 = 100;
    let r = a.sync().await.unwrap();
    assert_eq!(r.pushed, 0);
    assert_eq!(a.store.pending(vid).unwrap().len(), 1, "still pending for the next sync");
    server.state().force_412 = 0;
    assert_eq!(a.sync().await.unwrap().pushed, 1);
}

#[tokio::test]
async fn auth_errors() {
    let server = TestServer::start().await;
    let (a, _) = Device::first(&server).await;
    // Wrong OPAQUE password.
    let err = a.api.login(a.keyset.account_id, &[0u8; 32], enroll(&a.signer), None).await.unwrap_err();
    assert!(matches!(err, SyncError::BadCredentials), "{err}");
    // Setup is closed after the first account.
    let err = a
        .api
        .register(
            SETUP_TOKEN,
            Uuid::new_v4(),
            &test_kdf(),
            &AUTH,
            vec![],
            vec![],
            a.keyset.wrap(&SymKey::generate()).unwrap(),
            enroll(&a.signer),
        )
        .await
        .unwrap_err();
    assert!(matches!(err, SyncError::NotFound | SyncError::Server { .. }), "{err}");

    // No session → NotAuthenticated without touching the network.
    let fresh = Device::new(&server, copy_keyset(&a.keyset));
    assert!(matches!(fresh.sync().await, Err(SyncError::NotAuthenticated)));
    // A revoked session → 401 → NotAuthenticated, and the session is dropped.
    server.state().sessions.clear();
    assert!(matches!(a.sync().await, Err(SyncError::NotAuthenticated)));
    assert!(a.api.session().is_none());

    // New device needing approval.
    server.state().require_approval = true;
    let c = Device::new(&server, copy_keyset(&a.keyset));
    assert_eq!(c.login().await, LoginOutcome::PendingApproval);
    server.state().devices.get_mut(&c.signer.device_id()).unwrap().approved = true;
    assert!(matches!(c.login().await, LoginOutcome::Ok { .. }));
    // The account endpoint gives a new device what it needs to unlock.
    let info = c.api.account().await.unwrap();
    assert_eq!(info.sig_pk, a.keyset.sig_public().to_vec());
}

// ---------------------------------------------------------------- TLS pinning

mod tls {
    use std::sync::Arc;

    use kv_sync::{signing::SoftwareSigner, transport::spki_pin_of, ApiClient, ServerConfig, SyncError};
    use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
    use tokio::{
        io::{AsyncReadExt, AsyncWriteExt},
        net::TcpListener,
    };
    use uuid::Uuid;

    struct Pki {
        ca_der: Vec<u8>,
        leaf_der: Vec<u8>,
        leaf_key: Vec<u8>,
    }

    fn pki() -> Pki {
        let ca_key = rcgen::KeyPair::generate().unwrap();
        let mut ca = rcgen::CertificateParams::new(vec![]).unwrap();
        ca.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained);
        ca.distinguished_name.push(rcgen::DnType::CommonName, "Kestrove Test CA");
        let ca_cert = ca.self_signed(&ca_key).unwrap();
        let leaf_key = rcgen::KeyPair::generate().unwrap();
        let leaf = rcgen::CertificateParams::new(vec!["localhost".into()]).unwrap();
        let leaf_cert = leaf.signed_by(&leaf_key, &ca_cert, &ca_key).unwrap();
        Pki { ca_der: ca_cert.der().to_vec(), leaf_der: leaf_cert.der().to_vec(), leaf_key: leaf_key.serialize_der() }
    }

    /// Minimal HTTPS server answering every request with a ServerInfo JSON. Returns the port.
    async fn serve(p: &Pki, tls13_only: bool) -> u16 {
        let provider = Arc::new(rustls::crypto::ring::default_provider());
        let versions: &[&rustls::SupportedProtocolVersion] =
            if tls13_only { &[&rustls::version::TLS13] } else { &[&rustls::version::TLS12] };
        let cfg = rustls::ServerConfig::builder_with_provider(provider)
            .with_protocol_versions(versions)
            .unwrap()
            .with_no_client_auth()
            .with_single_cert(
                // Like a real server, send the chain (here leaf + issuing CA).
                vec![CertificateDer::from(p.leaf_der.clone()), CertificateDer::from(p.ca_der.clone())],
                PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(p.leaf_key.clone())),
            )
            .unwrap();
        let acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(cfg));
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        tokio::spawn(async move {
            while let Ok((sock, _)) = listener.accept().await {
                let acceptor = acceptor.clone();
                tokio::spawn(async move {
                    let Ok(mut tls) = acceptor.accept(sock).await else { return };
                    let mut buf = [0u8; 4096];
                    let _ = tls.read(&mut buf).await;
                    let body =
                        r#"{"version":"t","api":1,"single_tenant":true,"setup_open":false,"min_client":"0.1.0"}"#;
                    let resp = format!(
                        "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                        body.len()
                    );
                    let _ = tls.write_all(resp.as_bytes()).await;
                    let _ = tls.shutdown().await;
                });
            }
        });
        port
    }

    fn client(p: &Pki, port: u16, pins: Vec<String>) -> ApiClient {
        let cfg = ServerConfig {
            base_url: format!("https://localhost:{port}"),
            spki_pins: pins,
            extra_roots_der: vec![p.ca_der.clone()],
        };
        ApiClient::new(&cfg, Arc::new(SoftwareSigner::generate(Uuid::new_v4()))).unwrap()
    }

    #[tokio::test]
    async fn pins() {
        let p = pki();
        let port = serve(&p, true).await;
        let leaf_pin = spki_pin_of(&p.leaf_der).unwrap();
        let ca_pin = spki_pin_of(&p.ca_der).unwrap();
        let wrong = spki_pin_of(&pki().leaf_der).unwrap();

        // Primary pin on the leaf.
        assert!(client(&p, port, vec![leaf_pin.clone()]).server_info().await.is_ok());
        // Backup pin matches (primary is a stale key).
        assert!(client(&p, port, vec![wrong.clone(), leaf_pin]).server_info().await.is_ok());
        // Pinning the issuing CA key (it is in the presented chain).
        assert!(client(&p, port, vec![ca_pin]).server_info().await.is_ok());
        // Mismatch.
        let err = client(&p, port, vec![wrong.clone()]).server_info().await.unwrap_err();
        assert!(matches!(err, SyncError::PinMismatch), "{err:?}");
        // Without our CA, chain validation fails before pins are even considered.
        let cfg = ServerConfig::new(format!("https://localhost:{port}"), vec![wrong]);
        let c = ApiClient::new(&cfg, Arc::new(SoftwareSigner::generate(Uuid::new_v4()))).unwrap();
        assert!(matches!(c.server_info().await, Err(SyncError::Network(_))));
    }

    #[tokio::test]
    async fn tls12_is_refused() {
        let p = pki();
        let port = serve(&p, false).await;
        let pin = spki_pin_of(&p.leaf_der).unwrap();
        assert!(matches!(client(&p, port, vec![pin]).server_info().await, Err(SyncError::Network(_))));
    }
}