aven_core/sync/encrypted_tail/attachments/
client.rs

1use super::super::{Accepted, Authority, Downloads, codec, domain::Projection, hash, valid};
2use super::codec::{Descriptor, DescriptorAuthority};
3use crate::{
4    db::{self, Database, begin_immediate},
5    sync::{bootstrap_format, wire::ChangeWire},
6};
7use anyhow::{Context as _, Result, ensure};
8use sqlx::SqliteConnection;
9use std::path::Path;
10#[cfg(any(test, feature = "test-support"))]
11use std::sync::atomic::{AtomicU64, Ordering::Relaxed};
12#[cfg(any(test, feature = "test-support"))]
13use std::time::Instant;
14
15#[cfg(any(test, feature = "test-support"))]
16static INTEGRITY_SCAN_CALLS: AtomicU64 = AtomicU64::new(0);
17#[cfg(any(test, feature = "test-support"))]
18static INTEGRITY_SCAN_NANOS: AtomicU64 = AtomicU64::new(0);
19
20#[cfg(any(test, feature = "test-support"))]
21pub(crate) fn integrity_scan_metrics() -> (u64, u64) {
22    (
23        INTEGRITY_SCAN_CALLS.load(Relaxed),
24        INTEGRITY_SCAN_NANOS.load(Relaxed),
25    )
26}
27
28#[cfg(any(test, feature = "test-support"))]
29fn record_integrity_scan(start: Instant) {
30    INTEGRITY_SCAN_CALLS.fetch_add(1, Relaxed);
31    INTEGRITY_SCAN_NANOS.fetch_add(
32        u64::try_from(start.elapsed().as_nanos()).unwrap_or(u64::MAX),
33        Relaxed,
34    );
35}
36
37const DOWNLOAD_CURSOR: &str = "e2ee_image_download_after";
38const DOWNLOAD_CANDIDATES: &str = "
39    FROM local_e2ee_image_references r
40    JOIN local_e2ee_image_objects o ON o.object=r.object
41    JOIN task_attachments a ON a.workspace_id=r.workspace AND a.attachment_id=r.reference
42    JOIN tasks t ON t.workspace_id=a.workspace_id AND t.id=a.task_id
43    WHERE a.deleted=0 AND t.deleted=0
44      AND (o.verified=0 OR NOT EXISTS(
45          SELECT 1 FROM blob_inventory b WHERE b.sha256=o.sha256 AND b.available=1))";
46const DOWNLOAD_ACTIVE_REFERENCE: &str = "
47    FROM local_e2ee_image_references r
48    JOIN task_attachments a ON a.workspace_id=r.workspace AND a.attachment_id=r.reference
49    JOIN tasks t ON t.workspace_id=a.workspace_id AND t.id=a.task_id
50    WHERE r.object=o.object AND a.deleted=0 AND t.deleted=0";
51
52async fn select_download_candidate(
53    conn: &mut SqliteConnection,
54    after: Option<&[u8]>,
55) -> Result<Option<(String, Vec<u8>, Vec<u8>, String)>> {
56    let lower_bound = if after.is_some() {
57        "AND o.object>?"
58    } else {
59        ""
60    };
61    let mut query = sqlx::query_as(sqlx::AssertSqlSafe(format!(
62        "SELECT (SELECT r.workspace {DOWNLOAD_ACTIVE_REFERENCE} LIMIT 1),
63                o.object,o.descriptor,o.sha256
64         FROM local_e2ee_image_objects o
65         WHERE (o.verified=0 OR NOT EXISTS(
66             SELECT 1 FROM blob_inventory b WHERE b.sha256=o.sha256 AND b.available=1))
67           AND EXISTS(SELECT 1 {DOWNLOAD_ACTIVE_REFERENCE})
68           {lower_bound}
69         ORDER BY o.object LIMIT 1"
70    )));
71    if let Some(after) = after {
72        query = query.bind(after);
73    }
74    query.fetch_optional(conn).await.map_err(Into::into)
75}
76
77#[cfg(any(test, feature = "test-support"))]
78pub(crate) async fn initialize(
79    conn: &mut SqliteConnection,
80    association: &str,
81    generation: i64,
82    prefix: i64,
83    package: &bootstrap_format::Package,
84    key: &crate::sync::LocalSharedStatePackageKey,
85) -> Result<()> {
86    let index = bootstrap_format::attachment_index(package, key)?;
87    initialize_index(
88        conn,
89        association,
90        generation,
91        prefix,
92        &hash(&package.descriptor),
93        index,
94        true,
95    )
96    .await
97}
98
99pub(crate) async fn initialize_index(
100    conn: &mut SqliteConnection,
101    association: &str,
102    generation: i64,
103    prefix: i64,
104    descriptor: &[u8; 32],
105    index: bootstrap_format::AttachmentIndex,
106    verified: bool,
107) -> Result<()> {
108    db::set_meta(conn, DOWNLOAD_CURSOR, "").await?;
109    db::set_meta(
110        conn,
111        super::super::client::INITIAL_IMAGE_WATERMARK,
112        if verified { "ready" } else { "pending" },
113    )
114    .await?;
115    sqlx::query("INSERT INTO local_e2ee_image_initialization(singleton,association,sync_generation,prefix_count,descriptor) VALUES(1,?,?,?,?)").bind(association).bind(generation).bind(prefix).bind(descriptor.as_slice()).execute(&mut *conn).await?;
116    for (image, sha) in index.objects {
117        let d = Descriptor {
118            vault: index.context.vault_id,
119            stream: index.stream,
120            generation: index.context.generation_id,
121            object: image.id,
122            artifact: image.artifact,
123        };
124        sqlx::query("INSERT INTO local_e2ee_image_objects(object,descriptor,sha256,origin,verified) VALUES(?,?,?,'bootstrap',?)").bind(d.object.as_slice()).bind(d.encode()?).bind(sha).bind(verified).execute(&mut *conn).await?;
125    }
126    for r in index.references {
127        sqlx::query("INSERT INTO local_e2ee_image_references(workspace,reference,parent,object,origin,deleted) VALUES(?,?,?,?,'bootstrap',?)").bind(r.workspace).bind(r.reference).bind(r.task).bind(r.object.map(|id|id.to_vec())).bind(r.deleted).execute(&mut *conn).await?;
128    }
129    Ok(())
130}
131pub(crate) async fn validate(
132    conn: &mut SqliteConnection,
133    association: &str,
134    prefix: i64,
135    descriptor: &[u8; 32],
136) -> Result<()> {
137    let ok:bool=sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_e2ee_image_initialization WHERE singleton=1 AND association=? AND prefix_count=? AND descriptor=? AND association=(SELECT value FROM meta WHERE key='e2ee_association') AND sync_generation=CAST((SELECT value FROM meta WHERE key='sync_generation') AS INTEGER))").bind(association).bind(prefix).bind(descriptor.as_slice()).fetch_one(&mut *conn).await?;
138    ensure!(ok, "error encrypted-image-reinitialization-required");
139    super::super::client::initial_image_watermark(conn).await?;
140    Ok(())
141}
142
143pub(crate) async fn validate_integrity(conn: &mut SqliteConnection) -> Result<()> {
144    #[cfg(any(test, feature = "test-support"))]
145    let scan_start = Instant::now();
146    let missing:bool=sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM task_attachments a WHERE NOT EXISTS(SELECT 1 FROM local_e2ee_image_references r WHERE r.workspace=a.workspace_id AND r.reference=a.attachment_id) AND NOT EXISTS(SELECT 1 FROM changes c WHERE c.change_id=a.created_by_change_id AND c.server_seq IS NULL))").fetch_one(conn).await?;
147    #[cfg(any(test, feature = "test-support"))]
148    record_integrity_scan(scan_start);
149    ensure!(!missing, "error encrypted-image-reinitialization-required");
150    Ok(())
151}
152pub(crate) async fn accept(
153    conn: &mut SqliteConnection,
154    accepted: &Accepted,
155    change: &ChangeWire,
156) -> Result<()> {
157    let p = codec::parse(&accepted.record)?.projection;
158    match p {
159        Projection::Ref {
160            workspace,
161            task,
162            reference,
163            descriptor,
164            ..
165        } => {
166            let workspace: String =
167                sqlx::query_scalar("SELECT workspace_id FROM tasks WHERE id = ?")
168                    .bind(&task)
169                    .fetch_optional(&mut *conn)
170                    .await?
171                    .unwrap_or(workspace);
172            let d = Descriptor::decode(&descriptor)?;
173            let sha = change.payload["sha256"]
174                .as_str()
175                .context("error encrypted-image-hash")?;
176            let old: Option<(Vec<u8>, String)> = sqlx::query_as(
177                "SELECT descriptor,sha256 FROM local_e2ee_image_objects WHERE object=?",
178            )
179            .bind(d.object.as_slice())
180            .fetch_optional(&mut *conn)
181            .await?;
182            valid(
183                old.as_ref()
184                    .is_none_or(|(bytes, h)| *bytes == descriptor && h == sha),
185            )?;
186            sqlx::query("INSERT OR IGNORE INTO local_e2ee_image_objects(object,descriptor,sha256,origin) VALUES(?,?,?,?)").bind(d.object.as_slice()).bind(&descriptor).bind(sha).bind(&change.change_id).execute(&mut *conn).await?;
187            let old:Option<(String,Option<Vec<u8>>,String)>=sqlx::query_as("SELECT parent,object,origin FROM local_e2ee_image_references WHERE workspace=? AND reference=?").bind(&workspace).bind(&reference).fetch_optional(&mut *conn).await?;
188            valid(old.as_ref().is_none_or(|(p, o, id)| {
189                p == &task && o.as_deref() == Some(d.object.as_slice()) && id == &change.change_id
190            }))?;
191            if old.is_none() {
192                sqlx::query("UPDATE local_e2ee_image_objects SET verified=0 WHERE object=?")
193                    .bind(d.object.as_slice())
194                    .execute(&mut *conn)
195                    .await?;
196            }
197            sqlx::query("INSERT OR IGNORE INTO local_e2ee_image_references(workspace,reference,parent,object,origin,deleted) VALUES(?,?,?,?,?,0)").bind(workspace).bind(reference).bind(task).bind(d.object.as_slice()).bind(&change.change_id).execute(&mut *conn).await?;
198        }
199        Projection::Unref {
200            workspace,
201            task,
202            reference,
203        } => {
204            let workspace: String =
205                sqlx::query_scalar("SELECT workspace_id FROM tasks WHERE id = ?")
206                    .bind(&task)
207                    .fetch_optional(&mut *conn)
208                    .await?
209                    .unwrap_or(workspace);
210            let n=sqlx::query("UPDATE local_e2ee_image_references SET deleted=1 WHERE workspace=? AND reference=? AND parent=?").bind(workspace).bind(reference).bind(task).execute(&mut *conn).await?.rows_affected();
211            valid(n == 1)?;
212        }
213        _ => {}
214    }
215    sqlx::query("DELETE FROM local_e2ee_image_preparation WHERE operation_id=?")
216        .bind(&change.change_id)
217        .execute(conn)
218        .await?;
219    Ok(())
220}
221
222/// Exact image bytes for one bounded upload. No plaintext hash leaves core.
223pub struct Upload {
224    pub workspace: String,
225    pub descriptor: Vec<u8>,
226    pub object: [u8; 32],
227    pub commitment: [u8; 32],
228    pub records: Vec<Vec<u8>>,
229}
230impl Upload {
231    fn new(change: &ChangeWire, d: &Descriptor, records: Vec<Vec<u8>>) -> Result<Self> {
232        let descriptor = d.encode()?;
233        Ok(Self {
234            workspace: change.payload["workspace_id"]
235                .as_str()
236                .context("error encrypted-image-workspace")?
237                .into(),
238            commitment: hash(&descriptor),
239            descriptor,
240            object: d.object,
241            records,
242        })
243    }
244}
245
246/// Pins the source and stages exact ciphertext for a new image head.
247pub(in crate::sync::encrypted_tail) async fn stage(
248    conn: &mut SqliteConnection,
249    a: &Authority,
250    c: &ChangeWire,
251    blob_dir: &Path,
252) -> Result<(Projection, Upload)> {
253    let sha = c.payload["sha256"]
254        .as_str()
255        .context("error encrypted-image-source")?;
256    let source = read_source(
257        blob_dir,
258        sha,
259        c.payload["byte_size"]
260            .as_u64()
261            .context("error encrypted-image-size")?,
262    )
263    .await?;
264    let image = crate::attachments::decode::validate_image(
265        source,
266        Some(
267            c.payload["media_type"]
268                .as_str()
269                .context("error encrypted-image-media")?
270                .into(),
271        ),
272    )
273    .await?;
274    valid(
275        c.payload["width"].as_i64() == Some(image.facts.width)
276            && c.payload["height"].as_i64() == Some(image.facts.height),
277    )?;
278    let reusable:Option<Vec<u8>>=sqlx::query_scalar("SELECT o.descriptor FROM local_e2ee_image_objects o JOIN local_e2ee_image_references r ON r.object=o.object WHERE o.sha256=? AND r.workspace=? ORDER BY o.object LIMIT 1").bind(sha).bind(c.payload["workspace_id"].as_str()).fetch_optional(&mut *conn).await?;
279    let (d, records) = if let Some(bytes) = reusable {
280        let d = Descriptor::decode(&bytes)?;
281        let records = d.reconstruct(a, &image.bytes, sha)?;
282        (d, records)
283    } else {
284        Descriptor::seal(a, &image.bytes)?
285    };
286    let descriptor = d.encode()?;
287    sqlx::query("INSERT INTO local_e2ee_image_preparation(singleton,operation_id,descriptor,sha256) VALUES(1,?,?,?)").bind(&c.change_id).bind(&descriptor).bind(sha).execute(&mut *conn).await?;
288    for (i, record) in records.iter().enumerate() {
289        sqlx::query(
290            "INSERT INTO local_e2ee_image_staging(operation_id,chunk_index,bytes) VALUES(?,?,?)",
291        )
292        .bind(&c.change_id)
293        .bind(i as i64)
294        .bind(record)
295        .execute(&mut *conn)
296        .await?;
297    }
298    let (deleted,version):(bool,Option<String>)=sqlx::query_as("SELECT t.deleted,(SELECT version FROM field_versions f WHERE f.workspace_id=t.workspace_id AND f.entity_type='task' AND f.entity_id=t.id AND f.field='deleted') FROM tasks t WHERE t.id=?").bind(&c.entity_id).fetch_one(&mut *conn).await?;
299    let projection = Projection::Ref {
300        workspace: c.payload["workspace_id"]
301            .as_str()
302            .context("error encrypted-image-workspace")?
303            .into(),
304        task: c.entity_id.clone(),
305        reference: c.payload["attachment_id"]
306            .as_str()
307            .context("error encrypted-image-reference")?
308            .into(),
309        descriptor,
310        deleted,
311        version,
312    };
313    Ok((projection, Upload::new(c, &d, records)?))
314}
315
316/// Verifies the frozen Ref's staged ciphertext, rebuilding missing chunks only
317/// from the saved recipe and the pinned source.
318pub(in crate::sync::encrypted_tail) async fn frozen_upload(
319    conn: &mut SqliteConnection,
320    a: &Authority,
321    c: &ChangeWire,
322    record: &[u8],
323    blob_dir: &Path,
324) -> Result<Upload> {
325    let (bytes, saved_sha): (Vec<u8>, String) = sqlx::query_as(
326        "SELECT descriptor,sha256 FROM local_e2ee_image_preparation WHERE operation_id=?",
327    )
328    .bind(&c.change_id)
329    .fetch_optional(&mut *conn)
330    .await?
331    .context("error encrypted-image-preparation-required")?;
332    let Projection::Ref { descriptor, .. } = codec::parse(record)?.projection else {
333        anyhow::bail!("error encrypted-image-preparation")
334    };
335    valid(descriptor == bytes && c.payload["sha256"].as_str() == Some(&saved_sha))?;
336    let d = Descriptor::decode(&bytes)?;
337    let rows: Vec<(i64, Vec<u8>)> = sqlx::query_as("SELECT chunk_index,bytes FROM local_e2ee_image_staging WHERE operation_id=? ORDER BY chunk_index")
338        .bind(&c.change_id).fetch_all(&mut *conn).await?;
339    for (index, chunk) in &rows {
340        d.verify_chunk(usize::try_from(*index)?, chunk)?;
341    }
342    let records = if rows.len() == d.artifact.chunks.len() {
343        let records: Vec<Vec<u8>> = rows.into_iter().map(|(_, b)| b).collect();
344        d.verify(&records)?;
345        records
346    } else {
347        let source = read_source(blob_dir, &saved_sha, d.artifact.total).await?;
348        let records = d.reconstruct(a, &source, &saved_sha)?;
349        for (i, record) in records.iter().enumerate() {
350            sqlx::query("INSERT OR IGNORE INTO local_e2ee_image_staging(operation_id,chunk_index,bytes) VALUES(?,?,?)").bind(&c.change_id).bind(i as i64).bind(record).execute(&mut *conn).await?;
351        }
352        records
353    };
354    Upload::new(c, &d, records)
355}
356
357pub(in crate::sync::encrypted_tail) async fn supersede(
358    conn: &mut SqliteConnection,
359    a: &Authority,
360    change: &ChangeWire,
361    mut projection: Projection,
362    blob_dir: &Path,
363) -> Result<Projection> {
364    let Projection::Ref { descriptor, .. } = &mut projection else {
365        anyhow::bail!("error encrypted-image-preparation");
366    };
367    let (stored, sha): (Vec<u8>, String) = sqlx::query_as(
368        "SELECT descriptor,sha256 FROM local_e2ee_image_preparation WHERE operation_id=?",
369    )
370    .bind(&change.change_id)
371    .fetch_one(&mut *conn)
372    .await?;
373    valid(stored == *descriptor && change.payload["sha256"].as_str() == Some(&sha))?;
374    let old = Descriptor::decode(&stored)?;
375    let rows: Vec<(i64, Vec<u8>)> = sqlx::query_as("SELECT chunk_index,bytes FROM local_e2ee_image_staging WHERE operation_id=? ORDER BY chunk_index")
376        .bind(&change.change_id).fetch_all(&mut *conn).await?;
377    for (index, bytes) in &rows {
378        old.verify_chunk(usize::try_from(*index)?, bytes)?;
379    }
380    let source = if rows.len() == old.artifact.chunks.len() {
381        old.open(a, &rows.into_iter().map(|(_, b)| b).collect::<Vec<_>>())?
382            .to_vec()
383    } else {
384        let source = read_source(blob_dir, &sha, old.artifact.total).await?;
385        old.reconstruct(a, &source, &sha)?;
386        source
387    };
388    valid(hex::encode(hash(&source)) == sha)?;
389    let (next, records) = Descriptor::seal(a, &source)?;
390    // The owning outbox stays present and the source pin is replaced in this transaction.
391    sqlx::query("DELETE FROM local_e2ee_image_preparation WHERE operation_id=?")
392        .bind(&change.change_id)
393        .execute(&mut *conn)
394        .await?;
395    *descriptor = next.encode()?;
396    sqlx::query("INSERT INTO local_e2ee_image_preparation(singleton,operation_id,descriptor,sha256) VALUES(1,?,?,?)")
397        .bind(&change.change_id).bind(&*descriptor).bind(sha).execute(&mut *conn).await?;
398    for (index, record) in records.iter().enumerate() {
399        sqlx::query(
400            "INSERT INTO local_e2ee_image_staging(operation_id,chunk_index,bytes) VALUES(?,?,?)",
401        )
402        .bind(&change.change_id)
403        .bind(index as i64)
404        .bind(record)
405        .execute(&mut *conn)
406        .await?;
407    }
408    Ok(projection)
409}
410/// The local plaintext image source is missing or no longer matches its hash.
411#[derive(Debug)]
412pub struct ImageSourceUnavailable;
413impl std::fmt::Display for ImageSourceUnavailable {
414    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
415        f.write_str("error encrypted-image-source-unavailable")
416    }
417}
418async fn read_source(blob_dir: &Path, sha: &str, total: u64) -> Result<Vec<u8>> {
419    use tokio::io::AsyncReadExt;
420    async {
421        let file =
422            tokio::fs::File::open(crate::attachments::storage::object_path(blob_dir, sha)?).await?;
423        let mut bytes = Vec::new();
424        file.take(super::codec::IMAGE_BYTES as u64 + 1)
425            .read_to_end(&mut bytes)
426            .await?;
427        valid(
428            bytes.len() as u64 == total
429                && bytes.len() <= super::codec::IMAGE_BYTES
430                && hex::encode(hash(&bytes)) == sha,
431        )?;
432        anyhow::Ok(bytes)
433    }
434    .await
435    .context(ImageSourceUnavailable)
436}
437
438/// Image demand after initial catch-up. Reading it never consumes a selection turn.
439pub(in crate::sync::encrypted_tail) async fn downloads(
440    conn: &mut SqliteConnection,
441) -> Result<Downloads> {
442    let (pending, unavailable): (bool, bool) = sqlx::query_as(sqlx::AssertSqlSafe(format!(
443        "SELECT EXISTS(SELECT 1 {DOWNLOAD_CANDIDATES}),
444                EXISTS(SELECT 1 FROM local_e2ee_image_references r
445                       JOIN task_attachments a ON a.workspace_id=r.workspace AND a.attachment_id=r.reference
446                       JOIN tasks t ON t.workspace_id=a.workspace_id AND t.id=a.task_id
447                       WHERE r.object IS NULL AND a.deleted=0 AND t.deleted=0)"
448    )))
449    .fetch_one(conn)
450    .await?;
451    Ok(Downloads {
452        pending,
453        unavailable,
454    })
455}
456
457pub struct Download {
458    pub workspace: String,
459    pub object: [u8; 32],
460    pub descriptor_commitment: [u8; 32],
461    pub chunk_count: usize,
462    descriptor: Vec<u8>,
463    sha256: String,
464}
465impl Database {
466    /// Images whose bytes this device still has to download. Counting reads
467    /// only local state and never consumes a selection turn.
468    pub async fn encrypted_image_downloads_remaining(&self) -> Result<u64> {
469        let mut conn = self.acquire_reader().await?;
470        let count: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
471            "SELECT COUNT(DISTINCT o.object) {DOWNLOAD_CANDIDATES}"
472        )))
473        .fetch_one(&mut *conn)
474        .await?;
475        Ok(u64::try_from(count)?)
476    }
477    /// Selects and advances past one pending object before transfer. The local
478    /// cursor orders attempts, not acceptance, validation or content progress.
479    pub async fn prepare_encrypted_image_download(
480        &self,
481        a: &Authority,
482    ) -> Result<Option<Download>> {
483        let mut conn = self.acquire_writer().await?;
484        let mut tx = begin_immediate(&mut conn).await?;
485        super::super::client::validate_binding_and_cursor(&mut tx, a).await?;
486        ensure!(
487            super::super::client::initial_image_watermark(&mut tx).await? == "ready",
488            "error encrypted-image-initial-catch-up"
489        );
490        let after = hex::decode(
491            db::get_meta(&mut tx, DOWNLOAD_CURSOR)
492                .await?
493                .unwrap_or_default(),
494        )
495        .context("error encrypted-image-download-cursor")?;
496        valid(after.is_empty() || after.len() == 32)?;
497        let row = match select_download_candidate(&mut tx, Some(&after)).await? {
498            Some(row) => Some(row),
499            None => select_download_candidate(&mut tx, None).await?,
500        };
501        let result = row
502            .map(|(workspace, object, bytes, sha256)| -> Result<_> {
503                let d = Descriptor::decode(&bytes)?;
504                valid(object.as_slice() == d.object)?;
505                Ok(Download {
506                    workspace,
507                    object: d.object,
508                    descriptor_commitment: hash(&bytes),
509                    chunk_count: d.artifact.chunks.len(),
510                    descriptor: bytes,
511                    sha256,
512                })
513            })
514            .transpose()?;
515        if let Some(download) = &result {
516            db::set_meta(&mut tx, DOWNLOAD_CURSOR, &hex::encode(download.object)).await?;
517        }
518        tx.commit().await?;
519        Ok(result)
520    }
521    pub async fn install_encrypted_image(
522        &self,
523        a: &Authority,
524        blob_dir: &Path,
525        download: &Download,
526        records: &[Vec<u8>],
527        policy: crate::attachments::LifecyclePolicy,
528    ) -> Result<()> {
529        let d = Descriptor::decode(&download.descriptor)?;
530        let plain = d.open(a, records)?;
531        valid(hex::encode(hash(&plain)) == download.sha256)?;
532        self.install_image_plaintext(a, blob_dir, download, plain.to_vec(), policy)
533            .await
534    }
535    pub async fn complete_encrypted_image_from_local(
536        &self,
537        a: &Authority,
538        blob_dir: &Path,
539        download: &Download,
540        policy: crate::attachments::LifecyclePolicy,
541    ) -> Result<bool> {
542        let d = Descriptor::decode(&download.descriptor)?;
543        d.authority(a)?;
544        match read_source(blob_dir, &download.sha256, d.artifact.total).await {
545            Ok(bytes) => {
546                self.install_image_plaintext(a, blob_dir, download, bytes, policy)
547                    .await?;
548                Ok(true)
549            }
550            Err(_) => Ok(false),
551        }
552    }
553    async fn install_image_plaintext(
554        &self,
555        a: &Authority,
556        blob_dir: &Path,
557        download: &Download,
558        plain: Vec<u8>,
559        policy: crate::attachments::LifecyclePolicy,
560    ) -> Result<()> {
561        let d = Descriptor::decode(&download.descriptor)?;
562        let image = crate::attachments::decode::validate_image(plain, None)
563            .await
564            .map_err(|_| anyhow::anyhow!("error encrypted-image-invalid"))?;
565        let mut conn = self.acquire_writer().await?;
566        super::super::client::validate_binding_and_cursor(&mut conn, a).await?;
567        let reservation = crate::attachments::lifecycle::ensure_local_capacity(
568            &mut conn,
569            blob_dir,
570            &download.sha256,
571            image.bytes.len() as i64,
572            policy,
573            &crate::attachments::lifecycle::SystemClock,
574        )
575        .await?;
576        let result = async {
577            let mut tx = begin_immediate(&mut conn).await?;
578            super::super::client::validate_binding_and_cursor(&mut tx, a).await?;
579            let descriptor: Vec<u8> = sqlx::query_scalar(
580                "SELECT descriptor FROM local_e2ee_image_objects WHERE object=? AND sha256=?",
581            )
582            .bind(d.object.as_slice())
583            .bind(&download.sha256)
584            .fetch_one(&mut *tx)
585            .await?;
586            valid(descriptor == download.descriptor)?;
587            let facts: Vec<(i64, String, Option<i64>, Option<i64>)> = sqlx::query_as(
588                "SELECT byte_size,media_type,width,height FROM task_attachments WHERE sha256=?",
589            )
590            .bind(&download.sha256)
591            .fetch_all(&mut *tx)
592            .await?;
593            for (size, media, width, height) in facts {
594                valid(
595                    size == image.bytes.len() as i64
596                        && media == image.facts.media_type
597                        && width == Some(image.facts.width)
598                        && height == Some(image.facts.height),
599                )?;
600            }
601            let stored =
602                crate::attachments::storage::stage_blob(blob_dir, &download.sha256, &image.bytes)
603                    .await?;
604            crate::attachments::storage::upsert_inventory_available(
605                &mut tx,
606                &download.sha256,
607                stored.byte_size,
608                &image.facts.media_type,
609            )
610            .await?;
611            crate::attachments::lifecycle::reconcile_liveness_for_hashes_in_transaction(
612                &mut tx,
613                std::slice::from_ref(&download.sha256),
614                &crate::attachments::lifecycle::SystemClock,
615            )
616            .await?;
617            sqlx::query("UPDATE local_e2ee_image_objects SET verified=1 WHERE object=?")
618                .bind(d.object.as_slice())
619                .execute(&mut *tx)
620                .await?;
621            tx.commit().await?;
622            Ok(())
623        }
624        .await;
625        if let Some(reservation) = reservation {
626            crate::attachments::lifecycle::release_reservation(&mut conn, &reservation).await?;
627        }
628        result
629    }
630}
631
632impl Database {
633    pub async fn encrypted_tail_frozen_record(
634        &self,
635        a: &Authority,
636    ) -> Result<Option<(String, Vec<u8>)>> {
637        let mut conn = self.acquire_reader().await?;
638        super::super::client::validate_binding_and_cursor(&mut conn, a).await?;
639        let record: Option<Vec<u8>> =
640            sqlx::query_scalar("SELECT record FROM local_e2ee_outbox ORDER BY position LIMIT 1")
641                .fetch_optional(&mut *conn)
642                .await?;
643        record
644            .map(|r| Ok((codec::open(a, &r)?.change_id, r)))
645            .transpose()
646    }
647}
648
649impl Database {
650    /// Exact targeted repair uses only an authenticated admitted mapping.
651    pub async fn repair_encrypted_image(
652        &self,
653        a: &Authority,
654        blob_dir: &Path,
655        workspace: &str,
656        reference: &str,
657    ) -> Result<Upload> {
658        let mut conn = self.acquire_writer().await?;
659        let mut tx = begin_immediate(&mut conn).await?;
660        super::super::client::validate_binding_and_cursor(&mut tx, a).await?;
661        ensure!(!a.rotation_pending(), "error membership-rotation-pending");
662        let (descriptor,sha):(Vec<u8>,String)=sqlx::query_as("SELECT o.descriptor,o.sha256 FROM local_e2ee_image_references r JOIN local_e2ee_image_objects o ON o.object=r.object WHERE r.workspace=? AND r.reference=?").bind(workspace).bind(reference).fetch_optional(&mut *tx).await?.context("error encrypted-image-unavailable")?;
663        let d = Descriptor::decode(&descriptor)?;
664        let source = read_source(blob_dir, &sha, d.artifact.total).await?;
665        let records = d.reconstruct(a, &source, &sha)?;
666        tx.commit().await?;
667        Ok(Upload {
668            workspace: workspace.into(),
669            object: d.object,
670            commitment: hash(&descriptor),
671            descriptor,
672            records,
673        })
674    }
675}
676
677#[cfg(test)]
678mod tests {
679    use super::*;
680    use sqlx::Connection;
681
682    fn object(index: u32) -> Vec<u8> {
683        let mut object = vec![0; 32];
684        object[28..].copy_from_slice(&index.to_be_bytes());
685        object
686    }
687
688    #[tokio::test]
689    async fn download_selection_uses_keyset_order_and_wraps() {
690        let mut conn = SqliteConnection::connect(":memory:").await.unwrap();
691        sqlx::raw_sql(
692            "CREATE TABLE local_e2ee_image_objects(
693                 object BLOB PRIMARY KEY, descriptor BLOB, sha256 TEXT,
694                 verified INTEGER, origin TEXT);
695             CREATE TABLE local_e2ee_image_references(
696                 workspace TEXT, reference TEXT, parent TEXT, object BLOB,
697                 origin TEXT, deleted INTEGER, PRIMARY KEY(workspace,reference));
698             CREATE INDEX local_e2ee_image_references_object
699                 ON local_e2ee_image_references(object);
700             CREATE TABLE task_attachments(
701                 workspace_id TEXT, attachment_id TEXT, task_id TEXT, deleted INTEGER,
702                 PRIMARY KEY(workspace_id,attachment_id));
703             CREATE TABLE tasks(
704                 workspace_id TEXT, id TEXT, deleted INTEGER,
705                 PRIMARY KEY(workspace_id,id));
706             CREATE TABLE blob_inventory(sha256 TEXT PRIMARY KEY, available INTEGER);
707             INSERT INTO tasks VALUES('workspace','task',0);",
708        )
709        .execute(&mut conn)
710        .await
711        .unwrap();
712        let mut tx = conn.begin().await.unwrap();
713        for index in 0..4096_u32 {
714            let object = object(index);
715            let reference = format!("reference-{index:04}");
716            sqlx::query("INSERT INTO local_e2ee_image_objects VALUES(?,X'00',?,0,'test')")
717                .bind(&object)
718                .bind(format!("sha-{index}"))
719                .execute(&mut *tx)
720                .await
721                .unwrap();
722            sqlx::query(
723                "INSERT INTO local_e2ee_image_references VALUES('workspace',?,'task',?,'test',0)",
724            )
725            .bind(&reference)
726            .bind(&object)
727            .execute(&mut *tx)
728            .await
729            .unwrap();
730            sqlx::query("INSERT INTO task_attachments VALUES('workspace',?,'task',0)")
731                .bind(reference)
732                .execute(&mut *tx)
733                .await
734                .unwrap();
735        }
736        tx.commit().await.unwrap();
737
738        let selected = select_download_candidate(&mut conn, Some(&object(2047)))
739            .await
740            .unwrap()
741            .unwrap();
742        assert_eq!(selected.1, object(2048));
743        assert!(
744            select_download_candidate(&mut conn, Some(&object(4095)))
745                .await
746                .unwrap()
747                .is_none()
748        );
749        let wrapped = select_download_candidate(&mut conn, None)
750            .await
751            .unwrap()
752            .unwrap();
753        assert_eq!(wrapped.1, object(0));
754
755        let plan: Vec<(i64, i64, i64, String)> = sqlx::query_as(
756            "EXPLAIN QUERY PLAN SELECT o.object
757             FROM local_e2ee_image_objects o
758             WHERE o.object>? AND EXISTS(
759                 SELECT 1 FROM local_e2ee_image_references r
760                 WHERE r.object=o.object)
761             ORDER BY o.object LIMIT 1",
762        )
763        .bind(object(2047))
764        .fetch_all(&mut conn)
765        .await
766        .unwrap();
767        assert!(
768            plan.iter().any(|row| row.3.contains("object>?")),
769            "{plan:?}"
770        );
771        assert!(
772            !plan.iter().any(|row| row.3.contains("TEMP B-TREE")),
773            "{plan:?}"
774        );
775    }
776
777    #[tokio::test]
778    async fn valid_image_aead_does_not_make_invalid_plaintext_available() {
779        let root = tempfile::tempdir().unwrap();
780        let db = Database::open(&root.path().join("test.sqlite"))
781            .await
782            .unwrap();
783        let a = crate::sync::encrypted_tail::tests::authority();
784        let plain = b"not an image";
785        let (d, records) = Descriptor::seal(&a, plain).unwrap();
786        assert_eq!(d.open(&a, &records).unwrap().as_slice(), plain);
787        let bytes = d.encode().unwrap();
788        let download = Download {
789            workspace: "0000000000000000".into(),
790            object: d.object,
791            descriptor_commitment: hash(&bytes),
792            chunk_count: records.len(),
793            descriptor: bytes,
794            sha256: hex::encode(hash(plain)),
795        };
796        let error = db
797            .install_encrypted_image(&a, root.path(), &download, &records, Default::default())
798            .await
799            .unwrap_err();
800        assert_eq!(error.to_string(), "error encrypted-image-invalid");
801        assert!(!root.path().join("objects").exists());
802    }
803}
aven_core/sync/encrypted_tail/attachments/client.rs

aven_core/sync/encrypted_tail/attachments/
client.rs
