Implement receiving part of LSRPC logic

.gitignore

@@ -1,3 +1,4 @@
 /target
 *.db
-.DS_Store
+.DS_Store
+*.pem

Cargo.lock

@@ -154,6 +154,19 @@ dependencies = [
  "cipher",
 ]
 
+[[package]]
+name = "curve25519-dalek"
+version = "3.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f627126b946c25a4638eec0ea634fc52506dea98db118aae985118ce7c3d723f"
+dependencies = [
+ "byteorder",
+ "digest",
+ "rand_core 0.5.1",
+ "subtle",
+ "zeroize",
+]
+
 [[package]]
 name = "digest"
 version = "0.9.0"
@@ -639,6 +652,17 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "pem"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fd56cbd21fea48d0c440b41cd69c589faacade08c992d9a54e471b79d0fd13eb"
+dependencies = [
+ "base64",
+ "once_cell",
+ "regex",
+]
+
 [[package]]
 name = "percent-encoding"
 version = "2.1.0"
@@ -959,6 +983,8 @@ version = "1.0.0"
 dependencies = [
  "aes-gcm",
  "hex",
+ "lazy_static",
+ "pem",
  "r2d2",
  "r2d2_sqlite",
  "regex",
@@ -967,6 +993,7 @@ dependencies = [
  "serde_json",
  "tokio",
  "warp",
+ "x25519-dalek",
 ]
 
 [[package]]
@@ -1031,6 +1058,18 @@ dependencies = [
  "unicode-xid",
 ]
 
+[[package]]
+name = "synstructure"
+version = "0.12.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b834f2d66f734cb897113e34aaff2f1ab4719ca946f9a7358dba8f8064148701"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+ "unicode-xid",
+]
+
 [[package]]
 name = "tempfile"
 version = "3.2.0"
@@ -1371,3 +1410,35 @@ name = "winapi-x86_64-pc-windows-gnu"
 version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
+
+[[package]]
+name = "x25519-dalek"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bc614d95359fd7afc321b66d2107ede58b246b844cf5d8a0adcca413e439f088"
+dependencies = [
+ "curve25519-dalek",
+ "rand_core 0.5.1",
+ "zeroize",
+]
+
+[[package]]
+name = "zeroize"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "81a974bcdd357f0dca4d41677db03436324d45a4c9ed2d0b873a5a360ce41c36"
+dependencies = [
+ "zeroize_derive",
+]
+
+[[package]]
+name = "zeroize_derive"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c3f369ddb18862aba61aa49bf31e74d29f0f162dec753063200e1dc084345d16"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+ "synstructure",
+]

Cargo.toml

@@ -7,6 +7,8 @@ edition = "2018"
 [dependencies]
 aes-gcm = "0.8.0"
 hex = "0.4.3"
+lazy_static = "1.4.0"
+pem = "0.8.3"
 regex = "1.4"
 rusqlite = "0.24"
 r2d2_sqlite = "0.17.0"
@@ -15,3 +17,4 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 tokio = { version = "1", features = ["full"] }
 warp = "0.3"
+x25519-dalek = "1.1.0"

README.md

@@ -1,11 +1,20 @@
 
+### Step 1: Generating an X25519 key pair for your open group server
+
+To generate an X25519 key pair, simply run:
+
+```
+openssl genpkey -algorithm x25519 -out x25519_private_key.pem
+openssl pkey -in x25519_private_key.pem -pubout -out x25519_public_key.pem
+```
+
+Make sure you have openssl installed, and make sure you're pointing to the right openssl installation as well (e.g. macOS provides an old default implementation that doesn't have the X25519 algorithm).
+
+### Step 2: Building the project:
+
 To build and run the project, do:
 
 ```
 cargo build
 cargo run
 ```
-
-To run tests:
-
-`cargo test`

src/crypto.rs

@@ -10,10 +10,24 @@ impl warp::reject::Reject for DecryptionError { }
 // By default the aes-gcm crate will use software implementations of both AES and the POLYVAL universal hash function. When 
 // targeting modern x86/x86_64 CPUs, use the following RUSTFLAGS to take advantage of high performance AES-NI and CLMUL CPU
 // intrinsics:
+//
 // RUSTFLAGS="-Ctarget-cpu=sandybridge -Ctarget-feature=+aes,+sse2,+sse4.1,+ssse3"
 
 const IV_SIZE: usize = 12;
 
+pub fn get_x25519_symmetric_key(public_key: Vec<u8>, private_key: Vec<u8>) -> Result<Vec<u8>, warp::reject::Rejection> {
+    if public_key.len() != 32 {
+        println!("Couldn't create symmetric key using public key of invalid length.");
+        return Err(warp::reject::custom(DecryptionError)); 
+    }
+    let private_key: [u8; 32] = private_key.try_into().unwrap(); // Guaranteed to be 32 bytes
+    let dalek_private_key = x25519_dalek::StaticSecret::from(private_key);
+    let public_key: [u8; 32] = public_key.try_into().unwrap();
+    let dalek_public_key = x25519_dalek::PublicKey::from(public_key);
+    let symmetric_key = dalek_private_key.diffie_hellman(&dalek_public_key);
+    return Ok(symmetric_key.to_bytes().try_into().unwrap());
+}
+
 pub fn decrypt_aes_gcm(iv_and_ciphertext: Vec<u8>, symmetric_key: Vec<u8>) -> Result<Vec<u8>, warp::reject::Rejection> {
     if iv_and_ciphertext.len() < IV_SIZE { 
         println!("Ignoring ciphertext of invalid size.");

src/handlers.rs

@@ -3,6 +3,7 @@ use rusqlite::params;
 use warp::{Rejection, http::StatusCode};
 
 use super::models;
+use super::routes;
 use super::storage;
 
 #[derive(Debug)]
@@ -34,7 +35,7 @@ pub async fn insert_message(mut message: models::Message, pool: storage::Databas
 }
 
 /// Returns either the last `limit` messages or all messages since `from_server_id, limited to `limit`.
-pub async fn get_messages(options: models::QueryOptions, pool: storage::DatabaseConnectionPool) -> Result<impl warp::Reply, Rejection> {
+pub async fn get_messages(options: routes::QueryOptions, pool: storage::DatabaseConnectionPool) -> Result<impl warp::Reply, Rejection> {
     // Get a database connection
     let conn = storage::conn(&pool)?;
     // Unwrap parameters
@@ -85,7 +86,7 @@ pub async fn delete_message(row_id: i64, pool: storage::DatabaseConnectionPool)
 }
 
 /// Returns either the last `limit` deleted messages or all deleted messages since `from_server_id, limited to `limit`.
-pub async fn get_deleted_messages(options: models::QueryOptions, pool: storage::DatabaseConnectionPool) -> Result<impl warp::Reply, Rejection> {
+pub async fn get_deleted_messages(options: routes::QueryOptions, pool: storage::DatabaseConnectionPool) -> Result<impl warp::Reply, Rejection> {
     // Get a database connection
     let conn = storage::conn(&pool)?;
     // Unwrap parameters

src/lsrpc.rs

@@ -0,0 +1,89 @@
+use std::convert::TryInto;
+
+use regex::Regex;
+use serde::{Deserialize, Serialize};
+use warp::Rejection;
+
+use super::crypto;
+use super::storage;
+
+#[derive(Deserialize, Serialize, Debug)]
+struct LSRPCPayload {
+    pub ciphertext: Vec<u8>,
+    pub metadata: LSRPCPayloadMetadata
+}
+
+#[derive(Deserialize, Serialize, Debug)]
+struct LSRPCPayloadMetadata {
+    pub ephemeral_key: String
+}
+
+#[derive(Debug)]
+pub struct ParsingError;
+impl warp::reject::Reject for ParsingError { }
+
+pub async fn handle_lsrpc_request(blob: warp::hyper::body::Bytes, pool: storage::DatabaseConnectionPool) -> Result<impl warp::Reply, Rejection> {
+    let payload = parse_lsrpc_request(blob)?;
+    let plaintext = decrypt_lsrpc_request(payload)?;
+    println!("{}", String::from_utf8(plaintext).unwrap());
+    return Ok(warp::reply::reply());
+}
+
+fn parse_lsrpc_request(blob: warp::hyper::body::Bytes) -> Result<LSRPCPayload, Rejection> {
+    // The encoding of onion requests looks like: | 4 bytes: size N of ciphertext | N bytes: ciphertext | json as utf8 |
+    if blob.len() < 4 { 
+        println!("Ignoring blob of invalid size.");
+        return Err(warp::reject::custom(ParsingError)); 
+    }
+    // Extract the different components
+    let size = as_le_u32(&blob[0..4].try_into().unwrap()) as usize;
+    let ciphertext: Vec<u8> = blob[4..(4 + size)].try_into().unwrap();
+    let utf8_json: Vec<u8> = blob[(4 + size)..].try_into().unwrap();
+    // Parse JSON
+    let json = match String::from_utf8(utf8_json) {
+        Ok(json) => json,
+        Err(e) => {
+            println!("Couldn't parse string from bytes due to error: {:?}.", e);
+            return Err(warp::reject::custom(ParsingError));
+        }
+    };
+    // Parse metadata
+    let metadata: LSRPCPayloadMetadata = match serde_json::from_str(&json) {
+        Ok(metadata) => metadata,
+        Err(e) => {
+            println!("Couldn't parse LSRPC request metadata due to error: {:?}.", e);
+            return Err(warp::reject::custom(ParsingError));
+        }
+    };
+    // Check that the ephemeral public key is valid hex
+    let re = Regex::new(r"^[0-9a-fA-F]+$").unwrap(); // Force
+    if !re.is_match(&metadata.ephemeral_key) { 
+        println!("Ignoring non hex encoded LSRPC request ephemeral key.");
+        return Err(warp::reject::custom(ParsingError)); 
+    };
+    // Return
+    return Ok(LSRPCPayload { ciphertext : ciphertext, metadata : metadata });
+}
+
+fn decrypt_lsrpc_request(payload: LSRPCPayload) -> Result<Vec<u8>, Rejection> {
+    let public_key = hex::decode(payload.metadata.ephemeral_key).unwrap(); // Safe because it was validated in the parsing step
+    let symmetric_key = crypto::get_x25519_symmetric_key(public_key, get_private_key())?;
+    let plaintext = crypto::decrypt_aes_gcm(payload.ciphertext, symmetric_key)?;
+    return Ok(plaintext);
+}
+
+// Utilities
+
+// FIXME: get_private_key() should be a lazy static variable
+
+fn get_private_key() -> Vec<u8> {
+    let raw = std::fs::read_to_string("x25519_private_key.pem").unwrap();
+    return pem::parse(raw).unwrap().contents;
+}
+
+fn as_le_u32(array: &[u8; 4]) -> u32 {
+    ((array[0] as u32) << 00) +
+    ((array[1] as u32) << 08) +
+    ((array[2] as u32) << 16) +
+    ((array[3] as u32) << 24)
+}

src/main.rs

@@ -1,5 +1,6 @@
 mod crypto;
 mod handlers;
+mod lsrpc;
 mod models;
 mod routes;
 mod storage;

src/models.rs

@@ -16,9 +16,3 @@ impl Message {
         return !self.text.is_empty();
     }
 }
-
-#[derive(Debug, Deserialize)]
-pub struct QueryOptions {
-    pub limit: Option<u16>,
-    pub from_server_id: Option<i64>
-}

src/routes.rs

@@ -1,9 +1,31 @@
+use serde::Deserialize;
 use warp::{Filter, http::StatusCode, Rejection};
 
+use super::crypto;
 use super::handlers;
+use super::lsrpc;
 use super::models;
 use super::storage;
 
+#[derive(Debug, Deserialize)]
+pub struct QueryOptions {
+    pub limit: Option<u16>,
+    pub from_server_id: Option<i64>
+}
+
+/// POST /loki/v3/lsrpc
+pub fn lsrpc(
+    db_pool: storage::DatabaseConnectionPool
+) -> impl Filter<Extract = impl warp::Reply, Error = Rejection> + Clone {
+    return warp::post()
+        .and(warp::path("loki/v3/lsrpc"))
+        .and(warp::body::content_length_limit(10 * 1024 * 1024)) // Match storage server
+        .and(warp::body::bytes()) // Expect bytes
+        .and(warp::any().map(move || db_pool.clone()))
+        .and_then(lsrpc::handle_lsrpc_request)
+        .recover(handle_error);
+}
+
 /// POST /messages
 pub fn send_message(
     db_pool: storage::DatabaseConnectionPool
@@ -25,7 +47,7 @@ pub fn get_messages(
 ) -> impl Filter<Extract = impl warp::Reply, Error = Rejection> + Clone {
     return warp::get()
         .and(warp::path("messages"))
-        .and(warp::query::<models::QueryOptions>())
+        .and(warp::query::<QueryOptions>())
         .and(warp::any().map(move || db_pool.clone()))
         .and_then(handlers::get_messages)
         .recover(handle_error);
@@ -50,7 +72,7 @@ pub fn get_deleted_messages(
 ) -> impl Filter<Extract = impl warp::Reply, Error = Rejection> + Clone {
     return warp::get()
         .and(warp::path("deleted_messages"))
-        .and(warp::query::<models::QueryOptions>())
+        .and(warp::query::<QueryOptions>())
         .and(warp::any().map(move || db_pool.clone()))
         .and_then(handlers::get_deleted_messages)
         .recover(handle_error);
@@ -122,7 +144,8 @@ pub fn get_member_count(
 pub fn get_all(
     db_pool: &storage::DatabaseConnectionPool
 ) -> impl Filter<Extract = impl warp::Reply, Error = Rejection> + Clone {
-    return send_message(db_pool.clone())
+    return lsrpc(db_pool.clone())
+        .or(send_message(db_pool.clone()))
         .or(get_messages(db_pool.clone()))
         .or(delete_message(db_pool.clone()))
         .or(get_deleted_messages(db_pool.clone()))
@@ -138,6 +161,12 @@ async fn handle_error(e: Rejection) -> Result<impl warp::Reply, Rejection> {
     if let Some(models::ValidationError) = e.find() {
         return Ok(warp::reply::with_status(reply, StatusCode::BAD_REQUEST)); // 400
     }
+    if let Some(crypto::DecryptionError) = e.find() {
+        return Ok(warp::reply::with_status(reply, StatusCode::BAD_REQUEST)); // 400
+    }
+    if let Some(lsrpc::ParsingError) = e.find() {
+        return Ok(warp::reply::with_status(reply, StatusCode::BAD_REQUEST)); // 400
+    }
     if let Some(handlers::UnauthorizedError) = e.find() {
         return Ok(warp::reply::with_status(reply, StatusCode::FORBIDDEN)); // 403
     }

src/storage.rs

@@ -16,7 +16,8 @@ pub const BLOCK_LIST_TABLE: &str = "block_list";
 
 pub fn create_tables_if_needed(conn: &DatabaseConnection) {
     // Messages
-    // The `id` field is needed to make `rowid` stable
+    // The `id` field is needed to make `rowid` stable, which is important because otherwise
+    // the `id`s in this table won't correspond to those in the deleted messages table
     let messages_table_cmd = format!(
     "CREATE TABLE IF NOT EXISTS {} (
         id INTEGER PRIMARY KEY,