aven_core/sync/shared_state/
peer_install.rs

1//! Published reads and fresh-peer installation. Enrollment authority is supplied
2//! by the protected host, never recovered from a SQLite receipt.
3use super::*;
4use crate::db::installation::InstallationGuard;
5use crate::sync::{
6    bootstrap_staging::Component,
7    seed_claim::{membership::VerifiedEnrollment, peer},
8};
9use sha2::{Digest, Sha256};
10
11impl Database {
12    /// Copies one published record inside the current-membership transaction.
13    /// Image bytes are read from lifecycle ownership, never staging or restored.
14    pub async fn published_snapshot_read(
15        &self,
16        auth: &peer::Authentication<'_>,
17        descriptor: [u8; 32],
18        component: Option<Component>,
19        index: u64,
20    ) -> Result<Vec<u8>> {
21        let mut conn = self.acquire_writer().await?;
22        let mut tx = db::begin_immediate(&mut conn).await?;
23        let current =
24            crate::sync::seed_claim::membership::persistence::current(self, &mut tx).await?;
25        current.membership.authenticate(auth, false)?;
26        let binding = current.membership.publication().binding();
27        ensure!(
28            binding.descriptor_commitment == descriptor,
29            "error snapshot-publication-mismatch"
30        );
31        let bytes = match component {
32            None => {
33                ensure!(index == 0, "error snapshot-index");
34                current.evidence.descriptor.clone()
35            }
36            Some(Component::Image(object)) => {
37                // SQLite owns the bytes. A prune transaction cannot invalidate
38                // this owned response buffer, and this read creates no pin.
39                sqlx::query_scalar("SELECT c.bytes FROM server_e2ee_image_chunks c JOIN server_e2ee_images i ON i.object=c.object WHERE i.bootstrap=? AND c.object=? AND c.chunk_index=?")
40                    .bind(binding.bootstrap_id.as_slice()).bind(object.as_slice())
41                    .bind(i64::try_from(index)?).fetch_optional(&mut *tx).await?
42                    .context("error snapshot-image-unavailable")?
43            }
44            Some(component) => {
45                sqlx::query_scalar("SELECT bytes FROM server_bootstrap_chunks WHERE bootstrap=? AND component=? AND chunk_index=?")
46                    .bind(binding.bootstrap_id.as_slice()).bind(component.key())
47                    .bind(i64::try_from(index)?).fetch_optional(&mut *tx).await?
48                    .context("error snapshot-artifact-unavailable")?
49            }
50        };
51        ensure!(
52            bytes.len() <= crate::sync::bootstrap_staging::MAX_REQUEST_BYTES,
53            "error snapshot-limit"
54        );
55        tx.commit().await?;
56        Ok(bytes)
57    }
58
59    /// Receipt lookup still requires independently verified protected authority.
60    /// It never repairs a receipt or resets a later cursor or domain edit.
61    pub async fn peer_snapshot_receipt(
62        &self,
63        verified: &VerifiedEnrollment,
64        enrollment: [u8; 32],
65        client: &str,
66        guard: &InstallationGuard,
67    ) -> Result<Option<SharedStateInstallReport>> {
68        ensure!(
69            self.file_identity() == Some(guard.identity()),
70            "error snapshot-installation"
71        );
72        let mut conn = self.acquire_writer().await?;
73        let mut tx = db::begin_immediate(&mut conn).await?;
74        validate_target_identity(&mut tx, enrollment, client).await?;
75        let receipt = receipt(&mut tx, verified, enrollment, client).await?;
76        tx.commit().await?;
77        Ok(receipt)
78    }
79
80    /// Authenticated metadata and mappings commit independently of image bytes.
81    pub async fn install_peer_snapshot(
82        &self,
83        verified: &VerifiedEnrollment,
84        enrollment: [u8; 32],
85        client: &str,
86        guard: &InstallationGuard,
87        package: &bootstrap_format::download::Metadata,
88    ) -> Result<SharedStateInstallReport> {
89        if let Some(report) = self
90            .peer_snapshot_receipt(verified, enrollment, client, guard)
91            .await?
92        {
93            return Ok(report);
94        }
95        let binding = verified.publication().binding();
96        ensure!(
97            Sha256::digest(&package.descriptor).as_slice() == binding.descriptor_commitment,
98            "error snapshot-publication-mismatch"
99        );
100        let bootstrap_format::download::VerifiedMetadata { mut capture, index } =
101            bootstrap_format::download::decrypt(package, verified.key())?;
102        for blob in &mut capture.snapshot.tables.blob_inventory {
103            blob.available = 0;
104            blob.last_verified_at = None;
105        }
106        let mut conn = self.acquire_writer().await?;
107        let mut tx = db::begin_immediate(&mut conn).await?;
108        validate_target_identity(&mut tx, enrollment, client).await?;
109        if let Some(report) = receipt(&mut tx, verified, enrollment, client).await? {
110            tx.commit().await?;
111            return Ok(report);
112        }
113        ensure_empty_domain(&mut tx).await?;
114        let occupied: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_seed_source) OR EXISTS(SELECT 1 FROM local_seed_publication_intent) OR EXISTS(SELECT 1 FROM local_seed_genesis_pin) OR EXISTS(SELECT 1 FROM server_seed_claim) OR EXISTS(SELECT 1 FROM meta WHERE key IN ('sync_server_url','e2ee_association'))").fetch_one(&mut *tx).await?;
115        ensure!(!occupied, "error snapshot-target-not-fresh");
116        let report = install_in_transaction(&mut tx, &capture).await?;
117        crate::sync::crash::Crash::Snapshot.at("metadata");
118        let hashes = capture
119            .snapshot
120            .tables
121            .blob_inventory
122            .iter()
123            .map(|b| b.sha256.clone())
124            .collect::<Vec<_>>();
125        crate::attachments::lifecycle::reconcile_liveness_for_hashes_in_transaction(
126            &mut tx,
127            &hashes,
128            &crate::attachments::lifecycle::SystemClock,
129        )
130        .await?;
131        let generation = db::get_meta(&mut tx, "sync_generation")
132            .await?
133            .unwrap_or_else(|| "0".into())
134            .parse::<i64>()?
135            .checked_add(1)
136            .context("error snapshot-generation")?;
137        let association = association(verified);
138        crate::sync::encrypted_tail::dependencies::initialize(
139            &mut tx,
140            &association,
141            generation,
142            i64::try_from(binding.prefix_count)?,
143            &capture.snapshot.tables.task_dependencies,
144        )
145        .await?;
146        crate::sync::encrypted_tail::graphs::initialize(
147            &mut tx,
148            &capture.snapshot.tables.task_epic_links,
149        )
150        .await?;
151        crate::sync::encrypted_tail::attachments::client::initialize_index(
152            &mut tx,
153            &association,
154            generation,
155            i64::try_from(binding.prefix_count)?,
156            &binding.descriptor_commitment,
157            index,
158            false,
159        )
160        .await?;
161        db::set_meta(&mut tx, "sync_generation", &generation.to_string()).await?;
162        db::set_meta(&mut tx, "sync_cursor", &binding.prefix_count.to_string()).await?;
163        db::set_meta(&mut tx, "e2ee_association", &association).await?;
164        sqlx::query("INSERT INTO local_peer_snapshot_install(singleton,enrollment,checkpoint,descriptor,stream,prefix_count,client_id,association,sync_generation,attachment_count) VALUES(1,?,?,?,?,?,?,?,?,?)")
165            .bind(enrollment.as_slice()).bind(verified.checkpoint().as_slice()).bind(binding.descriptor_commitment.as_slice()).bind(binding.stream_id.as_slice()).bind(i64::try_from(binding.prefix_count)?).bind(client).bind(association).bind(generation).bind(i64::try_from(report.attachment_count)?).execute(&mut *tx).await?;
166        crate::sync::crash::Crash::Snapshot.at("before-commit");
167        tx.commit().await?;
168        crate::sync::crash::Crash::Snapshot.at("after-commit");
169        Ok(report)
170    }
171}
172
173fn association(v: &VerifiedEnrollment) -> String {
174    let b = v.publication().binding();
175    format!(
176        "{}:{}:{}",
177        hex::encode(b.vault_id),
178        hex::encode(b.stream_id),
179        hex::encode(b.bootstrap_id)
180    )
181}
182async fn validate_target_identity(
183    conn: &mut sqlx::SqliteConnection,
184    enrollment: [u8; 32],
185    client: &str,
186) -> Result<()> {
187    let matches: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_peer_enrollment WHERE identity=? AND client_id=? AND role='peer' AND client_id=(SELECT value FROM meta WHERE key='client_id'))").bind(enrollment.as_slice()).bind(client).fetch_one(conn).await?;
188    ensure!(matches, "error snapshot-enrollment-mismatch");
189    Ok(())
190}
191async fn receipt(
192    conn: &mut sqlx::SqliteConnection,
193    verified: &VerifiedEnrollment,
194    enrollment: [u8; 32],
195    client: &str,
196) -> Result<Option<SharedStateInstallReport>> {
197    let exists: bool =
198        sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_peer_snapshot_install)")
199            .fetch_one(&mut *conn)
200            .await?;
201    if !exists {
202        return Ok(None);
203    }
204    let b = verified.publication().binding();
205    let count: Option<i64> = sqlx::query_scalar("SELECT attachment_count FROM local_peer_snapshot_install WHERE singleton=1 AND enrollment=? AND checkpoint=? AND descriptor=? AND stream=? AND prefix_count=? AND client_id=? AND association=? AND association=(SELECT value FROM meta WHERE key='e2ee_association') AND sync_generation=CAST((SELECT value FROM meta WHERE key='sync_generation') AS INTEGER) AND CAST((SELECT value FROM meta WHERE key='sync_cursor') AS INTEGER)>=prefix_count")
206        .bind(enrollment.as_slice()).bind(verified.checkpoint().as_slice()).bind(b.descriptor_commitment.as_slice()).bind(b.stream_id.as_slice()).bind(i64::try_from(b.prefix_count)?).bind(client).bind(association(verified)).fetch_optional(&mut *conn).await?;
207    crate::sync::encrypted_tail::dependencies::validate(
208        conn,
209        &association(verified),
210        i64::try_from(b.prefix_count)?,
211    )
212    .await?;
213    crate::sync::encrypted_tail::attachments::client::validate(
214        conn,
215        &association(verified),
216        i64::try_from(b.prefix_count)?,
217        &b.descriptor_commitment,
218    )
219    .await?;
220    Ok(Some(SharedStateInstallReport {
221        prefix_count: b.prefix_count,
222        attachment_count: u64::try_from(count.context("error snapshot-receipt-mismatch")?)?,
223    }))
224}
aven_core/sync/shared_state/peer_install.rs

aven_core/sync/shared_state/
peer_install.rs
