feat: update server to Smol Mail protocol 1.1
This commit is contained in:
parent
71ad7e04c0
commit
24fe701ca2
8 changed files with 311 additions and 52 deletions
|
|
@ -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`:
|
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
|
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
|
## 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.
|
||||||
|
|
|
||||||
34
flake.nix
34
flake.nix
|
|
@ -76,20 +76,38 @@
|
||||||
|
|
||||||
maxEnvelope = mkOption {
|
maxEnvelope = mkOption {
|
||||||
type = types.ints.positive;
|
type = types.ints.positive;
|
||||||
default = 1048576;
|
default = 786432;
|
||||||
description = "Maximum accepted envelope size, in bytes.";
|
description = "Maximum accepted envelope size, in bytes.";
|
||||||
};
|
};
|
||||||
|
|
||||||
quota = mkOption {
|
quota = mkOption {
|
||||||
type = types.ints.positive;
|
type = types.ints.positive;
|
||||||
default = 67108864;
|
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 {
|
retentionDays = mkOption {
|
||||||
type = types.ints.positive;
|
type = types.ints.positive;
|
||||||
default = 30;
|
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 {
|
rateConnections = mkOption {
|
||||||
|
|
@ -104,6 +122,12 @@
|
||||||
description = "Max SEND operations per minute, per source IP.";
|
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 {
|
inviteToken = mkOption {
|
||||||
type = types.nullOr types.str;
|
type = types.nullOr types.str;
|
||||||
default = null;
|
default = null;
|
||||||
|
|
@ -154,9 +178,13 @@
|
||||||
--port ${toString cfg.port}
|
--port ${toString cfg.port}
|
||||||
--max-envelope ${toString cfg.maxEnvelope}
|
--max-envelope ${toString cfg.maxEnvelope}
|
||||||
--quota ${toString cfg.quota}
|
--quota ${toString cfg.quota}
|
||||||
|
--requests-quota ${toString cfg.requestsQuota}
|
||||||
--retention-days ${toString cfg.retentionDays}
|
--retention-days ${toString cfg.retentionDays}
|
||||||
|
--requests-retention-days ${toString cfg.requestsRetentionDays}
|
||||||
|
--max-tokens ${toString cfg.maxTokens}
|
||||||
--rate-connections ${toString cfg.rateConnections}
|
--rate-connections ${toString cfg.rateConnections}
|
||||||
--rate-sends ${toString cfg.rateSends}
|
--rate-sends ${toString cfg.rateSends}
|
||||||
|
--rate-tokens ${toString cfg.rateTokens}
|
||||||
)
|
)
|
||||||
${lib.optionalString (cfg.inviteToken != null)
|
${lib.optionalString (cfg.inviteToken != null)
|
||||||
''args+=(--invite-token ${lib.escapeShellArg cfg.inviteToken})''}
|
''args+=(--invite-token ${lib.escapeShellArg cfg.inviteToken})''}
|
||||||
|
|
|
||||||
|
|
@ -7,7 +7,7 @@ use sha2::{Digest, Sha256};
|
||||||
use std::sync::LazyLock;
|
use std::sync::LazyLock;
|
||||||
use x25519_dalek::{PublicKey, StaticSecret};
|
use x25519_dalek::{PublicKey, StaticSecret};
|
||||||
|
|
||||||
use crate::proto::{LABEL_ID, ID_LEN};
|
use crate::proto::{ID_LEN, LABEL_ID};
|
||||||
|
|
||||||
static BASE32_LOWER_UNPADDED: LazyLock<Encoding> = LazyLock::new(|| {
|
static BASE32_LOWER_UNPADDED: LazyLock<Encoding> = LazyLock::new(|| {
|
||||||
let mut spec = Specification::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()
|
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)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
@ -71,7 +110,7 @@ mod tests {
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[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 a = message_id(b"envelope-bytes");
|
||||||
let b = message_id(b"envelope-bytes");
|
let b = message_id(b"envelope-bytes");
|
||||||
let c = message_id(b"other-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; 32], &[0u8; 64], b"msg"));
|
||||||
assert!(!verify(&[0u8; 5], &[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"));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
18
src/main.rs
18
src/main.rs
|
|
@ -45,18 +45,26 @@ enum Command {
|
||||||
host: String,
|
host: String,
|
||||||
#[arg(long, default_value_t = DEFAULT_PORT)]
|
#[arg(long, default_value_t = DEFAULT_PORT)]
|
||||||
port: u16,
|
port: u16,
|
||||||
#[arg(long = "max-envelope", default_value_t = 1 << 20)]
|
#[arg(long = "max-envelope", default_value_t = 768 << 10)]
|
||||||
max_envelope: usize,
|
max_envelope: usize,
|
||||||
#[arg(long, default_value_t = 64 << 20)]
|
#[arg(long, default_value_t = 64 << 20)]
|
||||||
quota: i64,
|
quota: i64,
|
||||||
|
#[arg(long = "requests-quota", default_value_t = 2 << 20)]
|
||||||
|
requests_quota: i64,
|
||||||
#[arg(long = "retention-days", default_value_t = 30)]
|
#[arg(long = "retention-days", default_value_t = 30)]
|
||||||
retention_days: i64,
|
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")]
|
#[arg(long = "invite-token")]
|
||||||
invite_token: Option<String>,
|
invite_token: Option<String>,
|
||||||
#[arg(long = "rate-connections", default_value_t = 120)]
|
#[arg(long = "rate-connections", default_value_t = 120)]
|
||||||
rate_connections: u32,
|
rate_connections: u32,
|
||||||
#[arg(long = "rate-sends", default_value_t = 60)]
|
#[arg(long = "rate-sends", default_value_t = 60)]
|
||||||
rate_sends: u32,
|
rate_sends: u32,
|
||||||
|
#[arg(long = "rate-tokens", default_value_t = 30)]
|
||||||
|
rate_tokens: u32,
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -81,10 +89,14 @@ fn main() -> anyhow::Result<()> {
|
||||||
port,
|
port,
|
||||||
max_envelope,
|
max_envelope,
|
||||||
quota,
|
quota,
|
||||||
|
requests_quota,
|
||||||
retention_days,
|
retention_days,
|
||||||
|
requests_retention_days,
|
||||||
|
max_tokens,
|
||||||
invite_token,
|
invite_token,
|
||||||
rate_connections,
|
rate_connections,
|
||||||
rate_sends,
|
rate_sends,
|
||||||
|
rate_tokens,
|
||||||
} => server::run(server::ServeArgs {
|
} => server::run(server::ServeArgs {
|
||||||
key_path: key,
|
key_path: key,
|
||||||
db_path: db,
|
db_path: db,
|
||||||
|
|
@ -92,10 +104,14 @@ fn main() -> anyhow::Result<()> {
|
||||||
port,
|
port,
|
||||||
max_envelope,
|
max_envelope,
|
||||||
quota,
|
quota,
|
||||||
|
requests_quota,
|
||||||
retention_days,
|
retention_days,
|
||||||
|
requests_retention_days,
|
||||||
|
max_tokens,
|
||||||
invite_token,
|
invite_token,
|
||||||
rate_connections,
|
rate_connections,
|
||||||
rate_sends,
|
rate_sends,
|
||||||
|
rate_tokens,
|
||||||
}),
|
}),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
13
src/proto.rs
13
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 PROLOGUE: &[u8] = b"smolmail/1";
|
||||||
pub const LABEL_AUTH: &[u8] = b"smolmail/1 auth";
|
pub const LABEL_AUTH: &[u8] = b"smolmail/1 auth";
|
||||||
pub const LABEL_ID: &[u8] = b"smolmail/1 id";
|
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_ROTATE: &[u8] = b"smolmail/1 rotate";
|
||||||
|
pub const LABEL_REGISTER: &[u8] = b"smolmail/1 register";
|
||||||
|
|
||||||
pub const OP_AUTH: u8 = 0x00;
|
pub const OP_AUTH: u8 = 0x00;
|
||||||
pub const OP_RESOLVE: u8 = 0x01;
|
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_VERSION: u8 = 1;
|
||||||
pub const ENVELOPE_HEADER: usize = 69; // magic 4 + version 1 + to 32 + epk 32
|
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 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 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 MAX_CHAIN: usize = 16;
|
||||||
|
|
||||||
pub const DEFAULT_PORT: u16 = 1961;
|
pub const DEFAULT_PORT: u16 = 1961;
|
||||||
|
|
@ -98,6 +102,11 @@ impl<'a> Reader<'a> {
|
||||||
Ok(u16::from_be_bytes([b[0], b[1]]))
|
Ok(u16::from_be_bytes([b[0], b[1]]))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn i64(&mut self) -> Result<i64, ProtocolError> {
|
||||||
|
let b = self.take(8)?;
|
||||||
|
Ok(i64::from_be_bytes(b.try_into().unwrap()))
|
||||||
|
}
|
||||||
|
|
||||||
pub fn rest(&mut self) -> &'a [u8] {
|
pub fn rest(&mut self) -> &'a [u8] {
|
||||||
let out = &self.buf[self.pos..];
|
let out = &self.buf[self.pos..];
|
||||||
self.pos = self.buf.len();
|
self.pos = self.buf.len();
|
||||||
|
|
|
||||||
|
|
@ -18,10 +18,14 @@ pub struct ServeArgs {
|
||||||
pub port: u16,
|
pub port: u16,
|
||||||
pub max_envelope: usize,
|
pub max_envelope: usize,
|
||||||
pub quota: i64,
|
pub quota: i64,
|
||||||
|
pub requests_quota: i64,
|
||||||
pub retention_days: i64,
|
pub retention_days: i64,
|
||||||
|
pub requests_retention_days: i64,
|
||||||
|
pub max_tokens: u16,
|
||||||
pub invite_token: Option<String>,
|
pub invite_token: Option<String>,
|
||||||
pub rate_connections: u32,
|
pub rate_connections: u32,
|
||||||
pub rate_sends: u32,
|
pub rate_sends: u32,
|
||||||
|
pub rate_tokens: u32,
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn run(args: ServeArgs) -> anyhow::Result<()> {
|
pub fn run(args: ServeArgs) -> anyhow::Result<()> {
|
||||||
|
|
@ -33,24 +37,29 @@ pub fn run(args: ServeArgs) -> anyhow::Result<()> {
|
||||||
static_key.len()
|
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 {
|
let config = Arc::new(ServerConfig {
|
||||||
max_envelope: args.max_envelope,
|
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),
|
invite_token: args.invite_token.map(String::into_bytes),
|
||||||
|
server_static,
|
||||||
conn_limiter: RateLimiter::new(args.rate_connections),
|
conn_limiter: RateLimiter::new(args.rate_connections),
|
||||||
send_limiter: RateLimiter::new(args.rate_sends),
|
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();
|
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))?;
|
let listener = std::net::TcpListener::bind((args.host.as_str(), args.port))?;
|
||||||
log::info!("listening on {}:{}", args.host, args.port);
|
log::info!("listening on {}:{}", args.host, args.port);
|
||||||
|
log::info!("server public key: {}", b32(&server_static));
|
||||||
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));
|
|
||||||
|
|
||||||
for incoming in listener.incoming() {
|
for incoming in listener.incoming() {
|
||||||
let stream = match incoming {
|
let stream = match incoming {
|
||||||
|
|
@ -140,7 +149,7 @@ fn is_eof_like(e: &anyhow::Error) -> bool {
|
||||||
false
|
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) {
|
let store = match Store::open(&db_path) {
|
||||||
Ok(s) => s,
|
Ok(s) => s,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
|
|
@ -154,7 +163,7 @@ fn purge_loop(db_path: String, retention_secs: i64) {
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
.unwrap()
|
.unwrap()
|
||||||
.as_secs() as i64;
|
.as_secs() as i64;
|
||||||
match store.purge(now - retention_secs) {
|
match store.purge(now - main_retention_secs, now - requests_retention_secs) {
|
||||||
Ok(0) => {}
|
Ok(0) => {}
|
||||||
Ok(n) => log::info!("expired {n} message(s)"),
|
Ok(n) => log::info!("expired {n} message(s)"),
|
||||||
Err(e) => log::error!("purge failed: {e}"),
|
Err(e) => log::error!("purge failed: {e}"),
|
||||||
|
|
|
||||||
111
src/session.rs
111
src/session.rs
|
|
@ -1,21 +1,26 @@
|
||||||
//! Per-connection dispatch and the six wire operations.
|
//! 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::{
|
use crate::proto::{
|
||||||
valid_username, ProtocolError, Reader, AUTH_FAILED, AUTH_REQUIRED, BAD_VERSION, CERT_LEN,
|
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,
|
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,
|
LABEL_MAC, LABEL_REGISTER, LABEL_ROTATE, MAC_LEN, MALFORMED, MAX_CHAIN, NOT_PERMITTED, OK,
|
||||||
OP_REGISTER, OP_RESOLVE, OP_SEND, QUOTA_EXCEEDED, RATE_LIMITED, TOO_LARGE, UNKNOWN_USER,
|
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::ratelimit::RateLimiter;
|
||||||
use crate::store::Store;
|
use crate::store::Store;
|
||||||
|
|
||||||
pub struct ServerConfig {
|
pub struct ServerConfig {
|
||||||
pub max_envelope: usize,
|
pub max_envelope: usize,
|
||||||
pub quota: i64,
|
pub main_quota: i64,
|
||||||
|
pub requests_quota: i64,
|
||||||
|
pub max_tokens: u16,
|
||||||
pub invite_token: Option<Vec<u8>>,
|
pub invite_token: Option<Vec<u8>>,
|
||||||
|
pub server_static: [u8; KEY_LEN],
|
||||||
pub conn_limiter: RateLimiter,
|
pub conn_limiter: RateLimiter,
|
||||||
pub send_limiter: RateLimiter,
|
pub send_limiter: RateLimiter,
|
||||||
|
pub token_limiter: RateLimiter,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A parse failure (-> MALFORMED) or a storage failure (-> INTERNAL_ERROR).
|
/// 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 username = Self::read_str(r)?;
|
||||||
let identity = r.take(KEY_LEN)?.to_vec();
|
let identity = r.take(KEY_LEN)?.to_vec();
|
||||||
let signature = r.take(64)?.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()?;
|
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)?;
|
let bound = self.store.identity_of(&username)?;
|
||||||
// A wrong username and a wrong signature are both AUTH_FAILED: telling
|
// A wrong username and a wrong signature are both AUTH_FAILED: telling
|
||||||
// them apart would turn this into an account-existence oracle.
|
// them apart would turn this into an account-existence oracle.
|
||||||
|
|
@ -113,8 +129,18 @@ impl<'a> Session<'a> {
|
||||||
if !verify(&identity, &signature, &msg) {
|
if !verify(&identity, &signature, &msg) {
|
||||||
return Ok((AUTH_FAILED, Vec::new()));
|
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);
|
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 {
|
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 {
|
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();
|
let envelope = r.rest().to_vec();
|
||||||
|
|
||||||
if !self.config.send_limiter.allow(&self.peer_ip) {
|
if !self.config.send_limiter.allow(&self.peer_ip) {
|
||||||
return Ok((RATE_LIMITED, Vec::new()));
|
return Ok((RATE_LIMITED, Vec::new()));
|
||||||
}
|
}
|
||||||
|
|
@ -152,28 +184,63 @@ impl<'a> Session<'a> {
|
||||||
Some(u) => u,
|
Some(u) => u,
|
||||||
None => return Ok((UNKNOWN_USER, Vec::new())),
|
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.
|
// The ciphertext is never inspected; the server cannot read it.
|
||||||
let mid = message_id(&envelope);
|
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()))
|
Ok((OK, mid.to_vec()))
|
||||||
}
|
}
|
||||||
|
|
||||||
fn op_fetch(&mut self, r: &mut Reader) -> OpResult {
|
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()?;
|
r.done()?;
|
||||||
let username = self.username.as_ref().expect("AUTH_REQUIRED gate above");
|
let username = self.username.as_ref().expect("AUTH_REQUIRED gate above");
|
||||||
let keys = self.store.keys_of(username)?;
|
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();
|
let mut out = Vec::new();
|
||||||
out.extend_from_slice(&(records.len() as u16).to_be_bytes());
|
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(&mid);
|
||||||
out.extend_from_slice(&received_at.to_be_bytes());
|
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.len() as u32).to_be_bytes());
|
||||||
out.extend_from_slice(&envelope);
|
out.extend_from_slice(&envelope);
|
||||||
}
|
}
|
||||||
|
|
@ -202,6 +269,7 @@ impl<'a> Session<'a> {
|
||||||
fn op_register(&mut self, r: &mut Reader) -> OpResult {
|
fn op_register(&mut self, r: &mut Reader) -> OpResult {
|
||||||
let username = Self::read_str(r)?;
|
let username = Self::read_str(r)?;
|
||||||
let identity = r.take(KEY_LEN)?.to_vec();
|
let identity = r.take(KEY_LEN)?.to_vec();
|
||||||
|
let signature = r.take(64)?.to_vec();
|
||||||
let token_len = r.u8()? as usize;
|
let token_len = r.u8()? as usize;
|
||||||
let token = r.take(token_len)?.to_vec();
|
let token = r.take(token_len)?.to_vec();
|
||||||
let cert_len = r.u8()? as usize;
|
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
|
// identity is exactly KEY_LEN bytes by construction (Reader::take
|
||||||
// enforces it); no separate curve-point validity check is needed.
|
// 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 let Some(expected) = &self.config.invite_token {
|
||||||
if &token != expected {
|
if &token != expected {
|
||||||
return Ok((NOT_PERMITTED, Vec::new()));
|
return Ok((NOT_PERMITTED, Vec::new()));
|
||||||
|
|
@ -233,7 +312,8 @@ impl<'a> Session<'a> {
|
||||||
let old_pub = &cert[..32];
|
let old_pub = &cert[..32];
|
||||||
let new_pub = &cert[32..64];
|
let new_pub = &cert[32..64];
|
||||||
let when = &cert[64..72];
|
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() {
|
if new_pub != identity.as_slice() {
|
||||||
return Ok((MALFORMED, Vec::new()));
|
return Ok((MALFORMED, Vec::new()));
|
||||||
}
|
}
|
||||||
|
|
@ -246,10 +326,11 @@ impl<'a> Session<'a> {
|
||||||
return Ok((NOT_PERMITTED, Vec::new()));
|
return Ok((NOT_PERMITTED, Vec::new()));
|
||||||
}
|
}
|
||||||
let mut msg = LABEL_ROTATE.to_vec();
|
let mut msg = LABEL_ROTATE.to_vec();
|
||||||
|
msg.extend_from_slice(username.as_bytes());
|
||||||
msg.extend_from_slice(old_pub);
|
msg.extend_from_slice(old_pub);
|
||||||
msg.extend_from_slice(new_pub);
|
msg.extend_from_slice(new_pub);
|
||||||
msg.extend_from_slice(when);
|
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()));
|
return Ok((AUTH_FAILED, Vec::new()));
|
||||||
}
|
}
|
||||||
let chain = self.store.chain(&username)?;
|
let chain = self.store.chain(&username)?;
|
||||||
|
|
|
||||||
95
src/store.rs
95
src/store.rs
|
|
@ -24,14 +24,22 @@ CREATE TABLE IF NOT EXISTS rotations (
|
||||||
cert BLOB NOT NULL,
|
cert BLOB NOT NULL,
|
||||||
PRIMARY KEY (username, seq)
|
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 (
|
CREATE TABLE IF NOT EXISTS messages (
|
||||||
id BLOB PRIMARY KEY,
|
id BLOB PRIMARY KEY,
|
||||||
recipient BLOB NOT NULL,
|
recipient BLOB NOT NULL,
|
||||||
received_at INTEGER NOT NULL,
|
received_at INTEGER NOT NULL,
|
||||||
|
unsolicited INTEGER NOT NULL,
|
||||||
envelope BLOB NOT NULL
|
envelope BLOB NOT NULL
|
||||||
);
|
);
|
||||||
CREATE INDEX IF NOT EXISTS messages_by_recipient
|
CREATE INDEX IF NOT EXISTS messages_by_recipient
|
||||||
ON messages (recipient, received_at);
|
ON messages (recipient, received_at, id);
|
||||||
";
|
";
|
||||||
|
|
||||||
pub struct Store {
|
pub struct Store {
|
||||||
|
|
@ -77,6 +85,37 @@ impl Store {
|
||||||
rows.collect()
|
rows.collect()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn tokens_of(&self, username: &str) -> rusqlite::Result<Vec<Vec<u8>>> {
|
||||||
|
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<u16> {
|
||||||
|
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<u8>]) -> 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<Option<String>> {
|
pub fn username_for_key(&self, identity: &[u8]) -> rusqlite::Result<Option<String>> {
|
||||||
self.conn
|
self.conn
|
||||||
.query_row(
|
.query_row(
|
||||||
|
|
@ -152,12 +191,16 @@ impl Store {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn mailbox_bytes(&self, keys: &[Vec<u8>]) -> rusqlite::Result<i64> {
|
/// 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<u8>], unsolicited: bool) -> rusqlite::Result<i64> {
|
||||||
if keys.is_empty() {
|
if keys.is_empty() {
|
||||||
return Ok(0);
|
return Ok(0);
|
||||||
}
|
}
|
||||||
let sql = format!(
|
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())
|
placeholders(keys.len())
|
||||||
);
|
);
|
||||||
self.conn
|
self.conn
|
||||||
|
|
@ -169,10 +212,11 @@ impl Store {
|
||||||
mid: &[u8],
|
mid: &[u8],
|
||||||
recipient: &[u8],
|
recipient: &[u8],
|
||||||
envelope: &[u8],
|
envelope: &[u8],
|
||||||
|
unsolicited: bool,
|
||||||
) -> rusqlite::Result<()> {
|
) -> rusqlite::Result<()> {
|
||||||
self.conn.execute(
|
self.conn.execute(
|
||||||
"INSERT OR IGNORE INTO messages (id, recipient, received_at, envelope) \
|
"INSERT OR IGNORE INTO messages (id, recipient, received_at, unsolicited, envelope) \
|
||||||
VALUES (?1, ?2, ?3, ?4)",
|
VALUES (?1, ?2, ?3, ?4, ?5)",
|
||||||
(
|
(
|
||||||
mid,
|
mid,
|
||||||
recipient,
|
recipient,
|
||||||
|
|
@ -180,45 +224,58 @@ impl Store {
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
.unwrap()
|
.unwrap()
|
||||||
.as_secs() as i64,
|
.as_secs() as i64,
|
||||||
|
unsolicited as i64,
|
||||||
envelope,
|
envelope,
|
||||||
),
|
),
|
||||||
)?;
|
)?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Always returns at least one message, even if it alone exceeds `budget`,
|
/// Records strictly after `(after_received_at, after_id)`, oldest first.
|
||||||
/// so an oversized envelope cannot wedge a mailbox shut.
|
/// Always returns at least one message, even if it alone exceeds
|
||||||
|
/// `budget`, so an oversized envelope cannot wedge a mailbox shut.
|
||||||
pub fn pending(
|
pub fn pending(
|
||||||
&self,
|
&self,
|
||||||
keys: &[Vec<u8>],
|
keys: &[Vec<u8>],
|
||||||
|
after_received_at: i64,
|
||||||
|
after_id: &[u8],
|
||||||
budget: usize,
|
budget: usize,
|
||||||
) -> rusqlite::Result<Vec<(Vec<u8>, i64, Vec<u8>)>> {
|
) -> rusqlite::Result<Vec<(Vec<u8>, i64, bool, Vec<u8>)>> {
|
||||||
if keys.is_empty() {
|
if keys.is_empty() {
|
||||||
return Ok(Vec::new());
|
return Ok(Vec::new());
|
||||||
}
|
}
|
||||||
let sql = format!(
|
let sql = format!(
|
||||||
"SELECT id, received_at, envelope FROM messages \
|
"SELECT id, received_at, unsolicited, envelope FROM messages \
|
||||||
WHERE recipient IN ({}) ORDER BY received_at, id",
|
WHERE recipient IN ({}) AND (received_at, id) > (?, ?) \
|
||||||
|
ORDER BY received_at, id",
|
||||||
placeholders(keys.len())
|
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 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((
|
Ok((
|
||||||
row.get::<_, Vec<u8>>(0)?,
|
row.get::<_, Vec<u8>>(0)?,
|
||||||
row.get::<_, i64>(1)?,
|
row.get::<_, i64>(1)?,
|
||||||
row.get::<_, Vec<u8>>(2)?,
|
row.get::<_, i64>(2)? != 0,
|
||||||
|
row.get::<_, Vec<u8>>(3)?,
|
||||||
))
|
))
|
||||||
})?;
|
})?;
|
||||||
|
|
||||||
let mut out = Vec::new();
|
let mut out = Vec::new();
|
||||||
let mut used = 0usize;
|
let mut used = 0usize;
|
||||||
for row in rows {
|
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 {
|
if !out.is_empty() && used + envelope.len() > budget {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
used += envelope.len();
|
used += envelope.len();
|
||||||
out.push((mid, received_at, envelope));
|
out.push((mid, received_at, unsolicited, envelope));
|
||||||
}
|
}
|
||||||
Ok(out)
|
Ok(out)
|
||||||
}
|
}
|
||||||
|
|
@ -236,8 +293,12 @@ impl Store {
|
||||||
self.conn.execute(&sql, params_from_iter(params))
|
self.conn.execute(&sql, params_from_iter(params))
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn purge(&self, older_than: i64) -> rusqlite::Result<usize> {
|
/// Each tier has its own retention (SPEC.md sec 10).
|
||||||
self.conn
|
pub fn purge(&self, main_older_than: i64, requests_older_than: i64) -> rusqlite::Result<usize> {
|
||||||
.execute("DELETE FROM messages WHERE received_at < ?1", [older_than])
|
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),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue