559 lines
19 KiB
Rust
559 lines
19 KiB
Rust
//! SQLite persistence (relay.md §3). No sensitive data in the clear: the
|
|
//! `ciphertext` blobs are E2E, the pubkeys are public identifiers. The API is
|
|
//! designed to be swappable for Postgres+Redis post-v1 (relay.md §3 "scale
|
|
//! path"); for now there is a single writer (the SQLite-on-EFS constraint).
|
|
|
|
use std::collections::HashSet;
|
|
use std::str::FromStr;
|
|
use std::time::Duration;
|
|
|
|
use anyhow::Result;
|
|
use sqlx::Row;
|
|
use sqlx::sqlite::{SqliteConnectOptions, SqlitePool, SqlitePoolOptions};
|
|
|
|
use crate::auth::ct_eq;
|
|
|
|
/// Current unix milliseconds (application timestamp encoding, index.md §5).
|
|
pub fn now_ms() -> i64 {
|
|
chrono::Utc::now().timestamp_millis()
|
|
}
|
|
|
|
fn to_arr<const N: usize>(v: &[u8]) -> Option<[u8; N]> {
|
|
if v.len() != N {
|
|
return None;
|
|
}
|
|
let mut out = [0u8; N];
|
|
out.copy_from_slice(v);
|
|
Some(out)
|
|
}
|
|
|
|
/// Persisted client state.
|
|
#[derive(Debug, Clone)]
|
|
pub struct ClientRow {
|
|
pub x25519_pub: [u8; 32],
|
|
pub device_token: Option<String>,
|
|
pub platform: String,
|
|
pub state: String, // 'pending' | 'authorized'
|
|
}
|
|
|
|
/// A store-and-forward queued message.
|
|
#[derive(Debug, Clone)]
|
|
pub struct QueuedMsg {
|
|
pub id: i64,
|
|
pub from_pub: [u8; 32],
|
|
pub nonce: [u8; 12],
|
|
pub ciphertext: Vec<u8>,
|
|
pub created_at: i64,
|
|
}
|
|
|
|
/// A client in `pending` state (for re-sending `client_paired` to the agent).
|
|
#[derive(Debug, Clone)]
|
|
pub struct PendingClient {
|
|
pub ed25519_pub: [u8; 32],
|
|
pub x25519_pub: [u8; 32],
|
|
pub platform: String,
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
pub struct Store {
|
|
pool: SqlitePool,
|
|
}
|
|
|
|
impl Store {
|
|
/// Open/create the DB and apply the schema (idempotent). No WAL: the deploy
|
|
/// EFS/NFS does not support it; `busy_timeout` serializes the single writer.
|
|
pub async fn init(path: &str) -> Result<Store> {
|
|
let opts = SqliteConnectOptions::from_str(&format!("sqlite://{path}"))?
|
|
.create_if_missing(true)
|
|
.busy_timeout(Duration::from_secs(5))
|
|
.foreign_keys(true);
|
|
let pool = SqlitePoolOptions::new()
|
|
.max_connections(4)
|
|
.connect_with(opts)
|
|
.await?;
|
|
|
|
for stmt in SCHEMA {
|
|
// SCHEMA entries are 'static string literals (audited, no user data).
|
|
sqlx::query(*stmt).execute(&pool).await?;
|
|
}
|
|
Ok(Store { pool })
|
|
}
|
|
|
|
// ----- namespaces ---------------------------------------------------------
|
|
|
|
/// Create the namespace if absent (binding it immutably to the pubkey) and
|
|
/// bump `last_active`. Idempotent.
|
|
pub async fn upsert_namespace(&self, ns: &str, agent_pub: &[u8; 32]) -> Result<()> {
|
|
let now = now_ms();
|
|
sqlx::query(
|
|
"INSERT INTO namespaces (namespace_id, agent_ed25519_pub, created_at, last_active)
|
|
VALUES (?1, ?2, ?3, ?3)
|
|
ON CONFLICT(namespace_id) DO UPDATE SET last_active = ?3",
|
|
)
|
|
.bind(ns)
|
|
.bind(&agent_pub[..])
|
|
.bind(now)
|
|
.execute(&self.pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// `true` if the namespace exists.
|
|
pub async fn namespace_exists(&self, ns: &str) -> Result<bool> {
|
|
let row = sqlx::query("SELECT 1 FROM namespaces WHERE namespace_id = ?1")
|
|
.bind(ns)
|
|
.fetch_optional(&self.pool)
|
|
.await?;
|
|
Ok(row.is_some())
|
|
}
|
|
|
|
/// The namespace agent's ed25519 pubkey (None if it does not exist).
|
|
pub async fn agent_pub(&self, ns: &str) -> Result<Option<[u8; 32]>> {
|
|
let row = sqlx::query("SELECT agent_ed25519_pub FROM namespaces WHERE namespace_id = ?1")
|
|
.bind(ns)
|
|
.fetch_optional(&self.pool)
|
|
.await?;
|
|
Ok(row.and_then(|r| {
|
|
let b: Vec<u8> = r.get(0);
|
|
to_arr::<32>(&b)
|
|
}))
|
|
}
|
|
|
|
pub async fn touch_namespace(&self, ns: &str) -> Result<()> {
|
|
sqlx::query("UPDATE namespaces SET last_active = ?2 WHERE namespace_id = ?1")
|
|
.bind(ns)
|
|
.bind(now_ms())
|
|
.execute(&self.pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
// ----- pairing ------------------------------------------------------------
|
|
|
|
/// Open/replace the pairing window. `expiry_ms` already computed.
|
|
pub async fn pairing_start(&self, ns: &str, token: &[u8; 32], expiry_ms: i64) -> Result<()> {
|
|
sqlx::query(
|
|
"UPDATE namespaces
|
|
SET pairing_token = ?2, pairing_expiry = ?3, pairing_consumed = 0
|
|
WHERE namespace_id = ?1",
|
|
)
|
|
.bind(ns)
|
|
.bind(&token[..])
|
|
.bind(expiry_ms)
|
|
.execute(&self.pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn pairing_stop(&self, ns: &str) -> Result<()> {
|
|
sqlx::query(
|
|
"UPDATE namespaces
|
|
SET pairing_token = NULL, pairing_expiry = NULL, pairing_consumed = 0
|
|
WHERE namespace_id = ?1",
|
|
)
|
|
.bind(ns)
|
|
.execute(&self.pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Try to consume the pairing token (single-use, constant-time, not
|
|
/// expired). Returns `true` if pairing is allowed to proceed.
|
|
pub async fn consume_pairing_token(&self, ns: &str, token: &[u8; 32]) -> Result<bool> {
|
|
let row = sqlx::query(
|
|
"SELECT pairing_token, pairing_expiry, pairing_consumed
|
|
FROM namespaces WHERE namespace_id = ?1",
|
|
)
|
|
.bind(ns)
|
|
.fetch_optional(&self.pool)
|
|
.await?;
|
|
|
|
let Some(row) = row else { return Ok(false) };
|
|
let stored: Option<Vec<u8>> = row.get(0);
|
|
let expiry: Option<i64> = row.get(1);
|
|
let consumed: i64 = row.get(2);
|
|
|
|
let (Some(stored), Some(expiry)) = (stored, expiry) else {
|
|
return Ok(false); // no open window
|
|
};
|
|
if consumed != 0 || expiry <= now_ms() || !ct_eq(&stored, &token[..]) {
|
|
return Ok(false);
|
|
}
|
|
|
|
// Atomic guard against concurrent double-consume.
|
|
let res = sqlx::query(
|
|
"UPDATE namespaces SET pairing_consumed = 1
|
|
WHERE namespace_id = ?1 AND pairing_consumed = 0",
|
|
)
|
|
.bind(ns)
|
|
.execute(&self.pool)
|
|
.await?;
|
|
Ok(res.rows_affected() == 1)
|
|
}
|
|
|
|
// ----- clients ------------------------------------------------------------
|
|
|
|
/// Register/update a client as `pending` (after a successful pairing).
|
|
pub async fn upsert_pending_client(
|
|
&self,
|
|
ns: &str,
|
|
ed_pub: &[u8; 32],
|
|
x_pub: &[u8; 32],
|
|
device_token: &str,
|
|
platform: &str,
|
|
) -> Result<()> {
|
|
sqlx::query(
|
|
"INSERT INTO clients
|
|
(namespace_id, client_ed25519_pub, client_x25519_pub, device_token, platform, state, last_seen)
|
|
VALUES (?1, ?2, ?3, ?4, ?5, 'pending', ?6)
|
|
ON CONFLICT(namespace_id, client_ed25519_pub) DO UPDATE SET
|
|
client_x25519_pub = ?3,
|
|
device_token = CASE WHEN ?4 = '' THEN clients.device_token ELSE ?4 END,
|
|
platform = ?5, state = 'pending', last_seen = ?6",
|
|
)
|
|
.bind(ns)
|
|
.bind(&ed_pub[..])
|
|
.bind(&x_pub[..])
|
|
.bind(device_token)
|
|
.bind(platform)
|
|
.bind(now_ms())
|
|
.execute(&self.pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn get_client(&self, ns: &str, ed_pub: &[u8; 32]) -> Result<Option<ClientRow>> {
|
|
let row = sqlx::query(
|
|
"SELECT client_x25519_pub, device_token, platform, state
|
|
FROM clients WHERE namespace_id = ?1 AND client_ed25519_pub = ?2",
|
|
)
|
|
.bind(ns)
|
|
.bind(&ed_pub[..])
|
|
.fetch_optional(&self.pool)
|
|
.await?;
|
|
Ok(row.and_then(|r| {
|
|
let x: Vec<u8> = r.get(0);
|
|
Some(ClientRow {
|
|
x25519_pub: to_arr::<32>(&x)?,
|
|
device_token: r.get::<Option<String>, _>(1),
|
|
platform: r.get(2),
|
|
state: r.get(3),
|
|
})
|
|
}))
|
|
}
|
|
|
|
pub async fn is_authorized_client(&self, ns: &str, ed_pub: &[u8; 32]) -> Result<bool> {
|
|
let row = sqlx::query(
|
|
"SELECT 1 FROM clients
|
|
WHERE namespace_id = ?1 AND client_ed25519_pub = ?2 AND state = 'authorized'",
|
|
)
|
|
.bind(ns)
|
|
.bind(&ed_pub[..])
|
|
.fetch_optional(&self.pool)
|
|
.await?;
|
|
Ok(row.is_some())
|
|
}
|
|
|
|
/// Update the client's push token (APNs/FCM rotate it) + last_seen.
|
|
///
|
|
/// An **empty** `device_token` is treated as "no token available right now"
|
|
/// (e.g. the device connected before its APNs registration completed) and
|
|
/// must NOT clobber a previously stored, valid token — otherwise every push
|
|
/// to that client fails with `MissingDeviceToken`. In that case we still
|
|
/// bump `last_seen` but keep the existing token.
|
|
pub async fn update_client_device_token(
|
|
&self,
|
|
ns: &str,
|
|
ed_pub: &[u8; 32],
|
|
device_token: &str,
|
|
) -> Result<()> {
|
|
sqlx::query(
|
|
"UPDATE clients
|
|
SET device_token = CASE WHEN ?3 = '' THEN device_token ELSE ?3 END,
|
|
last_seen = ?4
|
|
WHERE namespace_id = ?1 AND client_ed25519_pub = ?2",
|
|
)
|
|
.bind(ns)
|
|
.bind(&ed_pub[..])
|
|
.bind(device_token)
|
|
.bind(now_ms())
|
|
.execute(&self.pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn list_pending_clients(&self, ns: &str) -> Result<Vec<PendingClient>> {
|
|
let rows = sqlx::query(
|
|
"SELECT client_ed25519_pub, client_x25519_pub, platform
|
|
FROM clients WHERE namespace_id = ?1 AND state = 'pending'",
|
|
)
|
|
.bind(ns)
|
|
.fetch_all(&self.pool)
|
|
.await?;
|
|
let mut out = Vec::new();
|
|
for r in rows {
|
|
let ed: Vec<u8> = r.get(0);
|
|
let x: Vec<u8> = r.get(1);
|
|
if let (Some(ed), Some(x)) = (to_arr::<32>(&ed), to_arr::<32>(&x)) {
|
|
out.push(PendingClient {
|
|
ed25519_pub: ed,
|
|
x25519_pub: x,
|
|
platform: r.get(2),
|
|
});
|
|
}
|
|
}
|
|
Ok(out)
|
|
}
|
|
|
|
/// Apply `authorize` (replace semantics, relay-protocol.md §6). Returns
|
|
/// `(authorized count, revoked pubkeys)`. Revoked clients must then be
|
|
/// disconnected; their queue has already been purged here.
|
|
pub async fn apply_authorize(
|
|
&self,
|
|
ns: &str,
|
|
new_list: &[[u8; 32]],
|
|
) -> Result<(i64, Vec<[u8; 32]>)> {
|
|
let new_set: HashSet<Vec<u8>> = new_list.iter().map(|k| k.to_vec()).collect();
|
|
|
|
let existing =
|
|
sqlx::query("SELECT client_ed25519_pub FROM clients WHERE namespace_id = ?1")
|
|
.bind(ns)
|
|
.fetch_all(&self.pool)
|
|
.await?;
|
|
|
|
let mut revoked = Vec::new();
|
|
for r in existing {
|
|
let pub_bytes: Vec<u8> = r.get(0);
|
|
if new_set.contains(&pub_bytes) {
|
|
// Present in the new list → authorized (leaves pending).
|
|
sqlx::query(
|
|
"UPDATE clients SET state = 'authorized'
|
|
WHERE namespace_id = ?1 AND client_ed25519_pub = ?2",
|
|
)
|
|
.bind(ns)
|
|
.bind(&pub_bytes)
|
|
.execute(&self.pool)
|
|
.await?;
|
|
} else {
|
|
// Absent → revoked: purge queue, forget device_token, remove.
|
|
self.purge_queue_for_bytes(ns, &pub_bytes).await?;
|
|
sqlx::query(
|
|
"DELETE FROM clients WHERE namespace_id = ?1 AND client_ed25519_pub = ?2",
|
|
)
|
|
.bind(ns)
|
|
.bind(&pub_bytes)
|
|
.execute(&self.pool)
|
|
.await?;
|
|
if let Some(k) = to_arr::<32>(&pub_bytes) {
|
|
revoked.push(k);
|
|
}
|
|
}
|
|
}
|
|
|
|
let count: i64 = sqlx::query(
|
|
"SELECT COUNT(*) FROM clients WHERE namespace_id = ?1 AND state = 'authorized'",
|
|
)
|
|
.bind(ns)
|
|
.fetch_one(&self.pool)
|
|
.await?
|
|
.get(0);
|
|
|
|
Ok((count, revoked))
|
|
}
|
|
|
|
// ----- queue (store-and-forward) -----------------------------------------
|
|
|
|
pub async fn queue_count(&self, ns: &str, to_pub: &[u8; 32]) -> Result<i64> {
|
|
let n: i64 =
|
|
sqlx::query("SELECT COUNT(*) FROM queue WHERE namespace_id = ?1 AND to_pub = ?2")
|
|
.bind(ns)
|
|
.bind(&to_pub[..])
|
|
.fetch_one(&self.pool)
|
|
.await?
|
|
.get(0);
|
|
Ok(n)
|
|
}
|
|
|
|
/// Enqueue a message. `Ok(false)` if the recipient's queue is full.
|
|
pub async fn enqueue(
|
|
&self,
|
|
ns: &str,
|
|
to_pub: &[u8; 32],
|
|
from_pub: &[u8; 32],
|
|
nonce: &[u8; 12],
|
|
ciphertext: &[u8],
|
|
max_per_dest: i64,
|
|
) -> Result<bool> {
|
|
if self.queue_count(ns, to_pub).await? >= max_per_dest {
|
|
return Ok(false);
|
|
}
|
|
sqlx::query(
|
|
"INSERT INTO queue (namespace_id, to_pub, from_pub, nonce, ciphertext, created_at)
|
|
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
|
|
)
|
|
.bind(ns)
|
|
.bind(&to_pub[..])
|
|
.bind(&from_pub[..])
|
|
.bind(&nonce[..])
|
|
.bind(ciphertext)
|
|
.bind(now_ms())
|
|
.execute(&self.pool)
|
|
.await?;
|
|
Ok(true)
|
|
}
|
|
|
|
pub async fn fetch_pending(&self, ns: &str, to_pub: &[u8; 32]) -> Result<Vec<QueuedMsg>> {
|
|
let rows = sqlx::query(
|
|
"SELECT id, from_pub, nonce, ciphertext, created_at
|
|
FROM queue WHERE namespace_id = ?1 AND to_pub = ?2 ORDER BY id ASC",
|
|
)
|
|
.bind(ns)
|
|
.bind(&to_pub[..])
|
|
.fetch_all(&self.pool)
|
|
.await?;
|
|
let mut out = Vec::new();
|
|
for r in rows {
|
|
let from: Vec<u8> = r.get(1);
|
|
let nonce: Vec<u8> = r.get(2);
|
|
if let (Some(from), Some(nonce)) = (to_arr::<32>(&from), to_arr::<12>(&nonce)) {
|
|
out.push(QueuedMsg {
|
|
id: r.get(0),
|
|
from_pub: from,
|
|
nonce,
|
|
ciphertext: r.get(3),
|
|
created_at: r.get(4),
|
|
});
|
|
}
|
|
}
|
|
Ok(out)
|
|
}
|
|
|
|
pub async fn delete_pending(&self, id: i64) -> Result<()> {
|
|
sqlx::query("DELETE FROM queue WHERE id = ?1")
|
|
.bind(id)
|
|
.execute(&self.pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
async fn purge_queue_for_bytes(&self, ns: &str, to_pub: &[u8]) -> Result<()> {
|
|
sqlx::query("DELETE FROM queue WHERE namespace_id = ?1 AND to_pub = ?2")
|
|
.bind(ns)
|
|
.bind(to_pub)
|
|
.execute(&self.pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
// ----- garbage collection -------------------------------------------------
|
|
|
|
/// Delete messages older than `ttl_days` and namespaces idle for `ttl_days`
|
|
/// (cascade to clients + queue). Returns `(messages, namespaces)` removed.
|
|
pub async fn gc(&self, ttl_days: i64) -> Result<(u64, u64)> {
|
|
let cutoff = now_ms() - ttl_days * 24 * 60 * 60 * 1000;
|
|
let msgs = sqlx::query("DELETE FROM queue WHERE created_at < ?1")
|
|
.bind(cutoff)
|
|
.execute(&self.pool)
|
|
.await?
|
|
.rows_affected();
|
|
let namespaces = sqlx::query("DELETE FROM namespaces WHERE last_active < ?1")
|
|
.bind(cutoff)
|
|
.execute(&self.pool)
|
|
.await?
|
|
.rows_affected();
|
|
Ok((msgs, namespaces))
|
|
}
|
|
}
|
|
|
|
const SCHEMA: &[&str] = &[
|
|
"CREATE TABLE IF NOT EXISTS namespaces (
|
|
namespace_id TEXT PRIMARY KEY,
|
|
agent_ed25519_pub BLOB NOT NULL UNIQUE,
|
|
created_at INTEGER NOT NULL,
|
|
last_active INTEGER NOT NULL,
|
|
pairing_token BLOB,
|
|
pairing_expiry INTEGER,
|
|
pairing_consumed INTEGER NOT NULL DEFAULT 0
|
|
)",
|
|
"CREATE TABLE IF NOT EXISTS clients (
|
|
namespace_id TEXT NOT NULL REFERENCES namespaces(namespace_id) ON DELETE CASCADE,
|
|
client_ed25519_pub BLOB NOT NULL,
|
|
client_x25519_pub BLOB NOT NULL,
|
|
device_token TEXT,
|
|
platform TEXT NOT NULL,
|
|
state TEXT NOT NULL,
|
|
last_seen INTEGER,
|
|
PRIMARY KEY (namespace_id, client_ed25519_pub)
|
|
)",
|
|
"CREATE TABLE IF NOT EXISTS queue (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
namespace_id TEXT NOT NULL REFERENCES namespaces(namespace_id) ON DELETE CASCADE,
|
|
to_pub BLOB NOT NULL,
|
|
from_pub BLOB NOT NULL,
|
|
nonce BLOB NOT NULL,
|
|
ciphertext BLOB NOT NULL,
|
|
created_at INTEGER NOT NULL
|
|
)",
|
|
"CREATE INDEX IF NOT EXISTS idx_queue_dest ON queue(namespace_id, to_pub, id)",
|
|
];
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
|
|
async fn temp_store() -> Store {
|
|
static SEQ: AtomicU64 = AtomicU64::new(0);
|
|
let nanos = std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.unwrap()
|
|
.as_nanos();
|
|
let seq = SEQ.fetch_add(1, Ordering::Relaxed);
|
|
let path = std::env::temp_dir().join(format!(
|
|
"relay-store-ut-{nanos}-{}-{seq}.db",
|
|
std::process::id()
|
|
));
|
|
Store::init(&path.to_string_lossy()).await.expect("init store")
|
|
}
|
|
|
|
/// An empty `device_token` (device connected before APNs registration
|
|
/// finished) must NOT wipe a previously stored, valid token — otherwise
|
|
/// every push fails with `MissingDeviceToken`.
|
|
#[tokio::test]
|
|
async fn empty_device_token_does_not_clobber() {
|
|
let s = temp_store().await;
|
|
let ns = "a".repeat(64);
|
|
let ed = [1u8; 32];
|
|
let x = [2u8; 32];
|
|
s.upsert_namespace(&ns, &[9u8; 32]).await.unwrap();
|
|
|
|
// Pair with a real token.
|
|
s.upsert_pending_client(&ns, &ed, &x, "realtoken", "ios").await.unwrap();
|
|
assert_eq!(
|
|
s.get_client(&ns, &ed).await.unwrap().unwrap().device_token.as_deref(),
|
|
Some("realtoken")
|
|
);
|
|
|
|
// A later connect with an empty token keeps the existing one.
|
|
s.update_client_device_token(&ns, &ed, "").await.unwrap();
|
|
assert_eq!(
|
|
s.get_client(&ns, &ed).await.unwrap().unwrap().device_token.as_deref(),
|
|
Some("realtoken")
|
|
);
|
|
|
|
// A non-empty token still updates.
|
|
s.update_client_device_token(&ns, &ed, "rotated").await.unwrap();
|
|
assert_eq!(
|
|
s.get_client(&ns, &ed).await.unwrap().unwrap().device_token.as_deref(),
|
|
Some("rotated")
|
|
);
|
|
|
|
// Re-pairing with an empty token also preserves the stored one.
|
|
s.upsert_pending_client(&ns, &ed, &x, "", "ios").await.unwrap();
|
|
assert_eq!(
|
|
s.get_client(&ns, &ed).await.unwrap().unwrap().device_token.as_deref(),
|
|
Some("rotated")
|
|
);
|
|
}
|
|
}
|