aven_core/sync/seed_claim/peer/
persistence.rs

1use super::*;
2use crate::db::{self, Database, begin_immediate, installation::InstallationGuard};
3use anyhow::Context;
4use sqlx::SqliteConnection;
5use std::collections::HashMap;
6
7/// Explicit current-head request context. Historical outcomes cannot authenticate it.
8pub struct Authentication<'a> {
9    pub vault: [u8; 32],
10    pub genesis: [u8; 32],
11    pub device: [u8; 32],
12    pub head: [u8; 32],
13    pub bearer: &'a Secret,
14}
15
16impl Database {
17    pub async fn peer_target_preflight(&self) -> Result<String> {
18        let mut conn = self.acquire_writer().await?;
19        let mut tx = begin_immediate(&mut conn).await?;
20        fresh(&mut tx).await?;
21        let client = db::get_meta(&mut tx, "client_id")
22            .await?
23            .context("error enrollment-client-missing")?;
24        tx.commit().await?;
25        Ok(client)
26    }
27    /// Rechecks that this pinned peer join is unfinished: nothing installed,
28    /// associated or set up, and the domain still empty. The enrollment pin
29    /// and installation fence stay in place.
30    pub async fn peer_retry_preflight(
31        &self,
32        identity: [u8; 32],
33        client: &str,
34        guard: &InstallationGuard,
35    ) -> Result<()> {
36        check(self.file_identity() == Some(guard.identity()))?;
37        let mut conn = self.acquire_writer().await?;
38        let mut tx = begin_immediate(&mut conn).await?;
39        let pinned: 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(identity.as_slice()).bind(client).fetch_one(&mut *tx).await?;
40        ensure!(pinned, "error enrollment-pin");
41        let occupied: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_peer_snapshot_install) OR 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?;
42        ensure!(!occupied, "error enrollment-retry-unavailable");
43        crate::sync::shared_state::ensure_empty_domain(&mut tx).await?;
44        tx.commit().await?;
45        Ok(())
46    }
47    pub async fn enrollment_pin(&self) -> Result<Option<([u8; 32], String, String)>> {
48        let mut conn = self.acquire_reader().await?;
49        let row: Option<(Vec<u8>, String, String)> = sqlx::query_as(
50            "SELECT identity,client_id,role FROM local_peer_enrollment WHERE singleton=1",
51        )
52        .fetch_optional(&mut *conn)
53        .await?;
54        row.map(|(id, client, role)| {
55            Ok((
56                id.try_into()
57                    .map_err(|_| anyhow::anyhow!("error enrollment-pin"))?,
58                client,
59                role,
60            ))
61        })
62        .transpose()
63    }
64    pub async fn pin_enrollment(
65        &self,
66        identity: [u8; 32],
67        client: &str,
68        role: &str,
69        guard: &InstallationGuard,
70    ) -> Result<()> {
71        check(self.file_identity() == Some(guard.identity()))?;
72        let mut conn = self.acquire_writer().await?;
73        let mut tx = begin_immediate(&mut conn).await?;
74        let stored: Option<(Vec<u8>, String, String)> = sqlx::query_as(
75            "SELECT identity,client_id,role FROM local_peer_enrollment WHERE singleton=1",
76        )
77        .fetch_optional(&mut *tx)
78        .await?;
79        check(db::get_meta(&mut tx, "client_id").await?.as_deref() == Some(client))?;
80        if let Some((id, c, r)) = stored {
81            check(id == identity && c == client && r == role)?;
82        } else {
83            if role == "peer" {
84                fresh(&mut tx).await?;
85            } else {
86                ensure_adopted(&mut tx).await?;
87            }
88            sqlx::query("INSERT INTO local_peer_enrollment(singleton,identity,client_id,role) VALUES(1,?,?,?)").bind(identity.as_slice()).bind(client).bind(role).execute(&mut *tx).await?;
89        }
90        tx.commit().await?;
91        Ok(())
92    }
93    pub async fn adopted_enrollment_client(&self) -> Result<String> {
94        let mut conn = self.acquire_reader().await?;
95        ensure_adopted(&mut conn).await?;
96        db::get_meta(&mut conn, "client_id")
97            .await?
98            .context("error enrollment-client-missing")
99    }
100    pub async fn enrollment_artifact(&self, kind: &str) -> Result<Option<Vec<u8>>> {
101        let mut conn = self.acquire_reader().await?;
102        Ok(sqlx::query_scalar(
103            "SELECT commitment FROM local_peer_enrollment_artifacts WHERE kind=?",
104        )
105        .bind(kind)
106        .fetch_optional(&mut *conn)
107        .await?)
108    }
109    pub async fn enrollment_artifacts(&self) -> Result<HashMap<String, Vec<u8>>> {
110        let mut conn = self.acquire_reader().await?;
111        let rows: Vec<(String, Vec<u8>)> =
112            sqlx::query_as("SELECT kind, commitment FROM local_peer_enrollment_artifacts")
113                .fetch_all(&mut *conn)
114                .await?;
115        Ok(rows.into_iter().collect())
116    }
117    /// Monotonic, nonsecret marker for write-once enrollment state changes.
118    pub async fn enrollment_artifact_marker(&self) -> Result<i64> {
119        let mut conn = self.acquire_reader().await?;
120        Ok(
121            sqlx::query_scalar("SELECT count(*) FROM local_peer_enrollment_artifacts")
122                .fetch_one(&mut *conn)
123                .await?,
124        )
125    }
126    pub async fn pin_enrollment_artifact(
127        &self,
128        identity: [u8; 32],
129        kind: &str,
130        digest: [u8; 32],
131    ) -> Result<()> {
132        let mut conn = self.acquire_writer().await?;
133        let mut tx = begin_immediate(&mut conn).await?;
134        let valid: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_peer_enrollment WHERE identity=? AND client_id=(SELECT value FROM meta WHERE key='client_id'))").bind(identity.as_slice()).fetch_one(&mut *tx).await?;
135        check(valid)?;
136        sqlx::query(
137            "INSERT OR IGNORE INTO local_peer_enrollment_artifacts(kind,commitment) VALUES(?,?)",
138        )
139        .bind(kind)
140        .bind(digest.as_slice())
141        .execute(&mut *tx)
142        .await?;
143        let saved: Vec<u8> = sqlx::query_scalar(
144            "SELECT commitment FROM local_peer_enrollment_artifacts WHERE kind=?",
145        )
146        .bind(kind)
147        .fetch_one(&mut *tx)
148        .await?;
149        check(saved == digest)?;
150        tx.commit().await?;
151        Ok(())
152    }
153}
154async fn ensure_adopted(conn: &mut SqliteConnection) -> Result<()> {
155    let ready: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_seed_publication_intent WHERE state='adopted' AND association=(SELECT value FROM meta WHERE key='e2ee_association') AND association_generation=CAST((SELECT value FROM meta WHERE key='sync_generation') AS INTEGER))").fetch_one(conn).await?;
156    ensure!(ready, "error enrollment-inviter-not-adopted");
157    Ok(())
158}
159async fn fresh(conn: &mut SqliteConnection) -> Result<()> {
160    crate::sync::shared_state::ensure_empty_target(conn).await?;
161    let occupied: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_seed_genesis_pin) OR EXISTS(SELECT 1 FROM server_seed_claim) OR EXISTS(SELECT 1 FROM local_peer_enrollment) OR EXISTS(SELECT 1 FROM meta WHERE key IN ('sync_server_url','e2ee_association'))").fetch_one(conn).await?;
162    ensure!(!occupied, "error enrollment-target-not-fresh");
163    Ok(())
164}
165
166#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
167pub enum RegistrationStatus {
168    Open,
169    Expired,
170    Consumed,
171}
aven_core/sync/seed_claim/peer/persistence.rs

aven_core/sync/seed_claim/peer/
persistence.rs
