diff --git a/README.md b/README.md index 46432e5..2281e76 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,7 @@ Add bunshin as a flake input and import the module: } ``` -`services.bunshin` also takes `host`, `port`, `dataDir`, `maxEnvelope`, `quota`, `retentionDays`, `rateConnections` and `rateSends`; see `flake.nix` for defaults. The module renders a `systemd` unit that runs `bunshin serve` under `DynamicUser`; it does not generate a key. +`services.bunshin` also takes `host`, `port`, `dataDir`, `maxEnvelope`, `quota`, `requestsQuota`, `retentionDays`, `requestsRetentionDays`, `maxTokens`, `rateConnections`, `rateSends` and `rateTokens`; see `flake.nix` for defaults. The module renders a `systemd` unit that runs `bunshin serve` under `DynamicUser`; it does not generate a key. Before the first deploy, generate the server's static key once (from a dev shell or `nix run`) and place it at the configured `keyFile`: @@ -60,8 +60,8 @@ bunshin keygen --key server.key bunshin serve --key server.key --db mail.db --host 0.0.0.0 --port 1961 ``` -`serve` accepts `--max-envelope`, `--quota`, `--retention-days`, `--invite-token`, `--rate-connections` and `--rate-sends` to control size limits, per-mailbox quota, message retention, registration gating and abuse control. Run `bunshin serve --help` for defaults. +`serve` accepts `--max-envelope`, `--quota`, `--requests-quota`, `--retention-days`, `--requests-retention-days`, `--max-tokens`, `--invite-token`, `--rate-connections`, `--rate-sends` and `--rate-tokens` to control size limits, the mailbox's two quota tiers, their retention, the accept-token cap, registration gating and abuse control. Run `bunshin serve --help` for defaults. ## Status -Implements SPEC.md version 1 in full: `AUTH`, `RESOLVE`, `SEND`, `FETCH`, `DELETE` and `REGISTER` (including invite tokens and key rotation chains). Verified end-to-end against the Python reference client over a live Noise connection, plus a raw-protocol test suite covering the rejection paths (auth failures, unknown users, malformed and oversized envelopes, rate limiting). +Implements SPEC.md version 1.1 in full: `AUTH`, `RESOLVE`, `SEND`, `FETCH`, `DELETE` and `REGISTER`, including accept tokens and the main/requests tier split, fetch cursors, 32-byte message ids, `REGISTER` proof of possession, and dual-signed key rotation chains. diff --git a/flake.nix b/flake.nix index c4873a7..c8a13f6 100644 --- a/flake.nix +++ b/flake.nix @@ -76,20 +76,38 @@ maxEnvelope = mkOption { type = types.ints.positive; - default = 1048576; + default = 786432; description = "Maximum accepted envelope size, in bytes."; }; quota = mkOption { type = types.ints.positive; default = 67108864; - description = "Per-mailbox storage quota, in bytes."; + description = "Per-mailbox main-tier storage quota, in bytes."; + }; + + requestsQuota = mkOption { + type = types.ints.positive; + default = 2097152; + description = "Per-mailbox requests-tier storage quota, in bytes."; }; retentionDays = mkOption { type = types.ints.positive; default = 30; - description = "Days a message is retained before being purged."; + description = "Days a main-tier message is retained before being purged."; + }; + + requestsRetentionDays = mkOption { + type = types.ints.positive; + default = 7; + description = "Days a requests-tier message is retained before being purged."; + }; + + maxTokens = mkOption { + type = types.ints.positive; + default = 1024; + description = "Maximum accept tokens a mailbox may hold."; }; rateConnections = mkOption { @@ -104,6 +122,12 @@ description = "Max SEND operations per minute, per source IP."; }; + rateTokens = mkOption { + type = types.ints.positive; + default = 30; + description = "Max SEND operations per minute, per accept token."; + }; + inviteToken = mkOption { type = types.nullOr types.str; default = null; @@ -154,9 +178,13 @@ --port ${toString cfg.port} --max-envelope ${toString cfg.maxEnvelope} --quota ${toString cfg.quota} + --requests-quota ${toString cfg.requestsQuota} --retention-days ${toString cfg.retentionDays} + --requests-retention-days ${toString cfg.requestsRetentionDays} + --max-tokens ${toString cfg.maxTokens} --rate-connections ${toString cfg.rateConnections} --rate-sends ${toString cfg.rateSends} + --rate-tokens ${toString cfg.rateTokens} ) ${lib.optionalString (cfg.inviteToken != null) ''args+=(--invite-token ${lib.escapeShellArg cfg.inviteToken})''} diff --git a/src/crypto.rs b/src/crypto.rs index a54f1e6..b2981cd 100644 --- a/src/crypto.rs +++ b/src/crypto.rs @@ -7,7 +7,7 @@ use sha2::{Digest, Sha256}; use std::sync::LazyLock; use x25519_dalek::{PublicKey, StaticSecret}; -use crate::proto::{LABEL_ID, ID_LEN}; +use crate::proto::{ID_LEN, LABEL_ID}; static BASE32_LOWER_UNPADDED: LazyLock = LazyLock::new(|| { let mut spec = Specification::new(); @@ -60,6 +60,45 @@ pub fn verify(pubkey: &[u8], signature: &[u8], message: &[u8]) -> bool { verifying_key.verify_strict(message, &signature).is_ok() } +/// HMAC-SHA256 (RFC 2104), spelled out rather than pulled in as a dependency: +/// the construction is a handful of lines over the SHA-256 already in use. +pub fn hmac_sha256(key: &[u8], message: &[u8]) -> [u8; 32] { + const BLOCK: usize = 64; + let mut block = [0u8; BLOCK]; + if key.len() > BLOCK { + block[..32].copy_from_slice(&Sha256::digest(key)); + } else { + block[..key.len()].copy_from_slice(key); + } + let mut ipad = [0x36u8; BLOCK]; + let mut opad = [0x5cu8; BLOCK]; + for i in 0..BLOCK { + ipad[i] ^= block[i]; + opad[i] ^= block[i]; + } + let mut inner = Sha256::new(); + inner.update(ipad); + inner.update(message); + let inner = inner.finalize(); + + let mut outer = Sha256::new(); + outer.update(opad); + outer.update(inner); + outer.finalize().into() +} + +/// Constant-time equality, so a MAC comparison cannot leak timing information. +pub fn ct_eq(a: &[u8], b: &[u8]) -> bool { + if a.len() != b.len() { + return false; + } + let mut diff = 0u8; + for (x, y) in a.iter().zip(b.iter()) { + diff |= x ^ y; + } + diff == 0 +} + #[cfg(test)] mod tests { use super::*; @@ -71,7 +110,7 @@ mod tests { } #[test] - fn message_id_is_16_bytes_and_deterministic() { + fn message_id_is_32_bytes_and_deterministic() { let a = message_id(b"envelope-bytes"); let b = message_id(b"envelope-bytes"); let c = message_id(b"other-bytes"); @@ -85,4 +124,20 @@ mod tests { assert!(!verify(&[0u8; 32], &[0u8; 64], b"msg")); assert!(!verify(&[0u8; 5], &[0u8; 64], b"msg")); } + + #[test] + fn hmac_matches_known_test_vector() { + // RFC 4231 test case 1. + let key = [0x0bu8; 20]; + let data = b"Hi There"; + let expected = "b0344c61d8db38535ca8afceaf0bf12b881dc200c9833da726e9376c2e32cff7"; + assert_eq!(data_encoding::HEXLOWER.encode(&hmac_sha256(&key, data)), expected); + } + + #[test] + fn ct_eq_matches_equality() { + assert!(ct_eq(b"abc", b"abc")); + assert!(!ct_eq(b"abc", b"abd")); + assert!(!ct_eq(b"abc", b"ab")); + } } diff --git a/src/main.rs b/src/main.rs index e802ca8..5bccf32 100644 --- a/src/main.rs +++ b/src/main.rs @@ -45,18 +45,26 @@ enum Command { host: String, #[arg(long, default_value_t = DEFAULT_PORT)] port: u16, - #[arg(long = "max-envelope", default_value_t = 1 << 20)] + #[arg(long = "max-envelope", default_value_t = 768 << 10)] max_envelope: usize, #[arg(long, default_value_t = 64 << 20)] quota: i64, + #[arg(long = "requests-quota", default_value_t = 2 << 20)] + requests_quota: i64, #[arg(long = "retention-days", default_value_t = 30)] retention_days: i64, + #[arg(long = "requests-retention-days", default_value_t = 7)] + requests_retention_days: i64, + #[arg(long = "max-tokens", default_value_t = 1024)] + max_tokens: u16, #[arg(long = "invite-token")] invite_token: Option, #[arg(long = "rate-connections", default_value_t = 120)] rate_connections: u32, #[arg(long = "rate-sends", default_value_t = 60)] rate_sends: u32, + #[arg(long = "rate-tokens", default_value_t = 30)] + rate_tokens: u32, }, } @@ -81,10 +89,14 @@ fn main() -> anyhow::Result<()> { port, max_envelope, quota, + requests_quota, retention_days, + requests_retention_days, + max_tokens, invite_token, rate_connections, rate_sends, + rate_tokens, } => server::run(server::ServeArgs { key_path: key, db_path: db, @@ -92,10 +104,14 @@ fn main() -> anyhow::Result<()> { port, max_envelope, quota, + requests_quota, retention_days, + requests_retention_days, + max_tokens, invite_token, rate_connections, rate_sends, + rate_tokens, }), } } diff --git a/src/proto.rs b/src/proto.rs index 2c71666..1b52e1a 100644 --- a/src/proto.rs +++ b/src/proto.rs @@ -6,7 +6,9 @@ pub const NOISE_PARAMS: &str = "Noise_NX_25519_ChaChaPoly_SHA256"; pub const PROLOGUE: &[u8] = b"smolmail/1"; pub const LABEL_AUTH: &[u8] = b"smolmail/1 auth"; pub const LABEL_ID: &[u8] = b"smolmail/1 id"; +pub const LABEL_MAC: &[u8] = b"smolmail/1 mac"; pub const LABEL_ROTATE: &[u8] = b"smolmail/1 rotate"; +pub const LABEL_REGISTER: &[u8] = b"smolmail/1 register"; pub const OP_AUTH: u8 = 0x00; pub const OP_RESOLVE: u8 = 0x01; @@ -31,9 +33,11 @@ pub const ENVELOPE_MAGIC: &[u8; 4] = b"SMOL"; pub const ENVELOPE_VERSION: u8 = 1; pub const ENVELOPE_HEADER: usize = 69; // magic 4 + version 1 + to 32 + epk 32 pub const ENVELOPE_MIN: usize = ENVELOPE_HEADER + 16; // + Poly1305 tag -pub const ID_LEN: usize = 16; +pub const ID_LEN: usize = 32; pub const KEY_LEN: usize = 32; -pub const CERT_LEN: usize = 136; // old_pub 32 + new_pub 32 + time 8 + signature 64 +pub const TOKEN_LEN: usize = 32; +pub const MAC_LEN: usize = 32; +pub const CERT_LEN: usize = 200; // old_pub 32 + new_pub 32 + time 8 + sig_old 64 + sig_new 64 pub const MAX_CHAIN: usize = 16; pub const DEFAULT_PORT: u16 = 1961; @@ -98,6 +102,11 @@ impl<'a> Reader<'a> { Ok(u16::from_be_bytes([b[0], b[1]])) } + pub fn i64(&mut self) -> Result { + let b = self.take(8)?; + Ok(i64::from_be_bytes(b.try_into().unwrap())) + } + pub fn rest(&mut self) -> &'a [u8] { let out = &self.buf[self.pos..]; self.pos = self.buf.len(); diff --git a/src/server.rs b/src/server.rs index 4e9dcc6..31bc548 100644 --- a/src/server.rs +++ b/src/server.rs @@ -18,10 +18,14 @@ pub struct ServeArgs { pub port: u16, pub max_envelope: usize, pub quota: i64, + pub requests_quota: i64, pub retention_days: i64, + pub requests_retention_days: i64, + pub max_tokens: u16, pub invite_token: Option, pub rate_connections: u32, pub rate_sends: u32, + pub rate_tokens: u32, } pub fn run(args: ServeArgs) -> anyhow::Result<()> { @@ -33,24 +37,29 @@ pub fn run(args: ServeArgs) -> anyhow::Result<()> { static_key.len() ); + let key_array: [u8; KEY_LEN] = static_key.clone().try_into().unwrap(); + let server_static = derive_public(&key_array); + let config = Arc::new(ServerConfig { max_envelope: args.max_envelope, - quota: args.quota, + main_quota: args.quota, + requests_quota: args.requests_quota, + max_tokens: args.max_tokens, invite_token: args.invite_token.map(String::into_bytes), + server_static, conn_limiter: RateLimiter::new(args.rate_connections), send_limiter: RateLimiter::new(args.rate_sends), + token_limiter: RateLimiter::new(args.rate_tokens), }); - let retention_secs = args.retention_days * 86400; + let main_retention_secs = args.retention_days * 86400; + let requests_retention_secs = args.requests_retention_days * 86400; let purge_db_path = args.db_path.clone(); - std::thread::spawn(move || purge_loop(purge_db_path, retention_secs)); + std::thread::spawn(move || purge_loop(purge_db_path, main_retention_secs, requests_retention_secs)); let listener = std::net::TcpListener::bind((args.host.as_str(), args.port))?; log::info!("listening on {}:{}", args.host, args.port); - - let key_array: [u8; KEY_LEN] = static_key.clone().try_into().unwrap(); - let public = derive_public(&key_array); - log::info!("server public key: {}", b32(&public)); + log::info!("server public key: {}", b32(&server_static)); for incoming in listener.incoming() { let stream = match incoming { @@ -140,7 +149,7 @@ fn is_eof_like(e: &anyhow::Error) -> bool { false } -fn purge_loop(db_path: String, retention_secs: i64) { +fn purge_loop(db_path: String, main_retention_secs: i64, requests_retention_secs: i64) { let store = match Store::open(&db_path) { Ok(s) => s, Err(e) => { @@ -154,7 +163,7 @@ fn purge_loop(db_path: String, retention_secs: i64) { .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_secs() as i64; - match store.purge(now - retention_secs) { + match store.purge(now - main_retention_secs, now - requests_retention_secs) { Ok(0) => {} Ok(n) => log::info!("expired {n} message(s)"), Err(e) => log::error!("purge failed: {e}"), diff --git a/src/session.rs b/src/session.rs index 5d72b2d..6c81ec9 100644 --- a/src/session.rs +++ b/src/session.rs @@ -1,21 +1,26 @@ //! Per-connection dispatch and the six wire operations. -use crate::crypto::{message_id, verify}; +use crate::crypto::{b32, ct_eq, hmac_sha256, message_id, verify}; use crate::proto::{ valid_username, ProtocolError, Reader, AUTH_FAILED, AUTH_REQUIRED, BAD_VERSION, CERT_LEN, ENVELOPE_MAGIC, ENVELOPE_MIN, ENVELOPE_VERSION, FETCH_BUDGET, ID_LEN, KEY_LEN, LABEL_AUTH, - LABEL_ROTATE, MALFORMED, MAX_CHAIN, NOT_PERMITTED, OK, OP_AUTH, OP_DELETE, OP_FETCH, - OP_REGISTER, OP_RESOLVE, OP_SEND, QUOTA_EXCEEDED, RATE_LIMITED, TOO_LARGE, UNKNOWN_USER, + LABEL_MAC, LABEL_REGISTER, LABEL_ROTATE, MAC_LEN, MALFORMED, MAX_CHAIN, NOT_PERMITTED, OK, + OP_AUTH, OP_DELETE, OP_FETCH, OP_REGISTER, OP_RESOLVE, OP_SEND, QUOTA_EXCEEDED, RATE_LIMITED, + TOKEN_LEN, TOO_LARGE, UNKNOWN_USER, }; use crate::ratelimit::RateLimiter; use crate::store::Store; pub struct ServerConfig { pub max_envelope: usize, - pub quota: i64, + pub main_quota: i64, + pub requests_quota: i64, + pub max_tokens: u16, pub invite_token: Option>, + pub server_static: [u8; KEY_LEN], pub conn_limiter: RateLimiter, pub send_limiter: RateLimiter, + pub token_limiter: RateLimiter, } /// A parse failure (-> MALFORMED) or a storage failure (-> INTERNAL_ERROR). @@ -100,8 +105,19 @@ impl<'a> Session<'a> { let username = Self::read_str(r)?; let identity = r.take(KEY_LEN)?.to_vec(); let signature = r.take(64)?.to_vec(); + let sync = r.u8()?; + let count = r.u16()? as usize; + let mut tokens = Vec::with_capacity(count); + for _ in 0..count { + tokens.push(r.take(TOKEN_LEN)?.to_vec()); + } r.done()?; + // sync = 0 leaves the stored set untouched, so it carries no tokens. + if sync > 1 || (sync == 0 && count != 0) { + return Ok((MALFORMED, Vec::new())); + } + let bound = self.store.identity_of(&username)?; // A wrong username and a wrong signature are both AUTH_FAILED: telling // them apart would turn this into an account-existence oracle. @@ -113,8 +129,18 @@ impl<'a> Session<'a> { if !verify(&identity, &signature, &msg) { return Ok((AUTH_FAILED, Vec::new())); } + + if sync == 1 { + if count > self.config.max_tokens as usize { + // Leaves the session unauthenticated, per SPEC.md sec 4. + return Ok((TOO_LARGE, Vec::new())); + } + self.store.set_tokens(&username, &tokens)?; + } + + let accepted = self.store.token_count(&username)?; self.username = Some(username); - Ok((OK, Vec::new())) + Ok((OK, accepted.to_be_bytes().to_vec())) } fn op_resolve(&mut self, r: &mut Reader) -> OpResult { @@ -134,7 +160,13 @@ impl<'a> Session<'a> { } fn op_send(&mut self, r: &mut Reader) -> OpResult { + let mac_len = r.u8()? as usize; + if mac_len != 0 && mac_len != MAC_LEN { + return Ok((MALFORMED, Vec::new())); + } + let mac = r.take(mac_len)?.to_vec(); let envelope = r.rest().to_vec(); + if !self.config.send_limiter.allow(&self.peer_ip) { return Ok((RATE_LIMITED, Vec::new())); } @@ -152,28 +184,63 @@ impl<'a> Session<'a> { Some(u) => u, None => return Ok((UNKNOWN_USER, Vec::new())), }; - let keys = self.store.keys_of(&username)?; - let used = self.store.mailbox_bytes(&keys)?; - if used + envelope.len() as i64 > self.config.quota { - return Ok((QUOTA_EXCEEDED, Vec::new())); - } + // The ciphertext is never inspected; the server cannot read it. let mid = message_id(&envelope); - self.store.store_message(&mid, recipient, &envelope)?; + + // A matching accept token (SPEC.md sec 5.8) puts the envelope in the + // mailbox's main tier; anything else lands in the smaller, short-lived + // requests tier. A failed match is never reported to the sender. + let matched_token = if mac.len() == MAC_LEN { + let mut msg = LABEL_MAC.to_vec(); + msg.extend_from_slice(&mid); + self.store + .tokens_of(&username)? + .into_iter() + .find(|t| ct_eq(&hmac_sha256(t, &msg), &mac)) + } else { + None + }; + + if let Some(token) = &matched_token { + if !self.config.token_limiter.allow(&b32(token)) { + return Ok((RATE_LIMITED, Vec::new())); + } + } + + let unsolicited = matched_token.is_none(); + let tier_quota = if unsolicited { + self.config.requests_quota + } else { + self.config.main_quota + }; + let keys = self.store.keys_of(&username)?; + let used = self.store.mailbox_bytes(&keys, unsolicited)?; + if used + envelope.len() as i64 > tier_quota { + return Ok((QUOTA_EXCEEDED, Vec::new())); + } + + self.store + .store_message(&mid, recipient, &envelope, unsolicited)?; Ok((OK, mid.to_vec())) } fn op_fetch(&mut self, r: &mut Reader) -> OpResult { + let after_received_at = r.i64()?; + let after_id = r.take(ID_LEN)?.to_vec(); r.done()?; let username = self.username.as_ref().expect("AUTH_REQUIRED gate above"); let keys = self.store.keys_of(username)?; - let records = self.store.pending(&keys, FETCH_BUDGET)?; + let records = self + .store + .pending(&keys, after_received_at, &after_id, FETCH_BUDGET)?; let mut out = Vec::new(); out.extend_from_slice(&(records.len() as u16).to_be_bytes()); - for (mid, received_at, envelope) in records { + for (mid, received_at, unsolicited, envelope) in records { out.extend_from_slice(&mid); out.extend_from_slice(&received_at.to_be_bytes()); + out.push(if unsolicited { 1 } else { 0 }); out.extend_from_slice(&(envelope.len() as u32).to_be_bytes()); out.extend_from_slice(&envelope); } @@ -202,6 +269,7 @@ impl<'a> Session<'a> { fn op_register(&mut self, r: &mut Reader) -> OpResult { let username = Self::read_str(r)?; let identity = r.take(KEY_LEN)?.to_vec(); + let signature = r.take(64)?.to_vec(); let token_len = r.u8()? as usize; let token = r.take(token_len)?.to_vec(); let cert_len = r.u8()? as usize; @@ -214,6 +282,17 @@ impl<'a> Session<'a> { // identity is exactly KEY_LEN bytes by construction (Reader::take // enforces it); no separate curve-point validity check is needed. + // Proof of possession, required on every registration and rotation + // (SPEC.md sec 6.1). Binding server_static stops the attestation from + // being replayed against another server. + let mut pop_msg = LABEL_REGISTER.to_vec(); + pop_msg.extend_from_slice(&self.config.server_static); + pop_msg.extend_from_slice(username.as_bytes()); + pop_msg.extend_from_slice(&identity); + if !verify(&identity, &signature, &pop_msg) { + return Ok((AUTH_FAILED, Vec::new())); + } + if let Some(expected) = &self.config.invite_token { if &token != expected { return Ok((NOT_PERMITTED, Vec::new())); @@ -233,7 +312,8 @@ impl<'a> Session<'a> { let old_pub = &cert[..32]; let new_pub = &cert[32..64]; let when = &cert[64..72]; - let signature = &cert[72..]; + let sig_old = &cert[72..136]; + let sig_new = &cert[136..200]; if new_pub != identity.as_slice() { return Ok((MALFORMED, Vec::new())); } @@ -246,10 +326,11 @@ impl<'a> Session<'a> { return Ok((NOT_PERMITTED, Vec::new())); } let mut msg = LABEL_ROTATE.to_vec(); + msg.extend_from_slice(username.as_bytes()); msg.extend_from_slice(old_pub); msg.extend_from_slice(new_pub); msg.extend_from_slice(when); - if !verify(old_pub, signature, &msg) { + if !verify(old_pub, sig_old, &msg) || !verify(new_pub, sig_new, &msg) { return Ok((AUTH_FAILED, Vec::new())); } let chain = self.store.chain(&username)?; diff --git a/src/store.rs b/src/store.rs index c204bf0..b2d7b7b 100644 --- a/src/store.rs +++ b/src/store.rs @@ -24,14 +24,22 @@ CREATE TABLE IF NOT EXISTS rotations ( cert BLOB NOT NULL, PRIMARY KEY (username, seq) ); +-- The mailbox owner's accept tokens (SPEC.md sec 5.8): an allow-list the +-- server can match against but never interpret. +CREATE TABLE IF NOT EXISTS tokens ( + username TEXT NOT NULL, + token BLOB NOT NULL, + PRIMARY KEY (username, token) +); CREATE TABLE IF NOT EXISTS messages ( id BLOB PRIMARY KEY, recipient BLOB NOT NULL, received_at INTEGER NOT NULL, + unsolicited INTEGER NOT NULL, envelope BLOB NOT NULL ); CREATE INDEX IF NOT EXISTS messages_by_recipient - ON messages (recipient, received_at); + ON messages (recipient, received_at, id); "; pub struct Store { @@ -77,6 +85,37 @@ impl Store { rows.collect() } + pub fn tokens_of(&self, username: &str) -> rusqlite::Result>> { + let mut stmt = self + .conn + .prepare("SELECT token FROM tokens WHERE username = ?1")?; + let rows = stmt.query_map([username], |row| row.get(0))?; + rows.collect() + } + + pub fn token_count(&self, username: &str) -> rusqlite::Result { + self.conn.query_row( + "SELECT COUNT(*) FROM tokens WHERE username = ?1", + [username], + |row| row.get(0), + ) + } + + /// Replaces the mailbox's stored token set wholesale (`AUTH` with + /// `sync = 1`, SPEC.md sec 4): this is how a token is both added and + /// removed, since the server never sees which changed. + pub fn set_tokens(&self, username: &str, tokens: &[Vec]) -> rusqlite::Result<()> { + let tx = self.conn.unchecked_transaction()?; + tx.execute("DELETE FROM tokens WHERE username = ?1", [username])?; + for token in tokens { + tx.execute( + "INSERT INTO tokens (username, token) VALUES (?1, ?2)", + (username, token), + )?; + } + tx.commit() + } + pub fn username_for_key(&self, identity: &[u8]) -> rusqlite::Result> { self.conn .query_row( @@ -152,12 +191,16 @@ impl Store { } } - pub fn mailbox_bytes(&self, keys: &[Vec]) -> rusqlite::Result { + /// Bytes stored for `keys` in one tier: the requests tier (no matching + /// accept token, SPEC.md sec 5.8) or the main tier. + pub fn mailbox_bytes(&self, keys: &[Vec], unsolicited: bool) -> rusqlite::Result { if keys.is_empty() { return Ok(0); } let sql = format!( - "SELECT COALESCE(SUM(LENGTH(envelope)), 0) FROM messages WHERE recipient IN ({})", + "SELECT COALESCE(SUM(LENGTH(envelope)), 0) FROM messages \ + WHERE unsolicited = {} AND recipient IN ({})", + unsolicited as i64, placeholders(keys.len()) ); self.conn @@ -169,10 +212,11 @@ impl Store { mid: &[u8], recipient: &[u8], envelope: &[u8], + unsolicited: bool, ) -> rusqlite::Result<()> { self.conn.execute( - "INSERT OR IGNORE INTO messages (id, recipient, received_at, envelope) \ - VALUES (?1, ?2, ?3, ?4)", + "INSERT OR IGNORE INTO messages (id, recipient, received_at, unsolicited, envelope) \ + VALUES (?1, ?2, ?3, ?4, ?5)", ( mid, recipient, @@ -180,45 +224,58 @@ impl Store { .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_secs() as i64, + unsolicited as i64, envelope, ), )?; Ok(()) } - /// Always returns at least one message, even if it alone exceeds `budget`, - /// so an oversized envelope cannot wedge a mailbox shut. + /// Records strictly after `(after_received_at, after_id)`, oldest first. + /// Always returns at least one message, even if it alone exceeds + /// `budget`, so an oversized envelope cannot wedge a mailbox shut. pub fn pending( &self, keys: &[Vec], + after_received_at: i64, + after_id: &[u8], budget: usize, - ) -> rusqlite::Result, i64, Vec)>> { + ) -> rusqlite::Result, i64, bool, Vec)>> { if keys.is_empty() { return Ok(Vec::new()); } let sql = format!( - "SELECT id, received_at, envelope FROM messages \ - WHERE recipient IN ({}) ORDER BY received_at, id", + "SELECT id, received_at, unsolicited, envelope FROM messages \ + WHERE recipient IN ({}) AND (received_at, id) > (?, ?) \ + ORDER BY received_at, id", placeholders(keys.len()) ); + let mut params: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(keys.len() + 2); + for key in keys { + params.push(key); + } + params.push(&after_received_at); + params.push(&after_id); + let mut stmt = self.conn.prepare(&sql)?; - let rows = stmt.query_map(params_from_iter(keys.iter()), |row| { + let rows = stmt.query_map(params_from_iter(params), |row| { Ok(( row.get::<_, Vec>(0)?, row.get::<_, i64>(1)?, - row.get::<_, Vec>(2)?, + row.get::<_, i64>(2)? != 0, + row.get::<_, Vec>(3)?, )) })?; let mut out = Vec::new(); let mut used = 0usize; for row in rows { - let (mid, received_at, envelope) = row?; + let (mid, received_at, unsolicited, envelope) = row?; if !out.is_empty() && used + envelope.len() > budget { break; } used += envelope.len(); - out.push((mid, received_at, envelope)); + out.push((mid, received_at, unsolicited, envelope)); } Ok(out) } @@ -236,8 +293,12 @@ impl Store { self.conn.execute(&sql, params_from_iter(params)) } - pub fn purge(&self, older_than: i64) -> rusqlite::Result { - self.conn - .execute("DELETE FROM messages WHERE received_at < ?1", [older_than]) + /// Each tier has its own retention (SPEC.md sec 10). + pub fn purge(&self, main_older_than: i64, requests_older_than: i64) -> rusqlite::Result { + self.conn.execute( + "DELETE FROM messages WHERE \ + (unsolicited = 0 AND received_at < ?1) OR (unsolicited = 1 AND received_at < ?2)", + (main_older_than, requests_older_than), + ) } }