aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/bayou.rs123
-rw-r--r--src/cryptoblob.rs58
-rw-r--r--src/main.rs47
-rw-r--r--src/time.rs9
-rw-r--r--src/uidindex.rs168
5 files changed, 405 insertions, 0 deletions
diff --git a/src/bayou.rs b/src/bayou.rs
new file mode 100644
index 0000000..8fc711e
--- /dev/null
+++ b/src/bayou.rs
@@ -0,0 +1,123 @@
+use std::time::Duration;
+
+use anyhow::Result;
+use rand::prelude::*;
+use serde::{Deserialize, Serialize};
+
+use k2v_client::K2vClient;
+use rusoto_core::HttpClient;
+use rusoto_credential::{AwsCredentials, StaticProvider};
+use rusoto_s3::S3Client;
+use rusoto_signature::Region;
+
+use crate::cryptoblob::Key;
+use crate::time::now_msec;
+
+pub trait BayouState:
+ Default + Clone + Serialize + for<'de> Deserialize<'de> + Send + Sync + 'static
+{
+ type Op: Clone + Serialize + for<'de> Deserialize<'de> + Send + Sync + 'static;
+
+ fn apply(&self, op: &Self::Op) -> Self;
+}
+
+pub struct Bayou<S: BayouState> {
+ bucket: String,
+ path: String,
+ key: Key,
+
+ k2v: K2vClient,
+ s3: S3Client,
+
+ checkpoint: (Timestamp, S),
+ history: Vec<(Timestamp, S::Op, Option<S>)>,
+}
+
+impl<S: BayouState> Bayou<S> {
+ pub fn new(
+ creds: AwsCredentials,
+ k2v_region: Region,
+ s3_region: Region,
+ bucket: String,
+ path: String,
+ key: Key,
+ ) -> Result<Self> {
+ let k2v_client = K2vClient::new(k2v_region, bucket.clone(), creds.clone(), None)?;
+ let static_creds = StaticProvider::new(
+ creds.aws_access_key_id().to_string(),
+ creds.aws_secret_access_key().to_string(),
+ creds.token().clone(),
+ None,
+ );
+ let s3_client = S3Client::new_with(HttpClient::new()?, static_creds, s3_region);
+
+ Ok(Self {
+ bucket,
+ path,
+ key,
+ k2v: k2v_client,
+ s3: s3_client,
+ checkpoint: (Timestamp::zero(), S::default()),
+ history: vec![],
+ })
+ }
+
+ pub async fn sync(&mut self) -> Result<()> {
+ // 1. List checkpoints
+ // 2. Load last checkpoint
+ // 3. List all operations starting from checkpoint
+ // 4. Check that first operation has same timestamp as checkpoint (if not zero)
+ // 5. Apply all operations in order
+ unimplemented!()
+ }
+
+ pub fn state(&self) -> &S {
+ unimplemented!()
+ }
+}
+
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub struct Timestamp {
+ pub msec: u64,
+ pub rand: u64,
+}
+
+impl Timestamp {
+ pub fn now() -> Self {
+ let mut rng = thread_rng();
+ Self {
+ msec: now_msec(),
+ rand: rng.gen::<u64>(),
+ }
+ }
+
+ pub fn after(other: &Self) -> Self {
+ let mut rng = thread_rng();
+ Self {
+ msec: std::cmp::max(now_msec(), other.msec + 1),
+ rand: rng.gen::<u64>(),
+ }
+ }
+
+ pub fn zero() -> Self {
+ Self { msec: 0, rand: 0 }
+ }
+
+ pub fn serialize(&self) -> String {
+ let mut bytes = [0u8; 16];
+ bytes[0..8].copy_from_slice(&u64::to_be_bytes(self.msec));
+ bytes[8..16].copy_from_slice(&u64::to_be_bytes(self.rand));
+ hex::encode(&bytes)
+ }
+
+ pub fn parse(v: &str) -> Option<Self> {
+ let bytes = hex::decode(v).ok()?;
+ if bytes.len() != 16 {
+ return None;
+ }
+ Some(Self {
+ msec: u64::from_be_bytes(bytes[0..8].try_into().unwrap()),
+ rand: u64::from_be_bytes(bytes[8..16].try_into().unwrap()),
+ })
+ }
+}
diff --git a/src/cryptoblob.rs b/src/cryptoblob.rs
new file mode 100644
index 0000000..6e51dbb
--- /dev/null
+++ b/src/cryptoblob.rs
@@ -0,0 +1,58 @@
+//! Helper functions for secret-key encrypted blobs
+//! that contain Zstd encrypted data
+
+use anyhow::{anyhow, Result};
+use serde::{Deserialize, Serialize};
+use zstd::stream::{decode_all as zstd_decode, encode_all as zstd_encode};
+
+use sodiumoxide::crypto::secretbox::xsalsa20poly1305::{self, gen_nonce, Nonce, NONCEBYTES};
+pub use sodiumoxide::crypto::secretbox::xsalsa20poly1305::{gen_key, Key, KEYBYTES};
+
+pub fn open(cryptoblob: &[u8], key: &Key) -> Result<Vec<u8>> {
+ if cryptoblob.len() < NONCEBYTES {
+ return Err(anyhow!("Cyphertext too short"));
+ }
+
+ // Decrypt -> get Zstd data
+ let nonce = Nonce::from_slice(&cryptoblob[..NONCEBYTES]).unwrap();
+ let zstdblob = xsalsa20poly1305::open(&cryptoblob[NONCEBYTES..], &nonce, key)
+ .map_err(|_| anyhow!("Could not decrypt blob"))?;
+
+ // Decompress zstd data
+ let mut reader = &zstdblob[..];
+ let data = zstd_decode(&mut reader)?;
+
+ Ok(data)
+}
+
+pub fn seal(plainblob: &[u8], key: &Key) -> Result<Vec<u8>> {
+ // Compress data using zstd
+ let mut reader = &plainblob[..];
+ let zstdblob = zstd_encode(&mut reader, 0)?;
+
+ // Encrypt
+ let nonce = gen_nonce();
+ let cryptoblob = xsalsa20poly1305::seal(&zstdblob, &nonce, key);
+
+ let mut res = Vec::with_capacity(NONCEBYTES + cryptoblob.len());
+ res.extend(nonce.as_ref());
+ res.extend(cryptoblob);
+
+ Ok(res)
+}
+
+pub fn open_deserialize<T: for<'de> Deserialize<'de>>(cryptoblob: &[u8], key: &Key) -> Result<T> {
+ let blob = open(cryptoblob, key)?;
+
+ Ok(rmp_serde::decode::from_read_ref::<_, T>(&blob)?)
+}
+
+pub fn seal_serialize<T: Serialize>(obj: T, key: &Key) -> Result<Vec<u8>> {
+ let mut wr = Vec::with_capacity(128);
+ let mut se = rmp_serde::Serializer::new(&mut wr)
+ .with_struct_map()
+ .with_string_variants();
+ obj.serialize(&mut se)?;
+
+ Ok(seal(&wr, key)?)
+}
diff --git a/src/main.rs b/src/main.rs
new file mode 100644
index 0000000..2b17e4c
--- /dev/null
+++ b/src/main.rs
@@ -0,0 +1,47 @@
+use anyhow::Result;
+
+use rusoto_credential::{EnvironmentProvider, ProvideAwsCredentials};
+use rusoto_signature::Region;
+
+mod bayou;
+mod cryptoblob;
+mod time;
+mod uidindex;
+
+use bayou::Bayou;
+use cryptoblob::Key;
+use uidindex::{UidIndex, UidIndexOp};
+
+#[tokio::main]
+async fn main() {
+ do_stuff().await.expect("Something failed");
+}
+
+async fn do_stuff() -> Result<()> {
+ let creds = EnvironmentProvider::default().credentials().await.unwrap();
+
+ let k2v_region = Region::Custom {
+ name: "garage-staging".to_owned(),
+ endpoint: "https://k2v-staging.home.adnab.me".to_owned(),
+ };
+
+ let s3_region = Region::Custom {
+ name: "garage-staging".to_owned(),
+ endpoint: "https://garage-staging.home.adnab.me".to_owned(),
+ };
+
+ let key = Key::from_slice(&[0u8; 32]).unwrap();
+
+ let mut mail_index = Bayou::<UidIndex>::new(
+ creds,
+ k2v_region,
+ s3_region,
+ "alex".into(),
+ "TestMailbox".into(),
+ key,
+ )?;
+
+ mail_index.sync().await?;
+
+ Ok(())
+}
diff --git a/src/time.rs b/src/time.rs
new file mode 100644
index 0000000..d34ee22
--- /dev/null
+++ b/src/time.rs
@@ -0,0 +1,9 @@
+use std::time::{SystemTime, UNIX_EPOCH};
+
+/// Returns milliseconds since UNIX Epoch
+pub fn now_msec() -> u64 {
+ SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Fix your clock :o")
+ .as_millis() as u64
+}
diff --git a/src/uidindex.rs b/src/uidindex.rs
new file mode 100644
index 0000000..7c5500f
--- /dev/null
+++ b/src/uidindex.rs
@@ -0,0 +1,168 @@
+use im::OrdMap;
+use serde::{de::Error, Deserialize, Deserializer, Serialize, Serializer};
+
+use crate::bayou::*;
+
+type ImapUid = u32;
+type ImapUidvalidity = u32;
+
+/// A Mail UUID is composed of two components:
+/// - a process identifier, 128 bits
+/// - a sequence number, 64 bits
+#[derive(Clone, Copy, PartialOrd, Ord, PartialEq, Eq)]
+pub struct MailUuid(pub [u8; 24]);
+
+#[derive(Clone)]
+pub struct UidIndex {
+ mail_uid: OrdMap<MailUuid, ImapUid>,
+ mail_flags: OrdMap<MailUuid, Vec<String>>,
+
+ mails_by_uid: OrdMap<ImapUid, MailUuid>,
+
+ uidvalidity: ImapUidvalidity,
+ uidnext: ImapUid,
+ internalseq: ImapUid,
+}
+
+#[derive(Clone, Serialize, Deserialize)]
+pub enum UidIndexOp {
+ MailAdd(MailUuid, ImapUid, Vec<String>),
+ MailDel(MailUuid),
+}
+
+impl Default for UidIndex {
+ fn default() -> Self {
+ Self {
+ mail_flags: OrdMap::new(),
+ mail_uid: OrdMap::new(),
+ mails_by_uid: OrdMap::new(),
+ uidvalidity: 1,
+ uidnext: 1,
+ internalseq: 1,
+ }
+ }
+}
+
+impl BayouState for UidIndex {
+ type Op = UidIndexOp;
+
+ fn apply(&self, op: &UidIndexOp) -> Self {
+ let mut new = self.clone();
+ match op {
+ UidIndexOp::MailAdd(uuid, uid, flags) => {
+ if *uid < new.internalseq {
+ new.uidvalidity += new.internalseq - *uid;
+ }
+ let new_uid = new.internalseq;
+
+ if let Some(prev_uid) = new.mail_uid.get(uuid) {
+ new.mails_by_uid.remove(prev_uid);
+ } else {
+ new.mail_flags.insert(*uuid, flags.clone());
+ }
+ new.mails_by_uid.insert(new_uid, *uuid);
+ new.mail_uid.insert(*uuid, new_uid);
+
+ new.internalseq += 1;
+ new.uidnext = new.internalseq;
+ }
+ UidIndexOp::MailDel(uuid) => {
+ if let Some(uid) = new.mail_uid.get(uuid) {
+ new.mails_by_uid.remove(uid);
+ new.mail_uid.remove(uuid);
+ new.mail_flags.remove(uuid);
+ }
+ new.internalseq += 1;
+ }
+ }
+ new
+ }
+}
+
+// ---- CUSTOM SERIALIZATION AND DESERIALIZATION ----
+
+#[derive(Serialize, Deserialize)]
+struct UidIndexSerializedRepr {
+ mails: Vec<(ImapUid, MailUuid, Vec<String>)>,
+ uidvalidity: ImapUidvalidity,
+ uidnext: ImapUid,
+ internalseq: ImapUid,
+}
+
+impl<'de> Deserialize<'de> for UidIndex {
+ fn deserialize<D>(d: D) -> Result<Self, D::Error>
+ where
+ D: Deserializer<'de>,
+ {
+ let val: UidIndexSerializedRepr = UidIndexSerializedRepr::deserialize(d)?;
+
+ let mut uidindex = UidIndex {
+ mail_flags: OrdMap::new(),
+ mail_uid: OrdMap::new(),
+ mails_by_uid: OrdMap::new(),
+ uidvalidity: val.uidvalidity,
+ uidnext: val.uidnext,
+ internalseq: val.internalseq,
+ };
+
+ for (uid, uuid, flags) in val.mails {
+ uidindex.mail_flags.insert(uuid, flags);
+ uidindex.mail_uid.insert(uuid, uid);
+ uidindex.mails_by_uid.insert(uid, uuid);
+ }
+
+ Ok(uidindex)
+ }
+}
+
+impl Serialize for UidIndex {
+ fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+ where
+ S: Serializer,
+ {
+ let mut mails = vec![];
+ for (uid, uuid) in self.mails_by_uid.iter() {
+ mails.push((
+ *uid,
+ *uuid,
+ self.mail_flags.get(uuid).cloned().unwrap_or_default(),
+ ));
+ }
+
+ let val = UidIndexSerializedRepr {
+ mails,
+ uidvalidity: self.uidvalidity,
+ uidnext: self.uidnext,
+ internalseq: self.internalseq,
+ };
+
+ val.serialize(serializer)
+ }
+}
+
+impl<'de> Deserialize<'de> for MailUuid {
+ fn deserialize<D>(d: D) -> Result<Self, D::Error>
+ where
+ D: Deserializer<'de>,
+ {
+ let v = String::deserialize(d)?;
+ let bytes = hex::decode(v).map_err(|_| D::Error::custom("invalid hex"))?;
+
+ if bytes.len() != 24 {
+ return Err(D::Error::custom("bad length"));
+ }
+
+ let mut tmp = [0u8; 24];
+ tmp[..].copy_from_slice(&bytes);
+ Ok(Self(tmp))
+ }
+}
+
+impl Serialize for MailUuid {
+ fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+ where
+ S: Serializer,
+ {
+ serializer.serialize_str(&hex::encode(self.0))
+ }
+}
generated by cgit v1.2.3 (git 2.25.1) at 2024-09-20 09:40:35 +0000