coven_storage/
remote.rs

1//! `CloudSyncObjectStorage` implementation backed by any `CloudHome`.
2//!
3//! Handles the cloud home path layout (where keys, heads, images, etc. live)
4//! and how objects are protected at rest. The underlying `CloudHome` only deals
5//! in raw bytes and flat keys; this layer applies the [`CloudCipher`] — sealing
6//! every object under the store key for an encrypted home, or storing it
7//! verbatim for a plaintext one — and drives the object-key suffix off the same
8//! choice (`.enc` for encrypted data-plane objects, no suffix for signed
9//! control-plane and recipient-sealed objects).
10
11use async_trait::async_trait;
12use std::path::Path;
13use std::sync::{Arc, RwLock};
14
15use super::provider_probe::ProviderProbeStorage;
16use super::CloudSyncObjectStorage;
17use crate::cloud::{BlobBody, CloudFileReadError, CloudHomeError, ExactCloudHome};
18use coven_keys::encryption::{
19    EncryptionError, EncryptionService, KeyTag, NoncePolicy, SealedBlobHeader,
20    SEALED_BLOB_HEADER_LEN,
21};
22use coven_keys::keys::UserKeypair;
23use coven_protocol::objects::ObjectSlot;
24#[cfg(test)]
25use coven_protocol::objects::ProtocolObjectDomain;
26use coven_protocol::objects::{
27    ExactObjectRef, PreparedExactObject, ProtocolObjectContext, ProtocolObjectProtection,
28    ResolvedProviderBinding, RotationGate, RotationPending, StorageError,
29};
30use coven_protocol::store_commit::ObjectHash;
31
32mod blob_io;
33mod cipher;
34mod rotation;
35mod storage_impl;
36
37#[cfg(any(test, feature = "test-utils"))]
38pub use blob_io::open_sealed_blob;
39pub use blob_io::BlobChunking;
40pub use blob_io::{BlobPathScheme, BlobRangeReader};
41#[cfg(any(test, feature = "test-utils"))]
42pub use cipher::CloudKeyringFacts;
43pub use cipher::{
44    cloud_aad_context, AdoptedCloudKeyRotation, CloudKeyringMerge, CloudSyncCipherStateAccess,
45};
46pub use rotation::{CloudSyncRotationStateAccess, PendingRotation, RotationStateError};
47
48/// How a cloud home protects its objects at rest. An `Encrypted` home seals
49/// every object under the store key (the default); a `Plaintext` home stores
50/// objects in the clear so the bucket is browsable, and drops the `.enc` suffix.
51#[derive(Clone)]
52pub enum CloudCipher {
53    Encrypted(EncryptionService),
54    Plaintext,
55}
56
57/// `CloudSyncObjectStorage` that delegates raw I/O to a `CloudHome` and handles the path
58/// layout and the at-rest protection (its [`CloudCipher`]).
59pub struct CloudSyncConnection {
60    /// `Arc` because ranged readers retain this provider across awaits.
61    home: Arc<dyn ExactCloudHome>,
62    provider_probes: ProviderProbeStorage,
63    cipher: Arc<RwLock<CloudCipher>>,
64    /// Whether a committed rotation is outstanding — see [`PendingRotation`].
65    /// Shared the same way `cipher` is, so a member removal or a refresh cycle
66    /// that discovers a rotation this device can't adopt blocks every seal path,
67    /// not just the one that discovered it.
68    pending_rotation: Arc<PendingRotation>,
69    /// How blob objects are keyed. Unlike the cipher, the scheme does not rotate
70    /// over a home's life, so it is a plain field with no lock.
71    blob_paths: BlobPathScheme,
72    /// How this installation chunks blobs and how wide its range requests are.
73    blob_chunking: BlobChunking,
74    store_id: String,
75    /// The device's signing identity. The control objects this storage writes
76    /// (its head, the min_schema floor) are signed with it so a reader can
77    /// attribute and verify them; the at-rest cipher proves confidentiality, not
78    /// authorship.
79    keypair: UserKeypair,
80}
81
82fn map_cloud_file_read_error(error: CloudFileReadError) -> StorageError {
83    match error {
84        CloudFileReadError::Source(error) => StorageError::from(error),
85        CloudFileReadError::SourceCleanup { source, cleanup } => StorageError::CleanupFailed {
86            operation: Box::new(StorageError::from(source)),
87            cleanup: Box::new(StorageError::LocalFilesystem(cleanup)),
88        },
89        CloudFileReadError::Local(error) => StorageError::LocalFilesystem(error),
90    }
91}
92
93impl CloudSyncConnection {
94    pub fn new(
95        home: Arc<dyn ExactCloudHome>,
96        cipher: CloudCipher,
97        blob_paths: BlobPathScheme,
98        store_id: impl Into<String>,
99        keypair: UserKeypair,
100    ) -> Self {
101        let provider_probes = ProviderProbeStorage::new(home.clone());
102        CloudSyncConnection {
103            home,
104            provider_probes,
105            cipher: Arc::new(RwLock::new(cipher)),
106            pending_rotation: Arc::new(PendingRotation::none()),
107            blob_paths,
108            blob_chunking: BlobChunking::DEFAULT,
109            store_id: store_id.into(),
110            keypair,
111        }
112    }
113
114    /// The running total of provider operations issued through this
115    /// connection's home — by this connection and by any other over the same
116    /// home, which on a device join is how the plaintext bootstrap reads and
117    /// the encrypted reads after them land in one total.
118    pub fn provider_requests(
119        &self,
120    ) -> Option<Arc<dyn coven_foundation::stage_timing::ProviderRequests>> {
121        self.home.provider_requests()
122    }
123
124    /// Seal and read blobs with `chunking` instead of [`BlobChunking::DEFAULT`].
125    /// The chunk size applies to blobs this storage seals from now on; already
126    /// stored blobs keep the size their own headers record, so installations
127    /// with different settings read each other's blobs unchanged.
128    pub fn with_blob_chunking(mut self, chunking: BlobChunking) -> Self {
129        self.blob_chunking = chunking;
130        self
131    }
132
133    pub fn blob_path_scheme(&self) -> BlobPathScheme {
134        self.blob_paths
135    }
136
137    pub fn store_id(&self) -> &str {
138        &self.store_id
139    }
140
141    pub async fn probe(&self) -> Result<(), CloudHomeError> {
142        self.home.probe().await
143    }
144
145    fn validate_blob_locator_home(
146        &self,
147        locator: &coven_protocol::blob::locator::BlobLocator,
148    ) -> Result<(), StorageError> {
149        let valid = matches!(
150            (locator, self.blob_paths, self.cipher.is_plaintext()),
151            (
152                coven_protocol::blob::locator::BlobLocator::Opaque { .. },
153                BlobPathScheme::Hashed,
154                false
155            ) | (
156                coven_protocol::blob::locator::BlobLocator::Browsable { .. },
157                BlobPathScheme::Plain,
158                true
159            )
160        );
161        if !valid {
162            return Err(StorageError::InvalidContent(
163                "blob locator protection does not match the cloud home's fixed storage mode"
164                    .to_string(),
165            ));
166        }
167        Ok(())
168    }
169
170    async fn validate_blob_append_authority(
171        &self,
172        locator: &coven_protocol::blob::locator::BlobLocator,
173        authority: &coven_protocol::objects::BlobWriteAuthority<'_>,
174    ) -> Result<(), StorageError> {
175        authority
176            .reference
177            .verify_registration(authority.registration)?;
178        if locator.uploader() != authority.reference {
179            return Err(StorageError::InvalidContent(format!(
180                "blob locator uploader {:?} differs from its exact write authority",
181                locator.uploader()
182            )));
183        }
184        if authority.registration.author_pubkey != hex::encode(self.keypair.public_key()) {
185            return Err(StorageError::InvalidContent(
186                "blob write authority is not this device's identity key".to_string(),
187            ));
188        }
189        let live = self
190            .home
191            .provider_binding()
192            .await
193            .map_err(StorageError::from)?;
194        if live.device != authority.registration.provider {
195            return Err(StorageError::InvalidContent(
196                "blob write authority differs from the authenticated provider principal"
197                    .to_string(),
198            ));
199        }
200        Ok(())
201    }
202
203    pub fn uses_identity(&self, identity: &UserKeypair) -> bool {
204        self.keypair.public_key() == identity.public_key()
205    }
206
207    #[cfg(any(test, feature = "test-utils"))]
208    pub fn connection_for_test_identity(&self, identity: UserKeypair) -> Self {
209        Self::new(
210            self.home.clone(),
211            self.cipher.read().unwrap().clone(),
212            self.blob_paths,
213            self.store_id.clone(),
214            identity,
215        )
216        .with_blob_chunking(self.blob_chunking)
217    }
218
219    #[cfg(any(test, feature = "test-utils"))]
220    pub fn connection_for_test_identity_and_home(
221        &self,
222        identity: UserKeypair,
223        home: Arc<dyn ExactCloudHome>,
224    ) -> Self {
225        Self::new(
226            home,
227            self.cipher.read().unwrap().clone(),
228            self.blob_paths,
229            self.store_id.clone(),
230            identity,
231        )
232        .with_blob_chunking(self.blob_chunking)
233    }
234
235    pub fn is_plaintext(&self) -> bool {
236        self.cipher.read().unwrap().is_plaintext()
237    }
238
239    fn cipher_suffix(&self) -> &'static str {
240        self.cipher.read().unwrap().suffix()
241    }
242
243    fn open_stored_data(
244        &self,
245        stored: Vec<u8>,
246        aad_context: &[u8],
247    ) -> Result<Vec<u8>, EncryptionError> {
248        self.cipher.read().unwrap().open(stored, aad_context)
249    }
250
251    fn seal_stored_data(
252        &self,
253        plaintext: Vec<u8>,
254        aad_context: &[u8],
255    ) -> Result<Vec<u8>, StorageError> {
256        let cipher = self.cipher.read().unwrap();
257        self.pending_rotation.check(cipher.current_generation())?;
258        Ok(cipher.seal(plaintext, aad_context))
259    }
260
261    fn seal_protocol_data(
262        &self,
263        context: &ProtocolObjectContext,
264        plaintext: Vec<u8>,
265        aad_context: &[u8],
266    ) -> Result<Vec<u8>, StorageError> {
267        match context.protection() {
268            ProtocolObjectProtection::StoreEncrypted => {
269                self.seal_stored_data(plaintext, aad_context)
270            }
271            ProtocolObjectProtection::SignedPlaintext
272            | ProtocolObjectProtection::RecipientSealed => {
273                Ok(CloudCipher::Plaintext.seal(plaintext, aad_context))
274            }
275            ProtocolObjectProtection::Circle(encryption) => {
276                Ok(CloudCipher::Encrypted(encryption.clone()).seal(plaintext, aad_context))
277            }
278        }
279    }
280
281    async fn verify_and_open_protocol_data(
282        &self,
283        operation: &'static str,
284        context: &ProtocolObjectContext,
285        object: ExactObjectRef,
286        stored: Vec<u8>,
287        aad_context: Vec<u8>,
288    ) -> Result<Vec<u8>, StorageError> {
289        let cipher = match context.protection() {
290            ProtocolObjectProtection::StoreEncrypted => self.cipher.read().unwrap().clone(),
291            ProtocolObjectProtection::SignedPlaintext => CloudCipher::Plaintext,
292            ProtocolObjectProtection::Circle(encryption) => {
293                CloudCipher::Encrypted(encryption.clone())
294            }
295            ProtocolObjectProtection::RecipientSealed => CloudCipher::Plaintext,
296        };
297        run_storage_cpu(
298            operation,
299            Box::new(move || {
300                object.verify(&stored)?;
301                cipher
302                    .open(stored, &aad_context)
303                    .map_err(|source| StorageError::Decryption {
304                        context: format!("protocol object {}", object.slot().logical_key()),
305                        source,
306                    })
307            }),
308        )
309        .await
310    }
311
312    async fn identify_and_open_protocol_data(
313        &self,
314        context: &ProtocolObjectContext,
315        slot: ObjectSlot,
316        stored: Vec<u8>,
317        aad_context: Vec<u8>,
318    ) -> Result<(Vec<u8>, PreparedExactObject), StorageError> {
319        let cipher = match context.protection() {
320            ProtocolObjectProtection::StoreEncrypted => self.cipher.read().unwrap().clone(),
321            ProtocolObjectProtection::SignedPlaintext => CloudCipher::Plaintext,
322            ProtocolObjectProtection::Circle(encryption) => {
323                CloudCipher::Encrypted(encryption.clone())
324            }
325            ProtocolObjectProtection::RecipientSealed => CloudCipher::Plaintext,
326        };
327        run_storage_cpu(
328            "identify and open protocol slot",
329            Box::new(move || {
330                let object = ExactObjectRef::new(
331                    slot.clone(),
332                    stored.len() as u64,
333                    ObjectHash::digest(&stored),
334                );
335                let prepared = PreparedExactObject::new(object, stored.clone())?;
336                let opened = cipher.open(stored, &aad_context).map_err(|source| {
337                    StorageError::Decryption {
338                        context: format!("protocol object {}", slot.logical_key()),
339                        source,
340                    }
341                })?;
342                Ok((opened, prepared))
343            }),
344        )
345        .await
346    }
347
348    /// The cloud object key for a blob under the home's [`BlobPathScheme`].
349    ///
350    /// **A cloud object is never rewritten with different bytes, so no two blobs ever
351    /// share a key.** `Hashed` gets that from the key itself; `Plain` gets it from the
352    /// blob's declared [`BlobReplacement`](coven_protocol::blob::BlobReplacement), which coven
353    /// enforces where a blob is derived from its row, in the database layer's blob
354    /// declarations — a replaceable blob's readable path must name it, and a write-once
355    /// blob's row can never be repointed. Either way, an object's *presence* at a blob's
356    /// key is proof of
357    /// its *content*, which is what lets the push skip an upload without asking a sealed
358    /// object what it holds.
359    ///
360    /// `Hashed` ignores `cloud_path` and shards by the id under the uploading
361    /// device: `{namespace}/{uploader}/{ab}/{cd}/{id}` — the id is right there, and the
362    /// `{uploader}` segment aligns the keyspace to the storage-access rule (a member
363    /// writes only under its own public key), so `uploader` is required and a missing one
364    /// is an error.
365    ///
366    /// `Plain` uses the consumer's `cloud_path` verbatim: `{namespace}/{cloud_path}`,
367    /// keeping the bucket browsable. Plain blob naming carries no uploader segment
368    /// and ignores `uploader`; the store still has membership authorization. A
369    /// `Plain` home with no `cloud_path` is an error — coven never silently falls
370    /// back to the hashed layout, which would scatter readable-path blobs under
371    /// unfindable shard keys.
372    pub fn blob_key(
373        scheme: BlobPathScheme,
374        namespace: &str,
375        uploader: Option<&str>,
376        id: &str,
377        cloud_path: Option<&str>,
378    ) -> Result<String, StorageError> {
379        match scheme {
380            BlobPathScheme::Hashed => {
381                let uploader = uploader.ok_or_else(|| {
382                    StorageError::Parse(format!(
383                        "an opaque-home blob requires an uploader for {namespace}/{id}"
384                    ))
385                })?;
386                Ok(coven_foundation::store_dir::StoreDir::uploader_hashed_key(
387                    namespace, uploader, id,
388                )?)
389            }
390            BlobPathScheme::Plain => {
391                let path = cloud_path.ok_or_else(|| {
392                    StorageError::Parse(format!(
393                        "unobfuscated blob-path home requires a cloud_path for blob {namespace}/{id}"
394                    ))
395                })?;
396                coven_foundation::store_dir::validate_path_token(namespace)?;
397                coven_foundation::store_dir::validate_cloud_path(path)?;
398                Ok(format!("{namespace}/{path}"))
399            }
400        }
401    }
402
403    #[cfg(test)]
404    async fn blob_write_registration(
405        &self,
406        label: &str,
407    ) -> coven_protocol::store_commit::ReferencedStoreDeviceRegistration {
408        use coven_protocol::store_commit::{
409            DeviceStreamAnchor, StoreCreationId, StoreDeviceRegistration,
410            StoreDeviceRegistrationOrigin, StoreDeviceRegistrationRef, StoreRootRef,
411        };
412
413        let root_bytes = format!("{label} Store root").into_bytes();
414        let root = StoreRootRef {
415            store_root_id: ObjectHash::digest(format!("{label} root id").as_bytes()),
416            store_root_hash: ObjectHash::digest(&root_bytes),
417            object: ExactObjectRef::new(
418                ObjectSlot::logical(format!("store-v1/store-protocol-root/{label}.json")).unwrap(),
419                root_bytes.len() as u64,
420                ObjectHash::digest(&root_bytes),
421            ),
422        };
423        let anchor_slot = |stream: &str| {
424            ObjectSlot::logical(format!(
425                "store-v1/test-device-streams/{label}/{stream}.json"
426            ))
427            .unwrap()
428        };
429        let provider = CloudSyncObjectStorage::provider_binding(self)
430            .await
431            .unwrap()
432            .device;
433        let registration = StoreDeviceRegistration::signed(
434            root,
435            StoreDeviceRegistrationOrigin::Founder {
436                creation_id: StoreCreationId::from_nonce(label),
437            },
438            provider,
439            DeviceStreamAnchor::StoreAnnouncements {
440                first_slot: anchor_slot("announcements"),
441            },
442            DeviceStreamAnchor::StoreAcknowledgements {
443                first_slot: anchor_slot("acknowledgements"),
444            },
445            DeviceStreamAnchor::StoreSnapshots {
446                first_slot: anchor_slot("snapshots"),
447            },
448            &self.keypair,
449        )
450        .unwrap();
451        let bytes = registration.to_bytes();
452        let reference = StoreDeviceRegistrationRef::from_registration(
453            &registration,
454            ExactObjectRef::new(
455                ObjectSlot::logical(format!(
456                    "store-v1/devices/{}/registration.json",
457                    registration.device_id
458                ))
459                .unwrap(),
460                bytes.len() as u64,
461                ObjectHash::digest(&bytes),
462            ),
463        );
464        coven_protocol::store_commit::ReferencedStoreDeviceRegistration::verified(
465            reference,
466            registration,
467        )
468        .expect("construct test blob write registration")
469    }
470
471    #[cfg(any(test, feature = "test-utils"))]
472    pub fn keyring_facts_for_test(&self) -> Option<CloudKeyringFacts> {
473        match &*self.cipher.read().unwrap() {
474            CloudCipher::Encrypted(encryption) => {
475                Some(CloudKeyringFacts::from_encryption(encryption))
476            }
477            CloudCipher::Plaintext => None,
478        }
479    }
480
481    #[cfg(any(test, feature = "test-utils"))]
482    pub fn adopt_key_rotation_for_test(
483        &self,
484        encryption: &EncryptionService,
485        custody: &dyn coven_keys::keys::MasterKeyCustody,
486    ) -> Result<String, coven_keys::keys::KeyError> {
487        CloudSyncCipherStateAccess::adopt_key_rotation(self, encryption, custody)
488            .map(|adopted| adopted.fingerprint().to_string())
489    }
490
491    #[cfg(any(test, feature = "test-utils"))]
492    pub fn mark_rotation_committed_for_test(
493        &self,
494        generation: u64,
495    ) -> Result<(), RotationStateError> {
496        self.pending_rotation.mark_committed(generation)
497    }
498
499    #[cfg(any(test, feature = "test-utils"))]
500    pub fn pending_rotation_generation_for_test(&self) -> Option<u64> {
501        self.pending_rotation.pending_generation()
502    }
503
504    #[cfg(any(test, feature = "test-utils"))]
505    pub fn clear_rotation_gate_for_test(&self) {
506        self.pending_rotation.install_durable_gate(None);
507    }
508}
509
510impl CloudSyncCipherStateAccess for CloudSyncConnection {
511    fn is_plaintext(&self) -> bool {
512        self.cipher.is_plaintext()
513    }
514
515    fn suffix(&self) -> &'static str {
516        self.cipher.suffix()
517    }
518
519    fn current_generation(&self) -> Option<u64> {
520        self.cipher.current_generation()
521    }
522
523    fn current_fingerprint(&self) -> Option<String> {
524        self.cipher.current_fingerprint()
525    }
526
527    fn open(&self, stored: Vec<u8>, aad_context: &[u8]) -> Result<Vec<u8>, EncryptionError> {
528        self.cipher.open(stored, aad_context)
529    }
530
531    fn seal(&self, plaintext: Vec<u8>, aad_context: &[u8]) -> Vec<u8> {
532        self.cipher.seal(plaintext, aad_context)
533    }
534
535    #[cfg(any(test, feature = "test-utils"))]
536    fn open_sealed_blob_for_test(
537        &self,
538        stored: &[u8],
539        aad_context: &[u8],
540    ) -> Result<
541        (coven_keys::encryption::KeyFingerprint, Vec<u8>),
542        coven_keys::encryption::EncryptionError,
543    > {
544        self.cipher.open_sealed_blob_for_test(stored, aad_context)
545    }
546
547    fn merged_keyring(
548        &self,
549        new_encryption: &EncryptionService,
550    ) -> Result<CloudKeyringMerge, EncryptionError> {
551        self.cipher.merged_keyring(new_encryption)
552    }
553
554    fn merge_key_rotation(
555        &self,
556        new_encryption: &EncryptionService,
557        custody: &dyn coven_keys::keys::MasterKeyCustody,
558    ) -> Result<Option<String>, coven_keys::keys::KeyError> {
559        self.cipher.merge_key_rotation(new_encryption, custody)
560    }
561}
562
563impl CloudSyncRotationStateAccess for CloudSyncConnection {
564    fn mark_candidate(
565        &self,
566        generation: u64,
567        mutation: ObjectHash,
568    ) -> Result<(), RotationStateError> {
569        self.pending_rotation.mark_candidate(generation, mutation)
570    }
571
572    fn mark_committed_mutation(
573        &self,
574        generation: u64,
575        mutation: ObjectHash,
576    ) -> Result<(), RotationStateError> {
577        self.pending_rotation
578            .mark_committed_mutation(generation, mutation)
579    }
580
581    fn remove_candidate(
582        &self,
583        generation: u64,
584        mutation: ObjectHash,
585    ) -> Result<(), RotationStateError> {
586        self.pending_rotation.remove_candidate(generation, mutation)
587    }
588
589    fn replace_candidate_mutation(
590        &self,
591        generation: u64,
592        previous: ObjectHash,
593        replacement: ObjectHash,
594    ) -> Result<(), RotationStateError> {
595        self.pending_rotation
596            .replace_candidate_mutation(generation, previous, replacement)
597    }
598
599    fn gate(&self) -> Option<RotationGate> {
600        self.pending_rotation.gate()
601    }
602
603    fn install_durable_gate(&self, gate: Option<RotationGate>) {
604        self.pending_rotation.install_durable_gate(gate);
605    }
606
607    fn check(&self, live_generation: Option<u64>) -> Result<(), RotationPending> {
608        self.pending_rotation.check(live_generation)
609    }
610}
611
612async fn run_storage_cpu<T>(
613    operation: &'static str,
614    work: Box<dyn FnOnce() -> Result<T, StorageError> + Send>,
615) -> Result<T, StorageError>
616where
617    T: Send + 'static,
618{
619    coven_foundation::blocking::run(work)
620        .await
621        .map_err(|source| StorageError::Blocking { operation, source })?
622}
623
624async fn read_source_exact(
625    source: &mut crate::local_file::PlaintextReader,
626    len: usize,
627    locator_hash: ObjectHash,
628) -> Result<Vec<u8>, StorageError> {
629    let mut bytes = Vec::with_capacity(len);
630    while bytes.len() < len {
631        let chunk = source.next_chunk(len - bytes.len()).await?;
632        if chunk.is_empty() {
633            return Err(StorageError::InvalidContent(format!(
634                "blob {locator_hash} stored body ended after {} of {len} required bytes",
635                bytes.len()
636            )));
637        }
638        bytes.extend_from_slice(&chunk);
639    }
640    Ok(bytes)
641}
642
643#[cfg(test)]
644mod tests;
coven_storage/remote.rs

coven_storage/
remote.rs
