aven_core/sync/encrypted_tail/attachments/
server.rs

1use super::super::{Context, hash, valid};
2use super::{codec::Descriptor, *};
3#[cfg(test)]
4use crate::sync::encrypted_tail::attachments::codec::DescriptorAuthority;
5use crate::sync::seed_claim::membership::Membership;
6use crate::{
7    attachments::lifecycle::LifecyclePolicy,
8    db::{Database, begin_immediate},
9    sync::seed_claim::Secret,
10};
11use anyhow::{Context as _, Result, ensure};
12use sqlx::SqliteConnection;
13
14async fn load(
15    conn: &mut SqliteConnection,
16    object: &[u8; 32],
17    commitment: &[u8; 32],
18) -> Result<(Descriptor, bool)> {
19    let (bytes, complete): (Vec<u8>, bool) =
20        sqlx::query_as("SELECT descriptor,complete FROM server_e2ee_images WHERE object=?")
21            .bind(object.as_slice())
22            .fetch_optional(conn)
23            .await?
24            .context("error encrypted-image-unknown")?;
25    valid(hash(&bytes) == *commitment)?;
26    let d = Descriptor::decode(&bytes)?;
27    valid(d.object == *object)?;
28    Ok((d, complete))
29}
30/// Stamps newly unreferenced images and clears the stamp from images that are
31/// referenced or ticketed again. Only the background prune runs this full
32/// scan, so an image's grace period starts at the first prune pass that
33/// observes it unreferenced.
34async fn refresh(conn: &mut SqliteConnection, now: i64) -> Result<()> {
35    sqlx::query("UPDATE server_e2ee_images SET unreferenced_at=CASE WHEN EXISTS(SELECT 1 FROM server_e2ee_image_references r JOIN server_e2ee_image_parents p ON p.workspace=r.workspace AND p.parent=r.parent WHERE r.object=server_e2ee_images.object AND r.deleted=0 AND (p.deleted=0 OR p.protected=1 OR p.version IS NULL)) OR EXISTS(SELECT 1 FROM server_e2ee_image_tickets t WHERE t.object=server_e2ee_images.object AND t.expires_at>?) THEN NULL ELSE COALESCE(unreferenced_at,?) END").bind(now).bind(now).execute(conn).await?;
36    Ok(())
37}
38/// Clears the unreferenced stamp of images a live parent references, so a
39/// parent that is deleted again later gets a full grace period.
40pub(in crate::sync::encrypted_tail) async fn retain_parent(
41    conn: &mut SqliteConnection,
42    workspace: &str,
43    parent: &str,
44) -> Result<()> {
45    sqlx::query("UPDATE server_e2ee_images SET unreferenced_at=NULL WHERE unreferenced_at IS NOT NULL AND object IN (SELECT r.object FROM server_e2ee_image_references r JOIN server_e2ee_image_parents p ON p.workspace=r.workspace AND p.parent=r.parent WHERE r.workspace=? AND r.parent=? AND r.deleted=0 AND (p.deleted=0 OR p.protected=1 OR p.version IS NULL))").bind(workspace).bind(parent).execute(conn).await?;
46    Ok(())
47}
48async fn protected(
49    conn: &mut SqliteConnection,
50    object: &[u8; 32],
51    workspace: &str,
52) -> Result<bool> {
53    Ok(sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_image_references r JOIN server_e2ee_image_parents p ON p.workspace=r.workspace AND p.parent=r.parent WHERE r.object=? AND r.workspace=? AND r.deleted=0 AND (p.deleted=0 OR p.protected=1 OR p.version IS NULL))").bind(object.as_slice()).bind(workspace).fetch_one(conn).await?)
54}
55async fn scope(conn: &mut SqliteConnection, object: &[u8; 32], workspace: &str) -> Result<()> {
56    valid(
57        sqlx::query_scalar(
58            "SELECT EXISTS(SELECT 1 FROM server_e2ee_image_scopes WHERE object=? AND workspace=?)",
59        )
60        .bind(object.as_slice())
61        .bind(workspace)
62        .fetch_one(conn)
63        .await?,
64    )
65}
66async fn reserve(
67    conn: &mut SqliteConnection,
68    d: &Descriptor,
69    workspace: &str,
70    device: &[u8; 32],
71    quota: i64,
72    now: i64,
73) -> Result<()> {
74    sqlx::query("DELETE FROM server_e2ee_image_tickets WHERE expires_at<=?")
75        .bind(now)
76        .execute(&mut *conn)
77        .await?;
78    let existing: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_image_tickets WHERE object=? AND workspace=? AND device=? AND expires_at>?)").bind(d.object.as_slice()).bind(workspace).bind(device.as_slice()).bind(now).fetch_one(&mut *conn).await?;
79    if existing {
80        return Ok(());
81    }
82    let reactivating: bool = sqlx::query_scalar("SELECT origin IS NULL AND NOT EXISTS(SELECT 1 FROM server_e2ee_image_chunks c WHERE c.object=i.object) AND NOT EXISTS(SELECT 1 FROM server_e2ee_image_tickets t WHERE t.object=i.object AND t.expires_at>?) FROM server_e2ee_images i WHERE object=?")
83        .bind(now).bind(d.object.as_slice()).fetch_one(&mut *conn).await?;
84    if reactivating {
85        let active: i64 = sqlx::query_scalar("SELECT count(*) FROM server_e2ee_images i WHERE origin IS NULL AND (EXISTS(SELECT 1 FROM server_e2ee_image_chunks c WHERE c.object=i.object) OR EXISTS(SELECT 1 FROM server_e2ee_image_tickets t WHERE t.object=i.object AND t.expires_at>?))")
86            .bind(now).fetch_one(&mut *conn).await?;
87        valid(active < 128)?;
88    }
89    let counted = protected(conn, &d.object, workspace).await? || sqlx::query_scalar::<_,bool>("SELECT EXISTS(SELECT 1 FROM server_e2ee_image_tickets WHERE object=? AND workspace=? AND expires_at>?)").bind(d.object.as_slice()).bind(workspace).bind(now).fetch_one(&mut *conn).await?;
90    if !counted {
91        let used: i64 = sqlx::query_scalar("SELECT COALESCE(SUM(i.byte_size),0) FROM server_e2ee_images i WHERE EXISTS(SELECT 1 FROM server_e2ee_image_references r JOIN server_e2ee_image_parents p ON p.workspace=r.workspace AND p.parent=r.parent WHERE r.object=i.object AND r.workspace=? AND r.deleted=0 AND (p.deleted=0 OR p.protected=1 OR p.version IS NULL)) OR EXISTS(SELECT 1 FROM server_e2ee_image_tickets t WHERE t.object=i.object AND t.workspace=? AND t.expires_at>?)").bind(workspace).bind(workspace).bind(now).fetch_one(&mut *conn).await?;
92        ensure!(
93            used.checked_add(d.byte_size()).is_some_and(|n| n <= quota),
94            "error attachment-quota-exceeded"
95        );
96    }
97    let count: i64 =
98        sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_tickets WHERE expires_at>?")
99            .bind(now)
100            .fetch_one(&mut *conn)
101            .await?;
102    valid(count < 128)?;
103    let mut id = [0; 32];
104    getrandom::fill(&mut id).map_err(|_| anyhow::anyhow!("error encrypted-image-entropy"))?;
105    let expires = now
106        .checked_add(crate::attachments::lifecycle::LEASE_TTL.as_secs() as i64)
107        .context("error encrypted-image-clock")?;
108    sqlx::query("INSERT INTO server_e2ee_image_tickets(reservation,object,workspace,device,expires_at) VALUES(?,?,?,?,?) ON CONFLICT(object,workspace,device) DO UPDATE SET reservation=excluded.reservation,expires_at=excluded.expires_at")
109        .bind(id.as_slice()).bind(d.object.as_slice()).bind(workspace).bind(device.as_slice()).bind(expires).execute(&mut *conn).await?;
110    sqlx::query("UPDATE server_e2ee_images SET unreferenced_at=NULL WHERE object=?")
111        .bind(d.object.as_slice())
112        .execute(conn)
113        .await?;
114    Ok(())
115}
116async fn ticket(
117    conn: &mut SqliteConnection,
118    object: &[u8; 32],
119    workspace: &str,
120    device: &[u8; 32],
121    ticket: &Ticket,
122    now: i64,
123) -> Result<()> {
124    valid(sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_image_tickets WHERE object=? AND workspace=? AND device=? AND reservation=? AND expires_at>?)").bind(object.as_slice()).bind(workspace).bind(device.as_slice()).bind(ticket.reservation.as_slice()).bind(now).fetch_one(conn).await?)
125}
126async fn status(
127    conn: &mut SqliteConnection,
128    d: &Descriptor,
129    workspace: &str,
130    device: &[u8; 32],
131    now: i64,
132) -> Result<Reply> {
133    let complete: bool =
134        sqlx::query_scalar("SELECT complete FROM server_e2ee_images WHERE object=?")
135            .bind(d.object.as_slice())
136            .fetch_one(&mut *conn)
137            .await?;
138    let indices: Vec<i64> = sqlx::query_scalar(
139        "SELECT chunk_index FROM server_e2ee_image_chunks WHERE object=? ORDER BY chunk_index",
140    )
141    .bind(d.object.as_slice())
142    .fetch_all(&mut *conn)
143    .await?;
144    let t:Option<(Vec<u8>,i64)>=sqlx::query_as("SELECT reservation,expires_at FROM server_e2ee_image_tickets WHERE object=? AND workspace=? AND device=? AND expires_at>?").bind(d.object.as_slice()).bind(workspace).bind(device.as_slice()).bind(now).fetch_optional(conn).await?;
145    let (reservation, expires_at) = match t {
146        Some((id, e)) => (
147            Some(
148                id.try_into()
149                    .map_err(|_| anyhow::anyhow!("error encrypted-image-storage"))?,
150            ),
151            Some(e),
152        ),
153        None => (None, None),
154    };
155    Ok(Reply::Status(Status {
156        complete,
157        missing: (0..d.artifact.chunks.len())
158            .filter(|i| !indices.contains(&(*i as i64)))
159            .collect(),
160        reservation,
161        expires_at,
162    }))
163}
164async fn eligible_object(
165    conn: &mut SqliteConnection,
166    m: &Membership,
167    d: &Descriptor,
168) -> Result<()> {
169    // Unknown generations fail before querying historical provenance.
170    valid(m.generations().iter().any(|g| g.id == d.generation))?;
171    let admitted = if d.generation != m.current_generation().id {
172        sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_images WHERE object=? AND origin IS NOT NULL AND descriptor=?)")
173            .bind(d.object.as_slice()).bind(d.encode()?).fetch_one(conn).await?
174    } else {
175        false
176    };
177    aven_protocol::tail::eligible_image(m, d.generation, admitted)?;
178    Ok(())
179}
180impl Database {
181    /// Deletes the stored bytes of at most `limit` images that have been
182    /// unreferenced and unticketed for at least `grace`, keeping their
183    /// descriptors. Returns how many images were pruned.
184    pub async fn prune_encrypted_images(
185        &self,
186        grace: std::time::Duration,
187        limit: usize,
188    ) -> Result<usize> {
189        let mut conn = self.acquire_writer().await?;
190        let mut tx = begin_immediate(&mut conn).await?;
191        let now = chrono::Utc::now().timestamp();
192        refresh(&mut tx, now).await?;
193        let cutoff = now
194            .checked_sub(i64::try_from(grace.as_secs())?)
195            .context("error encrypted-image-clock")?;
196        let objects:Vec<Vec<u8>>=sqlx::query_scalar("SELECT object FROM server_e2ee_images WHERE unreferenced_at<=? AND (complete=1 OR EXISTS(SELECT 1 FROM server_e2ee_image_chunks c WHERE c.object=server_e2ee_images.object)) ORDER BY object LIMIT ?").bind(cutoff).bind(i64::try_from(limit)?).fetch_all(&mut *tx).await?;
197        // Objects with live tickets were excluded by `refresh`, so no
198        // uploader can still write to a pruned object.
199        for object in &objects {
200            sqlx::query("UPDATE server_e2ee_images SET complete=0 WHERE object=?")
201                .bind(object)
202                .execute(&mut *tx)
203                .await?;
204            sqlx::query("DELETE FROM server_e2ee_image_chunks WHERE object=?")
205                .bind(object)
206                .execute(&mut *tx)
207                .await?;
208        }
209        tx.commit().await?;
210        Ok(objects.len())
211    }
212    pub async fn encrypted_image_exchange(
213        &self,
214        context: &Context,
215        bearer: &Secret,
216        op: Operation,
217        policy: LifecyclePolicy,
218    ) -> Result<Reply> {
219        let mut conn = self.acquire_writer().await?;
220        let mut tx = begin_immediate(&mut conn).await?;
221        let current =
222            crate::sync::seed_claim::membership::persistence::current(self, &mut tx).await?;
223        current
224            .membership
225            .authenticate(&context.authentication(bearer), false)?;
226        ensure!(
227            !current.membership.rotation_pending()
228                || matches!(&op, Operation::Status { .. } | Operation::Read { .. }),
229            "error membership-rotation-pending"
230        );
231        let binding = current.membership.publication().binding();
232        valid(
233            context.stream == binding.stream_id
234                && context.descriptor == binding.descriptor_commitment,
235        )?;
236        let now = chrono::Utc::now().timestamp();
237        let reply = match op {
238            Operation::Declare {
239                workspace,
240                descriptor,
241            } => {
242                workspace.parse::<crate::ids::WorkspaceId>()?;
243                let d = Descriptor::decode(&descriptor)?;
244                valid(d.vault == context.vault && d.stream == context.stream)?;
245                eligible_object(&mut tx, &current.membership, &d).await?;
246                let old: Option<Vec<u8>> =
247                    sqlx::query_scalar("SELECT descriptor FROM server_e2ee_images WHERE object=?")
248                        .bind(d.object.as_slice())
249                        .fetch_optional(&mut *tx)
250                        .await?;
251                if let Some(old) = old {
252                    valid(old == descriptor)?;
253                } else {
254                    let total: i64 = sqlx::query_scalar("SELECT count(*) FROM server_e2ee_images")
255                        .fetch_one(&mut *tx)
256                        .await?;
257                    valid(total < 65536)?;
258                    sqlx::query("INSERT INTO server_e2ee_images(object,byte_size,descriptor,complete,unreferenced_at) VALUES(?,?,?,0,?)").bind(d.object.as_slice()).bind(d.byte_size()).bind(&descriptor).bind(now).execute(&mut *tx).await?;
259                }
260                let known:bool=sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_image_scopes WHERE object=? AND workspace=?)").bind(d.object.as_slice()).bind(&workspace).fetch_one(&mut *tx).await?;
261                if !known {
262                    let count: i64 =
263                        sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_scopes")
264                            .fetch_one(&mut *tx)
265                            .await?;
266                    valid(count < 262144)?;
267                }
268                sqlx::query(
269                    "INSERT OR IGNORE INTO server_e2ee_image_scopes(object,workspace) VALUES(?,?)",
270                )
271                .bind(d.object.as_slice())
272                .bind(&workspace)
273                .execute(&mut *tx)
274                .await?;
275                load(&mut tx, &d.object, &hash(&descriptor)).await?;
276                reserve(
277                    &mut tx,
278                    &d,
279                    &workspace,
280                    &context.device,
281                    policy.quota_bytes,
282                    now,
283                )
284                .await?;
285                status(&mut tx, &d, &workspace, &context.device, now).await?
286            }
287            other => dispatch_existing(&mut tx, context, &current.membership, other, now).await?,
288        };
289        tx.commit().await?;
290        Ok(reply)
291    }
292}
293async fn dispatch_existing(
294    conn: &mut SqliteConnection,
295    context: &Context,
296    membership: &Membership,
297    op: Operation,
298    now: i64,
299) -> Result<Reply> {
300    let (workspace, object, commitment) = match &op {
301        Operation::Status {
302            workspace,
303            object,
304            descriptor_commitment,
305        }
306        | Operation::Put {
307            workspace,
308            object,
309            descriptor_commitment,
310            ..
311        }
312        | Operation::Complete {
313            workspace,
314            object,
315            descriptor_commitment,
316            ..
317        }
318        | Operation::Read {
319            workspace,
320            object,
321            descriptor_commitment,
322            ..
323        }
324        | Operation::Release {
325            workspace,
326            object,
327            descriptor_commitment,
328            ..
329        } => (workspace, *object, *descriptor_commitment),
330        _ => unreachable!(),
331    };
332    workspace.parse::<crate::ids::WorkspaceId>()?;
333    scope(conn, &object, workspace).await?;
334    let (d, _) = load(conn, &object, &commitment).await?;
335    valid(d.vault == context.vault && d.stream == context.stream)?;
336    eligible_object(conn, membership, &d).await?;
337    match &op {
338        Operation::Status { .. } => status(conn, &d, workspace, &context.device, now).await,
339        Operation::Read { index, .. } => {
340            valid(*index < d.artifact.chunks.len())?;
341            let admitted: bool = sqlx::query_scalar(
342                "SELECT origin IS NOT NULL FROM server_e2ee_images WHERE object=?",
343            )
344            .bind(object.as_slice())
345            .fetch_one(&mut *conn)
346            .await?;
347            valid(admitted)?;
348            let bytes: Option<Vec<u8>> = sqlx::query_scalar(
349                "SELECT bytes FROM server_e2ee_image_chunks WHERE object=? AND chunk_index=?",
350            )
351            .bind(object.as_slice())
352            .bind(*index as i64)
353            .fetch_optional(conn)
354            .await?;
355            Ok(match bytes {
356                Some(b) => {
357                    d.verify_chunk(*index, &b)?;
358                    Reply::Chunk(b)
359                }
360                None => Reply::Unavailable,
361            })
362        }
363        Operation::Put { reservation, .. }
364        | Operation::Complete { reservation, .. }
365        | Operation::Release { reservation, .. } => {
366            ticket(
367                conn,
368                &object,
369                workspace,
370                &context.device,
371                &Ticket {
372                    reservation: *reservation,
373                },
374                now,
375            )
376            .await?;
377            match &op {
378                Operation::Put { index, record, .. } => {
379                    d.verify_chunk(*index, record)?;
380                    let old:Option<Vec<u8>>=sqlx::query_scalar("SELECT bytes FROM server_e2ee_image_chunks WHERE object=? AND chunk_index=?").bind(object.as_slice()).bind(*index as i64).fetch_optional(&mut *conn).await?;
381                    valid(old.as_ref().is_none_or(|b| b == record))?;
382                    sqlx::query("INSERT OR IGNORE INTO server_e2ee_image_chunks(object,chunk_index,bytes) VALUES(?,?,?)").bind(object.as_slice()).bind(*index as i64).bind(record).execute(&mut *conn).await?;
383                }
384                Operation::Complete { .. } => {
385                    let records:Vec<Vec<u8>>=sqlx::query_scalar("SELECT bytes FROM server_e2ee_image_chunks WHERE object=? ORDER BY chunk_index").bind(object.as_slice()).fetch_all(&mut *conn).await?;
386                    d.verify(&records)?;
387                    sqlx::query("UPDATE server_e2ee_images SET complete=1 WHERE object=?")
388                        .bind(object.as_slice())
389                        .execute(&mut *conn)
390                        .await?;
391                }
392                Operation::Release { .. } => {
393                    sqlx::query("DELETE FROM server_e2ee_image_tickets WHERE reservation=?")
394                        .bind(reservation.as_slice())
395                        .execute(&mut *conn)
396                        .await?;
397                }
398                _ => unreachable!(),
399            }
400            Ok(Reply::Done)
401        }
402        _ => unreachable!(),
403    }
404}
405
406pub(in crate::sync::encrypted_tail) async fn admit(
407    conn: &mut SqliteConnection,
408    context: &Context,
409    membership: &Membership,
410    id: &str,
411    p: &super::super::domain::Projection,
412    t: Option<&Ticket>,
413) -> Result<()> {
414    use super::super::domain::Projection;
415    let now = chrono::Utc::now().timestamp();
416    match p {
417        Projection::Move {
418            source,
419            target,
420            tasks,
421        } => {
422            valid(t.is_none())?;
423            super::super::moves::admit(conn, source, target, tasks).await?;
424        }
425        Projection::Ref {
426            workspace,
427            task,
428            reference,
429            descriptor,
430            deleted,
431            version,
432        } => {
433            let authored_workspace = workspace;
434            let routed = super::super::moves::server_workspace(conn, task, workspace).await?;
435            let workspace = &routed;
436            let d = Descriptor::decode(descriptor)?;
437            eligible_object(conn, membership, &d).await?;
438            let (_, complete) = load(conn, &d.object, &hash(descriptor)).await?;
439            valid(complete && d.vault == context.vault && d.stream == context.stream)?;
440            let records: Vec<Vec<u8>> = sqlx::query_scalar(
441                "SELECT bytes FROM server_e2ee_image_chunks WHERE object=? ORDER BY chunk_index",
442            )
443            .bind(d.object.as_slice())
444            .fetch_all(&mut *conn)
445            .await?;
446            d.verify(&records)?;
447            let exists:bool=sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_image_references WHERE workspace=? AND reference=?)").bind(workspace).bind(reference).fetch_one(&mut *conn).await?;
448            valid(!exists)?;
449            if !protected(conn, &d.object, workspace).await? {
450                ticket(
451                    conn,
452                    &d.object,
453                    authored_workspace,
454                    &context.device,
455                    t.context("error encrypted-image-reservation-required")?,
456                    now,
457                )
458                .await?;
459            }
460            let count: i64 =
461                sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_references")
462                    .fetch_one(&mut *conn)
463                    .await?;
464            valid(count < 262144)?;
465            let state:Option<(Option<String>,bool,bool)>=sqlx::query_as("SELECT version,deleted,protected FROM server_e2ee_image_parents WHERE workspace=? AND parent=?").bind(workspace).bind(task).fetch_optional(&mut *conn).await?;
466            if state.is_none() {
467                let count: i64 =
468                    sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_parents")
469                        .fetch_one(&mut *conn)
470                        .await?;
471                valid(count < 262144)?;
472            }
473            let (known, state_deleted, sticky) = state.unwrap_or((None, false, false));
474            let mut state = crate::sync::persistence::parent_liveness::ParentState {
475                version: known.clone(),
476                deleted: state_deleted,
477                protected: sticky,
478            };
479            state.protect_hint(*deleted, version.as_deref());
480            sqlx::query("INSERT INTO server_e2ee_image_parents(workspace,parent,version,deleted,protected) VALUES(?,?,?,?,?) ON CONFLICT(workspace,parent) DO UPDATE SET protected=excluded.protected").bind(workspace).bind(task).bind(known).bind(state_deleted).bind(state.protected).execute(&mut *conn).await?;
481            sqlx::query("INSERT INTO server_e2ee_image_references(workspace,reference,parent,deleted,object) VALUES(?,?,?,0,?)").bind(workspace).bind(reference).bind(task).bind(d.object.as_slice()).execute(&mut *conn).await?;
482            sqlx::query("UPDATE server_e2ee_images SET origin=COALESCE(origin,?) WHERE object=?")
483                .bind(id)
484                .bind(d.object.as_slice())
485                .execute(&mut *conn)
486                .await?;
487            if let Some(t) = t {
488                sqlx::query("DELETE FROM server_e2ee_image_tickets WHERE reservation=? AND device=? AND workspace=? AND object=?").bind(t.reservation.as_slice()).bind(context.device.as_slice()).bind(authored_workspace).bind(d.object.as_slice()).execute(&mut *conn).await?;
489            }
490            sqlx::query(
491                "INSERT OR IGNORE INTO server_e2ee_image_scopes(object, workspace) VALUES (?, ?)",
492            )
493            .bind(d.object.as_slice())
494            .bind(workspace)
495            .execute(&mut *conn)
496            .await?;
497            let scopes: i64 = sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_scopes")
498                .fetch_one(&mut *conn)
499                .await?;
500            valid(scopes <= 262144)?;
501            retain_parent(conn, workspace, task).await?;
502        }
503        Projection::Unref {
504            workspace,
505            task,
506            reference,
507        } => {
508            let workspace = super::super::moves::server_workspace(conn, task, workspace).await?;
509            let changed=sqlx::query("UPDATE server_e2ee_image_references SET deleted=1 WHERE workspace=? AND reference=? AND parent=?").bind(&workspace).bind(reference).bind(task).execute(&mut *conn).await?.rows_affected();
510            valid(changed == 1)?;
511        }
512        _ => valid(t.is_none())?,
513    }
514    Ok(())
515}
516
517#[cfg(test)]
518mod tests {
519    use super::*;
520
521    #[tokio::test]
522    async fn invalid_workspaces_refuse_authenticated_declare_and_existing_dispatch() {
523        use crate::ids::InvalidWorkspaceId;
524        use crate::sync::seed_claim::membership::test_support::Fixture;
525        use aven_protocol::artifact::catalog::{Artifact, Chunk};
526
527        let f = Fixture::new().await;
528        let membership = Membership::from_publication(
529            f.seed.genesis(),
530            &f.package.descriptor,
531            f.publication.record(),
532        )
533        .unwrap();
534        let binding = membership.publication().binding();
535        let context = Context {
536            vault: binding.vault_id,
537            genesis: binding.genesis_commitment,
538            device: f.seed.genesis().device_id(),
539            head: membership.head(),
540            stream: binding.stream_id,
541            descriptor: binding.descriptor_commitment,
542        };
543        // A structurally valid, not-yet-uploaded object recipe.
544        let descriptor = Descriptor {
545            vault: context.vault,
546            stream: context.stream,
547            generation: membership.current_generation().id,
548            object: [44; 32],
549            artifact: Artifact {
550                total: 1,
551                aggregate: [45; 32],
552                chunks: vec![Chunk {
553                    length: 223,
554                    hash: [46; 32],
555                    nonce: [47; 24],
556                }],
557            },
558        }
559        .encode()
560        .unwrap();
561        for op in [
562            Operation::Declare {
563                workspace: "invalid".into(),
564                descriptor: descriptor.clone(),
565            },
566            Operation::Status {
567                workspace: "0123456789ABCDEI".into(),
568                object: [44; 32],
569                descriptor_commitment: hash(&descriptor),
570            },
571        ] {
572            let error = f
573                .db
574                .encrypted_image_exchange(&context, f.seed.bearer(), op, LifecyclePolicy::default())
575                .await
576                .err()
577                .unwrap();
578            assert!(error.is::<InvalidWorkspaceId>());
579            assert_eq!(
580                error.to_string(),
581                "workspace ID must be 16 Crockford Base32 characters"
582            );
583        }
584        let mut conn = f.db.acquire_reader().await.unwrap();
585        let counts: (i64, i64, i64) = sqlx::query_as(
586            "SELECT (SELECT count(*) FROM server_e2ee_images), (SELECT count(*) FROM server_e2ee_image_scopes), (SELECT count(*) FROM server_e2ee_image_tickets)",
587        ).fetch_one(&mut *conn).await.unwrap();
588        assert_eq!(counts, (0, 0, 0));
589    }
590
591    #[tokio::test]
592    async fn retired_declaration_cannot_bypass_active_object_limit() {
593        let root = tempfile::tempdir().unwrap();
594        let db = Database::open(&root.path().join("test.sqlite"))
595            .await
596            .unwrap();
597        let a = crate::sync::encrypted_tail::tests::authority();
598        let mut conn = db.acquire_writer().await.unwrap();
599        let mut tx = begin_immediate(&mut conn).await.unwrap();
600        let (retired, _) = Descriptor::seal(&a, b"x").unwrap();
601        sqlx::query("INSERT INTO server_e2ee_images(object,byte_size,descriptor) VALUES(?,?,?)")
602            .bind(retired.object.as_slice())
603            .bind(retired.byte_size())
604            .bind(retired.encode().unwrap())
605            .execute(&mut *tx)
606            .await
607            .unwrap();
608        // These are unadmitted staging objects, not fabricated accepted references.
609        for _ in 0..128 {
610            let (d, records) = Descriptor::seal(&a, b"x").unwrap();
611            sqlx::query(
612                "INSERT INTO server_e2ee_images(object,byte_size,descriptor) VALUES(?,?,?)",
613            )
614            .bind(d.object.as_slice())
615            .bind(d.byte_size())
616            .bind(d.encode().unwrap())
617            .execute(&mut *tx)
618            .await
619            .unwrap();
620            sqlx::query(
621                "INSERT INTO server_e2ee_image_chunks(object,chunk_index,bytes) VALUES(?,0,?)",
622            )
623            .bind(d.object.as_slice())
624            .bind(&records[0])
625            .execute(&mut *tx)
626            .await
627            .unwrap();
628        }
629        assert!(
630            reserve(
631                &mut tx,
632                &retired,
633                "0000000000000000",
634                &a.context.device,
635                i64::MAX,
636                123
637            )
638            .await
639            .is_err()
640        );
641        let tickets: i64 = sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_tickets")
642            .fetch_one(&mut *tx)
643            .await
644            .unwrap();
645        assert_eq!(tickets, 0);
646    }
647}
aven_core/sync/encrypted_tail/attachments/server.rs

aven_core/sync/encrypted_tail/attachments/
server.rs
