initial commit

.gitignore

@@ -0,0 +1,8 @@
+target
+Cargo.lock
+*.db
+
+.idea/
+*.iml
+*.ipr
+

Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "llama_db"
+version = "0.1.0"
+authors = ["Fluence Labs"]
+publish = false
+description = "LlamaDb wrapper for running into Fluence WasmVm"
+edition = "2018"
+
+[profile.release]
+debug = false
+lto = true
+opt-level = 3
+debug-assertions = false
+overflow-checks = false
+panic = "abort"
+
+[lib]
+name = "llama_db"
+path = "src/lib.rs"
+crate-type = ["cdylib"]
+
+[dependencies]
+lazy_static = "1.1.0"
+fluence = { version = "0.1.0" }
+llamadb = { git = "https://github.com/fluencelabs/llamadb.git", branch = "master" }
+libsecp256k1 = "0.2.2"
+sha2 = "0.8.0"

src/lib.rs

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2018 Fluence Labs Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+//! Wrapper for Llamadb (a test for Fluence network).
+//!
+//! Provides the FFI (`main`) for interact with Llamadb.
+
+#[macro_use]
+extern crate lazy_static;
+
+use std::error::Error;
+use std::sync::Mutex;
+
+use fluence::sdk::*;
+use llamadb::tempdb::{ExecuteStatementResponse, TempDb};
+
+use crate::signature::*;
+
+mod signature;
+#[cfg(test)]
+mod tests;
+
+/// Result for all possible Error types.
+type GenResult<T> = ::std::result::Result<T, Box<Error>>;
+
+lazy_static! {
+    static ref DATABASE: Mutex<TempDb> = Mutex::new(TempDb::new());
+}
+
+/// Flag to toggle signature validation
+static CHECK_SIGNATURE: bool = false;
+
+/// Executes SQL and converts llamadb error to string.
+#[invocation_handler]
+fn main(input: String) -> String {
+    let result = if CHECK_SIGNATURE {
+        check_input(&input).and_then(run_query)
+    } else {
+        run_query(&input)
+    };
+
+    match result {
+        Ok(response) => response,
+        Err(err_msg) => format!("[Error] {}", err_msg),
+    }
+}
+
+/// Acquires lock, does query, releases lock, returns query result.
+fn run_query(sql_query: &str) -> GenResult<String> {
+    let mut db = DATABASE.lock()?;
+    db.do_query(sql_query)
+        .map(statement_to_string)
+        .map_err(Into::into)
+}
+
+/// Converts query result to CSV String.
+fn statement_to_string(statement: ExecuteStatementResponse) -> String {
+    match statement {
+        ExecuteStatementResponse::Created => "table created".to_string(),
+        ExecuteStatementResponse::Dropped => "table was dropped".to_string(),
+        ExecuteStatementResponse::Inserted(number) => format!("rows inserted: {}", number),
+        ExecuteStatementResponse::Select { column_names, rows } => {
+            let col_names = column_names.to_vec().join(", ") + "\n";
+            let rows_as_str = rows
+                .map(|row| {
+                    row.iter()
+                        .map(|elem| elem.to_string())
+                        .collect::<Vec<String>>()
+                        .join(", ")
+                })
+                .collect::<Vec<String>>()
+                .join("\n");
+
+            col_names + &rows_as_str
+        }
+        ExecuteStatementResponse::Deleted(number) => format!("rows deleted: {}", number),
+        ExecuteStatementResponse::Explain(result) => result,
+        ExecuteStatementResponse::Updated(number) => format!("rows updated: {}", number),
+    }
+}

src/signature.rs

@@ -0,0 +1,119 @@
+use crate::GenResult;
+use core::fmt;
+use std::num::ParseIntError;
+
+use secp256k1::{verify, Message, PublicKey, Signature};
+use sha2::{Digest, Sha256};
+
+#[derive(Debug)]
+struct Error(String);
+
+impl std::error::Error for Error {}
+
+impl fmt::Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+/// Representation of a signed request.
+/// Format is: signature\nnonce\npayload
+/// nonce and payload are concatenated here to avoid reallocation on hashing
+struct Signed<'a> {
+    signature: &'a str,
+    nonce_payload: &'a str,
+}
+
+impl<'a> Signed<'a> {
+    pub fn payload(&self) -> GenResult<&'a str> {
+        let pos = self.nonce_payload.find("\n").ok_or(err_msg(&format!(
+            "Invalid input: no \\n between nonce and payload in `{}`. \
+             Should be <signature hex>\\n<nonce>\\n<sql_query>",
+            self.nonce_payload
+        )))?;
+        Ok(&self.nonce_payload[pos + 1..])
+    }
+}
+
+lazy_static! {
+    static ref PK: PublicKey = get_pk();
+}
+
+/// hard-coded public key, could be replaced directly in a final Wasm binary
+// Full: 64 + 1 byte prefix
+static PUBLIC_KEY: [u8; 65] = [
+    0x04,
+    0xba,0x94,0x28,0x52,0xe4,0x35,0x39,0x1b,0x2a,0x6a,0x99,0x3f,0x33,0xf7,0x0c,0x43,
+    0x92,0x35,0xf4,0x84,0x15,0xe2,0x5c,0x66,0x8d,0x97,0xd5,0xb0,0xf2,0x63,0xb5,0x4e,
+    0x9d,0x98,0xcc,0xae,0xfd,0xc5,0xc6,0xca,0x23,0x7f,0xfc,0x0f,0x2f,0x63,0x35,0x13,
+    0x30,0xfa,0xaf,0xe3,0x1d,0x12,0x03,0x79,0xb8,0xdf,0xf9,0x82,0xa3,0x73,0x58,0xe3
+];
+
+fn get_pk() -> PublicKey {
+    PublicKey::parse_slice(&PUBLIC_KEY, None).expect("Invalid public key")
+}
+
+fn err_msg(s: &str) -> Box<Error> {
+    Error(s.to_string()).into()
+}
+
+/// Converts hex string to a Vec<u8>
+fn decode_hex(s: &str) -> GenResult<Vec<u8>> {
+    if s.len() % 2 != 0 {
+        return Err(err_msg(&format!("Invalid hex length: {} isn't divisible by 2", s.len())))
+    }
+    (0..s.len())
+        .step_by(2)
+        .map(|i| u8::from_str_radix(&s[i..i + 2], 16))
+        .collect::<Result<Vec<u8>, ParseIntError>>()
+        .map_err(Into::into)
+}
+
+/// SHA-256 hash of the string
+fn hash_message(message: &str) -> [u8; 32] {
+    let mut sha = Sha256::default();
+    sha.input(message.as_bytes());
+    let hash = sha.result();
+    let mut result = [0; 32];
+    //TODO: is there a better way for GenericArray<u8, 32> -> [u8; 32] ?
+    result.copy_from_slice(hash.as_slice());
+    result
+}
+
+/// Verifies if signature is correct
+fn check_signature(hash: &[u8; 32], signature: &str) -> GenResult<bool> {
+    let signature = decode_hex(signature)?;
+    let signature = Signature::parse_slice(signature.as_slice())
+        .map_err(|e| err_msg(&format!("Error parsing signature: {:?}", e)))?;
+    let message = Message::parse(hash);
+
+    Ok(verify(&message, &signature, &PK))
+}
+
+/// Parse input as `signature\nnonce\npayload`
+fn parse_signed(input: &String) -> GenResult<Signed> {
+    let pos: usize = input.find("\n").ok_or(err_msg(&format!(
+        "Invalid input: no '\\n' between signature and nonce in `{}`. \
+         Should be <signature hex>\\n<nonce>\\n<sql_query>",
+        input
+    )))?;
+    let signature: &str = &input[..pos];
+    let nonce_payload: &str = &input[pos + 1..];
+    Ok(Signed {
+        signature,
+        nonce_payload,
+    })
+}
+
+/// Checks if input is signed
+/// returns payload string on success
+/// throws an error on failure
+pub fn check_input(input: &String) -> GenResult<&str> {
+    let signed = parse_signed(input)?;
+    let hash = hash_message(signed.nonce_payload);
+    if check_signature(&hash, signed.signature)? {
+        signed.payload()
+    } else {
+        Err(err_msg("Invalid signature"))
+    }
+}

src/tests.rs

@@ -0,0 +1,160 @@
+/*
+ * Copyright 2018 Fluence Labs Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+//! Module with tests.
+
+#[test]
+fn integration_sql_test() {
+    //
+    // Success cases.
+    //
+
+    let create_table = execute_sql("CREATE TABLE Users(id INT, name TEXT, age INT)");
+    assert_eq!(create_table, "table created");
+
+    let insert_one = execute_sql("INSERT INTO Users VALUES(1, 'Sara', 23)");
+    assert_eq!(insert_one, "rows inserted: 1");
+
+    let insert_several =
+        execute_sql("INSERT INTO Users VALUES(2, 'Bob', 19), (3, 'Caroline', 31), (4, 'Max', 25)");
+    assert_eq!(insert_several, "rows inserted: 3");
+
+    let create_table_role = execute_sql("CREATE TABLE Roles(user_id INT, role VARCHAR(128))");
+    assert_eq!(create_table_role, "table created");
+
+    let insert_roles = execute_sql(
+        "INSERT INTO Roles VALUES(1, 'Teacher'), (2, 'Student'), (3, 'Scientist'), (4, 'Writer')",
+    );
+    assert_eq!(insert_roles, "rows inserted: 4");
+
+    let empty_select = execute_sql("SELECT * FROM Users WHERE name = 'unknown'");
+    assert_eq!(empty_select, "id, name, age\n");
+
+    let select_all = execute_sql(
+        "SELECT min(user_id) as min, max(user_id) as max, \
+         count(user_id) as count, sum(user_id) as sum, avg(user_id) as avg FROM Roles",
+    );
+    assert_eq!(select_all, "min, max, count, sum, avg\n1, 4, 4, 10, 2.5");
+
+    let select_with_join = execute_sql(
+        "SELECT u.name AS Name, r.role AS Role FROM Users u JOIN Roles \
+         r ON u.id = r.user_id WHERE r.role = 'Writer'",
+    );
+    assert_eq!(select_with_join, "name, role\nMax, Writer");
+
+    let explain = execute_sql("EXPLAIN SELECT id, name FROM Users");
+    assert_eq!(
+        explain,
+        "query plan\n".to_string()
+            + "column names: (`id`, `name`)\n"
+            + "(scan `users` :source-id 0\n"
+            + "  (yield\n"
+            + "    (column-field :source-id 0 :column-offset 0)\n"
+            + "    (column-field :source-id 0 :column-offset 1)))"
+    );
+
+    let delete = execute_sql(
+        "DELETE FROM Users WHERE id = (SELECT user_id FROM Roles WHERE role = 'Student');",
+    );
+    assert_eq!(delete, "rows deleted: 1");
+
+    let update_query = execute_sql("UPDATE Users SET name = 'Min' WHERE name = 'Max'");
+    assert_eq!(update_query, "rows updated: 1");
+
+    //
+    // Error cases.
+    //
+
+    let empty_str = execute_sql("");
+    assert_eq!(empty_str, "[Error] Expected SELECT, INSERT, CREATE, DELETE, TRUNCATE or EXPLAIN statement; got no more tokens");
+
+    let invalid_sql = execute_sql("123");
+    assert_eq!(invalid_sql, "[Error] Expected SELECT, INSERT, CREATE, DELETE, TRUNCATE or EXPLAIN statement; got Number(\"123\")");
+
+    let bad_query = execute_sql("SELECT salary FROM Users");
+    assert_eq!(bad_query, "[Error] column does not exist: salary");
+
+    let lexer_error = execute_sql("π");
+    assert_eq!(lexer_error, "[Error] Lexer error: Unknown character π");
+
+    let incompatible_types = execute_sql("SELECT * FROM Users WHERE age = 'Bob'");
+    assert_eq!(
+        incompatible_types,
+        "[Error] 'Bob' cannot be cast to Integer { signed: true, bytes: 8 }"
+    );
+
+    let not_supported_order_by = execute_sql("SELECT * FROM Users ORDER BY name");
+    assert_eq!(
+        not_supported_order_by,
+        "[Error] order by in not implemented"
+    );
+
+    let truncate = execute_sql("TRUNCATE TABLE Users");
+    assert_eq!(truncate, "rows deleted: 3");
+
+    let drop_table = execute_sql("DROP TABLE Users");
+    assert_eq!(drop_table, "table was dropped");
+
+    let select_by_dropped_table = execute_sql("SELECT * FROM Users");
+    assert_eq!(
+        select_by_dropped_table,
+        "[Error] table does not exist: users"
+    );
+}
+
+//
+// Private helper functions.
+//
+use std::mem;
+use std::ptr;
+
+pub unsafe fn read_result_from_mem(ptr: *mut u8) -> Vec<u8> {
+    const RESULT_SIZE_BYTES: usize = 4;
+
+    // read string length from current pointer
+    let mut len_as_bytes: [u8; RESULT_SIZE_BYTES] = [0; RESULT_SIZE_BYTES];
+    ptr::copy_nonoverlapping(ptr, len_as_bytes.as_mut_ptr(), RESULT_SIZE_BYTES);
+
+    let input_len: u32 = mem::transmute(len_as_bytes);
+    let total_len = input_len as usize + RESULT_SIZE_BYTES;
+
+    // creates object from raw bytes
+    let mut input = Vec::from_raw_parts(ptr, total_len, total_len);
+
+    // drains RESULT_SIZE_BYTES from the beginning of created vector, it allows to effectively
+    // skips (without additional allocations) length of the result.
+    {
+        input.drain(0..4);
+    }
+    input
+}
+
+/// Executes sql and returns result as a String.
+fn execute_sql(sql: &str) -> String {
+    unsafe {
+        use std::mem;
+
+        let mut sql_str = sql.to_string();
+        // executes query
+        let result_str_ptr = super::invoke(sql_str.as_bytes_mut().as_mut_ptr(), sql_str.len());
+        // ownership of memory of 'sql_str' will be taken through 'ptr' inside
+        // 'invoke' method, have to forget 'sql_str'  for prevent deallocation
+        mem::forget(sql_str);
+        // converts results
+        let result_str = read_result_from_mem(result_str_ptr.as_ptr());
+        String::from_utf8(result_str).unwrap()
+    }
+}