kestrove-core/crates/kv-sync/src/client.rs

//! Typed HTTP client for the sync API (docs/03, api/openapi.yaml).

use std::{
    sync::{Arc, Mutex},
    time::{SystemTime, UNIX_EPOCH},
};

use kv_proto::{
    AccountInfo, ApiError, Changes, DeviceEnroll, DeviceInfo, ItemPut, ItemRecord, KdfParams, LoginFinish,
    LoginFinishResponse, LoginStart, LoginStartResponse, LoginStatus, Manifest, PreloginResponse,
    RegistrationStartResponse, ServerInfo, SetupFinish, SetupStart, Vault,
};
use reqwest::{Method, StatusCode};
use serde::{de::DeserializeOwned, Serialize};
use uuid::Uuid;

use crate::{
    opaque, signing,
    transport::{classify, http_client, normalize_base_url},
    wire::*,
    DeviceSigner, ServerConfig, SyncError,
};

pub const PAGE_LIMIT: u32 = 500;

/// Result of an OPAQUE login.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LoginOutcome {
    /// Session established (and stored in the client).
    Ok { session: String },
    /// New device: an enrolled device must approve it (or a TOTP must be supplied), then log in again.
    PendingApproval,
}

pub struct ApiClient {
    base: String,
    http: reqwest::Client,
    signer: Arc<dyn DeviceSigner>,
    session: Mutex<Option<String>>,
}

fn now_unix() -> u64 {
    SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0)
}

impl ApiClient {
    pub fn new(cfg: &ServerConfig, signer: Arc<dyn DeviceSigner>) -> Result<Self, SyncError> {
        Ok(Self {
            base: normalize_base_url(&cfg.base_url)?,
            http: http_client(cfg)?,
            signer,
            session: Mutex::new(None),
        })
    }

    pub fn base_url(&self) -> &str {
        &self.base
    }

    pub fn signer(&self) -> &Arc<dyn DeviceSigner> {
        &self.signer
    }

    pub fn session(&self) -> Option<String> {
        self.session.lock().unwrap_or_else(|p| p.into_inner()).clone()
    }

    pub fn set_session(&self, s: Option<String>) {
        *self.session.lock().unwrap_or_else(|p| p.into_inner()) = s;
    }

    async fn send(
        &self,
        method: Method,
        path: &str,
        body: Option<Vec<u8>>,
        if_match: Option<u64>,
        signed: bool,
    ) -> Result<(StatusCode, Vec<u8>), SyncError> {
        let url = format!("{}{}", self.base, path);
        let body = body.unwrap_or_default();
        let mut req = self.http.request(method.clone(), &url);
        if !body.is_empty() {
            req = req.header("content-type", "application/json");
        }
        if let Some(rev) = if_match {
            req = req.header("if-match", format!("\"{rev}\""));
        }
        if signed {
            let session = self.session().ok_or(SyncError::NotAuthenticated)?;
            let h = signing::sign_request(self.signer.as_ref(), method.as_str(), &url, &body, &session, now_unix())?;
            for (k, v) in h.pairs() {
                req = req.header(k, v);
            }
        }
        let resp = req.body(body).send().await.map_err(classify)?;
        let status = resp.status();
        let bytes = resp.bytes().await.map_err(classify)?.to_vec();
        if status.is_success() {
            return Ok((status, bytes));
        }
        let api: Option<ApiError> = serde_json::from_slice(&bytes).ok();
        let (code, message) = api.map(|e| (e.code, e.message)).unwrap_or_default();
        Err(match status {
            StatusCode::PRECONDITION_FAILED => SyncError::RevisionConflict,
            StatusCode::CONFLICT => SyncError::VersionConflict,
            StatusCode::NOT_FOUND => SyncError::NotFound,
            StatusCode::UNAUTHORIZED if signed => {
                self.set_session(None);
                SyncError::NotAuthenticated
            }
            _ => SyncError::Server { status: status.as_u16(), code, message },
        })
    }

    async fn json<T: DeserializeOwned>(
        &self,
        method: Method,
        path: &str,
        body: Option<&impl Serialize>,
        if_match: Option<u64>,
        signed: bool,
    ) -> Result<T, SyncError> {
        let body = body.map(serde_json::to_vec).transpose().map_err(|e| SyncError::Protocol(e.to_string()))?;
        let (_, bytes) = self.send(method, path, body, if_match, signed).await?;
        serde_json::from_slice(&bytes).map_err(|e| SyncError::Protocol(format!("{path}: {e}")))
    }

    async fn no_content(
        &self,
        method: Method,
        path: &str,
        body: Option<&impl Serialize>,
        signed: bool,
    ) -> Result<(), SyncError> {
        let body = body.map(serde_json::to_vec).transpose().map_err(|e| SyncError::Protocol(e.to_string()))?;
        self.send(method, path, body, None, signed).await.map(|_| ())
    }

    // ---------------------------------------------------------------- unauthenticated

    pub async fn server_info(&self) -> Result<ServerInfo, SyncError> {
        self.json(Method::GET, PATH_SERVER_INFO, None::<&()>, None, false).await
    }

    /// Account id and KDF parameters of the (single) account: what a new device needs to derive
    /// `AUTH_INPUT` before it can build KE1.
    pub async fn prelogin(&self) -> Result<PreloginResponse, SyncError> {
        self.json(Method::GET, PATH_PRELOGIN, None::<&()>, None, false).await
    }

    /// OPAQUE registration plus account creation (first device only, needs the setup token).
    #[allow(clippy::too_many_arguments)]
    pub async fn register(
        &self,
        setup_token: &str,
        account_id: Uuid,
        kdf: &KdfParams,
        auth_input: &[u8],
        enc_pk: Vec<u8>,
        sig_pk: Vec<u8>,
        wrapped_keyset: kv_proto::Envelope,
        device: DeviceEnroll,
    ) -> Result<(), SyncError> {
        let reg = opaque::registration_start(auth_input)?;
        let start =
            SetupStart { setup_token: setup_token.to_string(), account_id, registration_request: reg.request.clone() };
        let resp: RegistrationStartResponse =
            self.json(Method::POST, PATH_SETUP_START, Some(&start), None, false).await?;
        let upload = reg.finish(auth_input, &resp.registration_response)?;
        let req = SetupFinish {
            setup_token: setup_token.to_string(),
            account_id,
            kdf: kdf.clone(),
            opaque_registration_upload: upload,
            enc_pk,
            sig_pk,
            wrapped_keyset,
            device,
        };
        self.no_content(Method::POST, PATH_SETUP_FINISH, Some(&req), false).await
    }

    /// OPAQUE login. On success the session is stored in this client.
    pub async fn login(
        &self,
        account_id: Uuid,
        auth_input: &[u8],
        device: DeviceEnroll,
        totp: Option<String>,
    ) -> Result<LoginOutcome, SyncError> {
        let start = opaque::login_start(auth_input)?;
        let req = LoginStart { account_id, ke1: start.ke1.clone(), device };
        let resp: LoginStartResponse = self.json(Method::POST, PATH_LOGIN_START, Some(&req), None, false).await?;
        let done = start.finish(auth_input, &resp.ke2)?;
        let fin = LoginFinish { login_id: resp.login_id, ke3: done.ke3, totp };
        let resp: LoginFinishResponse = match self.json(Method::POST, PATH_LOGIN_FINISH, Some(&fin), None, false).await
        {
            Err(SyncError::Server { status: 401, .. }) => return Err(SyncError::BadCredentials),
            other => other?,
        };
        match (resp.status, resp.session) {
            (LoginStatus::Ok, Some(s)) => {
                self.set_session(Some(s.clone()));
                Ok(LoginOutcome::Ok { session: s })
            }
            (LoginStatus::Ok, None) => Err(SyncError::Protocol("login ok without a session".into())),
            (LoginStatus::PendingApproval, _) => Ok(LoginOutcome::PendingApproval),
        }
    }

    // ---------------------------------------------------------------- authenticated

    pub async fn logout(&self) -> Result<(), SyncError> {
        let r = self.no_content(Method::POST, PATH_LOGOUT, None::<&()>, true).await;
        self.set_session(None);
        r
    }

    pub async fn account(&self) -> Result<AccountInfo, SyncError> {
        self.json(Method::GET, PATH_ACCOUNT, None::<&()>, None, true).await
    }

    pub async fn devices(&self) -> Result<Vec<DeviceInfo>, SyncError> {
        self.json(Method::GET, PATH_DEVICES, None::<&()>, None, true).await
    }

    /// Approves a pending device (this device must be approved itself).
    pub async fn approve_device(&self, device: Uuid) -> Result<(), SyncError> {
        self.no_content(Method::POST, &format!("{PATH_DEVICES}/{device}/approve"), None::<&()>, true).await
    }

    /// Revokes a device permanently and ends its sessions.
    pub async fn revoke_device(&self, device: Uuid) -> Result<(), SyncError> {
        self.no_content(Method::DELETE, &format!("{PATH_DEVICES}/{device}"), None::<&()>, true).await
    }

    pub async fn vaults(&self) -> Result<Vec<Vault>, SyncError> {
        self.json(Method::GET, PATH_VAULTS, None::<&()>, None, true).await
    }

    pub async fn create_vault(&self, v: &Vault) -> Result<(), SyncError> {
        self.no_content(Method::POST, PATH_VAULTS, Some(v), true).await
    }

    pub async fn changes(&self, vault: Uuid, since: u64) -> Result<Changes, SyncError> {
        let path = format!("{PATH_VAULTS}/{vault}/changes?since={since}&limit={PAGE_LIMIT}");
        self.json(Method::GET, &path, None::<&()>, None, true).await
    }

    pub async fn put_item(&self, vault: Uuid, item: Uuid, base: u64, body: &ItemPut) -> Result<ItemRecord, SyncError> {
        self.json(Method::PUT, &format!("{PATH_VAULTS}/{vault}/items/{item}"), Some(body), Some(base), true).await
    }

    /// Tombstones an item. `body` is the sealed tombstone `{id, seq, deleted: true}` (docs/03).
    pub async fn delete_item(
        &self,
        vault: Uuid,
        item: Uuid,
        base: u64,
        body: &ItemPut,
    ) -> Result<ItemRecord, SyncError> {
        self.json(Method::DELETE, &format!("{PATH_VAULTS}/{vault}/items/{item}"), Some(body), Some(base), true).await
    }

    /// `None` when the vault has no manifest yet.
    pub async fn get_manifest(&self, vault: Uuid) -> Result<Option<Manifest>, SyncError> {
        match self.json(Method::GET, &format!("{PATH_VAULTS}/{vault}/manifest"), None::<&()>, None, true).await {
            Ok(m) => Ok(Some(m)),
            Err(SyncError::NotFound) => Ok(None),
            Err(e) => Err(e),
        }
    }

    pub async fn put_manifest(&self, vault: Uuid, m: &Manifest) -> Result<(), SyncError> {
        self.no_content(Method::PUT, &format!("{PATH_VAULTS}/{vault}/manifest"), Some(m), true).await
    }
}