Merge pull request #26 from garious/add-accountant

Add testnode and client-demo

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "silk"
 description = "A silky smooth implementation of the Loom architecture"
-version = "0.2.2"
+version = "0.3.0"
 documentation = "https://docs.rs/silk"
 homepage = "http://loomprotocol.com/"
 repository = "https://github.com/loomprotocol/silk"
@@ -15,6 +15,14 @@ license = "Apache-2.0"
 name = "silk-demo"
 path = "src/bin/demo.rs"
 
+[[bin]]
+name = "silk-client-demo"
+path = "src/bin/client-demo.rs"
+
+[[bin]]
+name = "silk-testnode"
+path = "src/bin/testnode.rs"
+
 [badges]
 codecov = { repository = "loomprotocol/silk", branch = "master", service = "github" }
 
@@ -29,3 +37,6 @@ sha2-asm = {version="0.3", optional=true}
 generic-array = { version = "0.9.0", default-features = false, features = ["serde"] }
 serde = "1.0.27"
 serde_derive = "1.0.27"
+ring = "0.12.1"
+untrusted = "0.5.1"
+bincode = "1.0.0"

README.md

@@ -24,7 +24,7 @@ Create a *Historian* and send it *events* to generate an *event log*, where each
 is tagged with the historian's latest *hash*. Then ensure the order of events was not tampered
 with by verifying each entry's hash can be generated from the hash in the previous entry:
 
-![historian](https://user-images.githubusercontent.com/55449/36499105-7c8db6a0-16fd-11e8-8b88-c6e0f52d7a50.png)
+![historian](https://user-images.githubusercontent.com/55449/36633440-f76f7bb8-1952-11e8-8328-387861d3d464.png)
 
 ```rust
 extern crate silk;
@@ -37,7 +37,8 @@ use std::sync::mpsc::SendError;
 
 fn create_log(hist: &Historian) -> Result<(), SendError<Event>> {
     sleep(Duration::from_millis(15));
-    hist.sender.send(Event::UserDataKey(Sha256Hash::default()))?;
+    let data = Sha256Hash::default();
+    hist.sender.send(Event::Discovery { data })?;
     sleep(Duration::from_millis(10));
     Ok(())
 }
@@ -62,7 +63,7 @@ Running the program should produce a log similar to:
 
 ```rust
 Entry { num_hashes: 0, end_hash: [0, ...], event: Tick }
-Entry { num_hashes: 2, end_hash: [67, ...], event: UserDataKey(3735928559) }
+Entry { num_hashes: 2, end_hash: [67, ...], event: Discovery { data: [37, ...] } }
 Entry { num_hashes: 3, end_hash: [123, ...], event: Tick }
 ```
 

diagrams/historian.msc

@@ -4,10 +4,10 @@ msc {
   logger=>historian [ label = "e0 = Entry{hash: h0, n: 0, event: Tick}" ] ;
   logger=>logger [ label = "h1 = hash(h0)" ] ;
   logger=>logger [ label = "h2 = hash(h1)" ] ;
-  client=>historian [ label = "UserData(d0)" ] ;
-  historian=>logger [ label = "UserData(d0)" ] ;
+  client=>historian [ label = "Discovery(d0)" ] ;
+  historian=>logger [ label = "Discovery(d0)" ] ;
   logger=>logger [ label = "h3 = hash(h2 + d0)" ] ;
-  logger=>historian [ label = "e1 = Entry{hash: hash(h3), n: 2, event: UserData(d0)}" ] ;
+  logger=>historian [ label = "e1 = Entry{hash: hash(h3), n: 2, event: Discovery(d0)}" ] ;
   logger=>logger [ label = "h4 = hash(h3)" ] ;
   logger=>logger [ label = "h5 = hash(h4)" ] ;
   logger=>logger [ label = "h6 = hash(h5)" ] ;

src/accountant.rs

@@ -0,0 +1,229 @@
+//! The `accountant` is a client of the `historian`. It uses the historian's
+//! event log to record transactions. Its users can deposit funds and
+//! transfer funds to other users.
+
+use log::{Event, PublicKey, Sha256Hash, Signature};
+use historian::Historian;
+use ring::signature::Ed25519KeyPair;
+use std::sync::mpsc::{RecvError, SendError};
+use std::collections::HashMap;
+
+pub struct Accountant {
+    pub historian: Historian<u64>,
+    pub balances: HashMap<PublicKey, u64>,
+    pub end_hash: Sha256Hash,
+}
+
+impl Accountant {
+    pub fn new(start_hash: &Sha256Hash, ms_per_tick: Option<u64>) -> Self {
+        let hist = Historian::<u64>::new(start_hash, ms_per_tick);
+        Accountant {
+            historian: hist,
+            balances: HashMap::new(),
+            end_hash: *start_hash,
+        }
+    }
+
+    pub fn process_event(self: &mut Self, event: &Event<u64>) {
+        match *event {
+            Event::Claim { key, data, .. } => {
+                if self.balances.contains_key(&key) {
+                    if let Some(x) = self.balances.get_mut(&key) {
+                        *x += data;
+                    }
+                } else {
+                    self.balances.insert(key, data);
+                }
+            }
+            Event::Transaction { from, to, data, .. } => {
+                if let Some(x) = self.balances.get_mut(&from) {
+                    *x -= data;
+                }
+                if self.balances.contains_key(&to) {
+                    if let Some(x) = self.balances.get_mut(&to) {
+                        *x += data;
+                    }
+                } else {
+                    self.balances.insert(to, data);
+                }
+            }
+            _ => (),
+        }
+    }
+
+    pub fn sync(self: &mut Self) {
+        let mut entries = vec![];
+        while let Ok(entry) = self.historian.receiver.try_recv() {
+            entries.push(entry);
+        }
+        // TODO: Does this cause the historian's channel to get blocked?
+        //use log::verify_slice_u64;
+        //println!("accountant: verifying {} entries...", entries.len());
+        //assert!(verify_slice_u64(&entries, &self.end_hash));
+        //println!("accountant: Done verifying {} entries.", entries.len());
+        if let Some(last_entry) = entries.last() {
+            self.end_hash = last_entry.end_hash;
+        }
+        for e in &entries {
+            self.process_event(&e.event);
+        }
+    }
+
+    pub fn deposit_signed(
+        self: &Self,
+        key: PublicKey,
+        data: u64,
+        sig: Signature,
+    ) -> Result<(), SendError<Event<u64>>> {
+        let event = Event::Claim { key, data, sig };
+        self.historian.sender.send(event)
+    }
+
+    pub fn deposit(
+        self: &Self,
+        n: u64,
+        keypair: &Ed25519KeyPair,
+    ) -> Result<(), SendError<Event<u64>>> {
+        use log::{get_pubkey, sign_serialized};
+        let key = get_pubkey(keypair);
+        let sig = sign_serialized(&n, keypair);
+        self.deposit_signed(key, n, sig)
+    }
+
+    pub fn transfer_signed(
+        self: &mut Self,
+        from: PublicKey,
+        to: PublicKey,
+        data: u64,
+        sig: Signature,
+    ) -> Result<(), SendError<Event<u64>>> {
+        if self.get_balance(&from).unwrap() < data {
+            // TODO: Replace the SendError result with a custom one.
+            println!("Error: Insufficient funds");
+            return Ok(());
+        }
+        let event = Event::Transaction {
+            from,
+            to,
+            data,
+            sig,
+        };
+        self.historian.sender.send(event)
+    }
+
+    pub fn transfer(
+        self: &mut Self,
+        n: u64,
+        keypair: &Ed25519KeyPair,
+        to: PublicKey,
+    ) -> Result<(), SendError<Event<u64>>> {
+        use log::{get_pubkey, sign_transaction_data};
+
+        let from = get_pubkey(keypair);
+        let sig = sign_transaction_data(&n, keypair, &to);
+        self.transfer_signed(from, to, n, sig)
+    }
+
+    pub fn get_balance(self: &mut Self, pubkey: &PublicKey) -> Result<u64, RecvError> {
+        self.sync();
+        Ok(*self.balances.get(pubkey).unwrap_or(&0))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::thread::sleep;
+    use std::time::Duration;
+    use log::{generate_keypair, get_pubkey};
+    use historian::ExitReason;
+
+    #[test]
+    fn test_accountant() {
+        let zero = Sha256Hash::default();
+        let mut acc = Accountant::new(&zero, Some(2));
+        let alice_keypair = generate_keypair();
+        let bob_keypair = generate_keypair();
+        acc.deposit(10_000, &alice_keypair).unwrap();
+        acc.deposit(1_000, &bob_keypair).unwrap();
+
+        sleep(Duration::from_millis(30));
+        let bob_pubkey = get_pubkey(&bob_keypair);
+        acc.transfer(500, &alice_keypair, bob_pubkey).unwrap();
+
+        sleep(Duration::from_millis(30));
+        assert_eq!(acc.get_balance(&bob_pubkey).unwrap(), 1_500);
+
+        drop(acc.historian.sender);
+        assert_eq!(
+            acc.historian.thread_hdl.join().unwrap().1,
+            ExitReason::RecvDisconnected
+        );
+    }
+
+    #[test]
+    fn test_invalid_transfer() {
+        let zero = Sha256Hash::default();
+        let mut acc = Accountant::new(&zero, Some(2));
+        let alice_keypair = generate_keypair();
+        let bob_keypair = generate_keypair();
+        acc.deposit(10_000, &alice_keypair).unwrap();
+        acc.deposit(1_000, &bob_keypair).unwrap();
+
+        sleep(Duration::from_millis(30));
+        let bob_pubkey = get_pubkey(&bob_keypair);
+        acc.transfer(10_001, &alice_keypair, bob_pubkey).unwrap();
+
+        sleep(Duration::from_millis(30));
+        let alice_pubkey = get_pubkey(&alice_keypair);
+        assert_eq!(acc.get_balance(&alice_pubkey).unwrap(), 10_000);
+        assert_eq!(acc.get_balance(&bob_pubkey).unwrap(), 1_000);
+
+        drop(acc.historian.sender);
+        assert_eq!(
+            acc.historian.thread_hdl.join().unwrap().1,
+            ExitReason::RecvDisconnected
+        );
+    }
+
+    #[test]
+    fn test_multiple_claims() {
+        let zero = Sha256Hash::default();
+        let mut acc = Accountant::new(&zero, Some(2));
+        let keypair = generate_keypair();
+        acc.deposit(1, &keypair).unwrap();
+        acc.deposit(2, &keypair).unwrap();
+
+        let pubkey = get_pubkey(&keypair);
+        sleep(Duration::from_millis(30));
+        assert_eq!(acc.get_balance(&pubkey).unwrap(), 3);
+
+        drop(acc.historian.sender);
+        assert_eq!(
+            acc.historian.thread_hdl.join().unwrap().1,
+            ExitReason::RecvDisconnected
+        );
+    }
+
+    #[test]
+    fn test_transfer_to_newb() {
+        let zero = Sha256Hash::default();
+        let mut acc = Accountant::new(&zero, Some(2));
+        let alice_keypair = generate_keypair();
+        let bob_keypair = generate_keypair();
+        acc.deposit(10_000, &alice_keypair).unwrap();
+
+        sleep(Duration::from_millis(30));
+        let bob_pubkey = get_pubkey(&bob_keypair);
+        acc.transfer(500, &alice_keypair, bob_pubkey).unwrap();
+
+        sleep(Duration::from_millis(30));
+        assert_eq!(acc.get_balance(&bob_pubkey).unwrap(), 500);
+
+        drop(acc.historian.sender);
+        assert_eq!(
+            acc.historian.thread_hdl.join().unwrap().1,
+            ExitReason::RecvDisconnected
+        );
+    }
+}

src/accountant_skel.rs

@@ -0,0 +1,75 @@
+use std::io;
+use accountant::Accountant;
+use log::{PublicKey, Signature};
+//use serde::Serialize;
+
+pub struct AccountantSkel {
+    pub obj: Accountant,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub enum Request {
+    Deposit {
+        key: PublicKey,
+        val: u64,
+        sig: Signature,
+    },
+    Transfer {
+        from: PublicKey,
+        to: PublicKey,
+        val: u64,
+        sig: Signature,
+    },
+    GetBalance {
+        key: PublicKey,
+    },
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub enum Response {
+    Balance { key: PublicKey, val: u64 },
+}
+
+impl AccountantSkel {
+    pub fn new(obj: Accountant) -> Self {
+        AccountantSkel { obj }
+    }
+
+    pub fn process_request(self: &mut Self, msg: Request) -> Option<Response> {
+        match msg {
+            Request::Deposit { key, val, sig } => {
+                let _ = self.obj.deposit_signed(key, val, sig);
+                None
+            }
+            Request::Transfer { from, to, val, sig } => {
+                let _ = self.obj.transfer_signed(from, to, val, sig);
+                None
+            }
+            Request::GetBalance { key } => {
+                let val = self.obj.get_balance(&key).unwrap();
+                Some(Response::Balance { key, val })
+            }
+        }
+    }
+
+    /// TCP Server that forwards messages to Accountant methods.
+    pub fn serve(self: &mut Self, addr: &str) -> io::Result<()> {
+        use std::net::TcpListener;
+        use std::io::{Read, Write};
+        use bincode::{deserialize, serialize};
+        let listener = TcpListener::bind(addr)?;
+        let mut buf = vec![0u8; 1024];
+        loop {
+            //println!("skel: Waiting for incoming connections...");
+            let (mut stream, _from_addr) = listener.accept()?;
+            let _sz = stream.read(&mut buf)?;
+
+            // TODO: Return a descriptive error message if deserialization fails.
+            let req = deserialize(&buf).expect("deserialize request");
+
+            if let Some(resp) = self.process_request(req) {
+                stream.write(&serialize(&resp).expect("serialize response"))?;
+            }
+        }
+    }
+}

src/accountant_stub.rs

@@ -0,0 +1,116 @@
+//! The `accountant` is a client of the `historian`. It uses the historian's
+//! event log to record transactions. Its users can deposit funds and
+//! transfer funds to other users.
+
+use std::net::TcpStream;
+use std::io;
+use std::io::{Read, Write};
+use bincode::{deserialize, serialize};
+use log::{PublicKey, Signature};
+use ring::signature::Ed25519KeyPair;
+use accountant_skel::{Request, Response};
+
+pub struct AccountantStub {
+    pub addr: String,
+}
+
+impl AccountantStub {
+    pub fn new(addr: &str) -> Self {
+        AccountantStub {
+            addr: addr.to_string(),
+        }
+    }
+
+    pub fn deposit_signed(
+        self: &mut Self,
+        key: PublicKey,
+        val: u64,
+        sig: Signature,
+    ) -> io::Result<usize> {
+        let req = Request::Deposit { key, val, sig };
+        let data = serialize(&req).unwrap();
+        let mut stream = TcpStream::connect(&self.addr)?;
+        stream.write(&data)
+    }
+
+    pub fn deposit(self: &mut Self, n: u64, keypair: &Ed25519KeyPair) -> io::Result<usize> {
+        use log::{get_pubkey, sign_serialized};
+        let key = get_pubkey(keypair);
+        let sig = sign_serialized(&n, keypair);
+        self.deposit_signed(key, n, sig)
+    }
+
+    pub fn transfer_signed(
+        self: &mut Self,
+        from: PublicKey,
+        to: PublicKey,
+        val: u64,
+        sig: Signature,
+    ) -> io::Result<usize> {
+        let req = Request::Transfer { from, to, val, sig };
+        let data = serialize(&req).unwrap();
+        let mut stream = TcpStream::connect(&self.addr)?;
+        stream.write(&data)
+    }
+
+    pub fn transfer(
+        self: &mut Self,
+        n: u64,
+        keypair: &Ed25519KeyPair,
+        to: PublicKey,
+    ) -> io::Result<usize> {
+        use log::{get_pubkey, sign_transaction_data};
+        let from = get_pubkey(keypair);
+        let sig = sign_transaction_data(&n, keypair, &to);
+        self.transfer_signed(from, to, n, sig)
+    }
+
+    pub fn get_balance(self: &mut Self, pubkey: &PublicKey) -> io::Result<u64> {
+        let mut stream = TcpStream::connect(&self.addr)?;
+        let req = Request::GetBalance { key: *pubkey };
+        let data = serialize(&req).expect("serialize GetBalance");
+        stream.write(&data)?;
+        let mut buf = vec![0u8; 1024];
+        stream.read(&mut buf)?;
+        let resp = deserialize(&buf).expect("deserialize balance");
+        let Response::Balance { key, val } = resp;
+        assert_eq!(key, *pubkey);
+        Ok(val)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use accountant::Accountant;
+    use accountant_skel::AccountantSkel;
+    use std::thread::{sleep, spawn};
+    use std::time::Duration;
+    use log::{generate_keypair, get_pubkey, Sha256Hash};
+
+    #[test]
+    fn test_accountant_stub() {
+        let addr = "127.0.0.1:8000";
+        spawn(move || {
+            let zero = Sha256Hash::default();
+            let acc = Accountant::new(&zero, None);
+            let mut skel = AccountantSkel::new(acc);
+            skel.serve(addr).unwrap();
+        });
+
+        sleep(Duration::from_millis(30));
+
+        let mut acc = AccountantStub::new(addr);
+        let alice_keypair = generate_keypair();
+        let bob_keypair = generate_keypair();
+        acc.deposit(10_000, &alice_keypair).unwrap();
+        acc.deposit(1_000, &bob_keypair).unwrap();
+
+        sleep(Duration::from_millis(30));
+        let bob_pubkey = get_pubkey(&bob_keypair);
+        acc.transfer(500, &alice_keypair, bob_pubkey).unwrap();
+
+        sleep(Duration::from_millis(300));
+        assert_eq!(acc.get_balance(&bob_pubkey).unwrap(), 1_500);
+    }
+}

src/bin/client-demo.rs

@@ -0,0 +1,45 @@
+extern crate silk;
+
+fn main() {
+    use silk::accountant_stub::AccountantStub;
+    use std::thread::sleep;
+    use std::time::Duration;
+    use silk::log::{generate_keypair, get_pubkey};
+
+    let addr = "127.0.0.1:8000";
+    let mut acc = AccountantStub::new(addr);
+    let alice_keypair = generate_keypair();
+    let bob_keypair = generate_keypair();
+    let txs = 10_000;
+    println!("Depositing {} units in Alice's account...", txs);
+    acc.deposit(txs, &alice_keypair).unwrap();
+    //acc.deposit(1_000, &bob_keypair).unwrap();
+    println!("Done.");
+
+    sleep(Duration::from_millis(30));
+    let alice_pubkey = get_pubkey(&alice_keypair);
+    let bob_pubkey = get_pubkey(&bob_keypair);
+    println!("Transferring 1 unit {} times...", txs);
+    for _ in 0..txs {
+        acc.transfer(1, &alice_keypair, bob_pubkey).unwrap();
+    }
+    println!("Done.");
+
+    sleep(Duration::from_millis(20));
+    let mut alice_val = acc.get_balance(&alice_pubkey).unwrap();
+    while alice_val > 0 {
+        println!("Checking on Alice's Balance {}", alice_val);
+        sleep(Duration::from_millis(20));
+        alice_val = acc.get_balance(&alice_pubkey).unwrap();
+    }
+    println!("Done. Checking balances.");
+    println!(
+        "Alice's Final Balance {}",
+        acc.get_balance(&alice_pubkey).unwrap()
+    );
+
+    println!(
+        "Bob's Final Balance {}",
+        acc.get_balance(&bob_pubkey).unwrap()
+    );
+}

src/bin/demo.rs

@@ -6,9 +6,10 @@ use std::thread::sleep;
 use std::time::Duration;
 use std::sync::mpsc::SendError;
 
-fn create_log(hist: &Historian) -> Result<(), SendError<Event>> {
+fn create_log(hist: &Historian<Sha256Hash>) -> Result<(), SendError<Event<Sha256Hash>>> {
     sleep(Duration::from_millis(15));
-    hist.sender.send(Event::UserDataKey(Sha256Hash::default()))?;
+    let data = Sha256Hash::default();
+    hist.sender.send(Event::Discovery { data })?;
     sleep(Duration::from_millis(10));
     Ok(())
 }
@@ -18,7 +19,7 @@ fn main() {
     let hist = Historian::new(&seed, Some(10));
     create_log(&hist).expect("send error");
     drop(hist.sender);
-    let entries: Vec<Entry> = hist.receiver.iter().collect();
+    let entries: Vec<Entry<Sha256Hash>> = hist.receiver.iter().collect();
     for entry in &entries {
         println!("{:?}", entry);
     }

src/bin/testnode.rs

@@ -0,0 +1,14 @@
+extern crate silk;
+
+use silk::accountant_skel::AccountantSkel;
+use silk::accountant::Accountant;
+use silk::log::Sha256Hash;
+
+fn main() {
+    let addr = "127.0.0.1:8000";
+    let zero = Sha256Hash::default();
+    let acc = Accountant::new(&zero, Some(1000));
+    let mut skel = AccountantSkel::new(acc);
+    println!("Listening on {}", addr);
+    skel.serve(addr).unwrap();
+}

src/historian.rs

@@ -6,14 +6,16 @@
 //! The resulting stream of entries represents ordered events in time.
 
 use std::thread::JoinHandle;
-use std::sync::mpsc::{Receiver, Sender};
+use std::sync::mpsc::{Receiver, SyncSender};
 use std::time::{Duration, SystemTime};
-use log::{extend_and_hash, hash, Entry, Event, Sha256Hash};
+use log::{hash, hash_event, verify_event, Entry, Event, Sha256Hash};
+use serde::Serialize;
+use std::fmt::Debug;
 
-pub struct Historian {
-    pub sender: Sender<Event>,
-    pub receiver: Receiver<Entry>,
-    pub thread_hdl: JoinHandle<(Entry, ExitReason)>,
+pub struct Historian<T> {
+    pub sender: SyncSender<Event<T>>,
+    pub receiver: Receiver<Entry<T>>,
+    pub thread_hdl: JoinHandle<(Entry<T>, ExitReason)>,
 }
 
 #[derive(Debug, PartialEq, Eq)]
@@ -21,15 +23,13 @@ pub enum ExitReason {
     RecvDisconnected,
     SendDisconnected,
 }
-fn log_event(
-    sender: &Sender<Entry>,
+fn log_event<T: Serialize + Clone + Debug>(
+    sender: &SyncSender<Entry<T>>,
     num_hashes: &mut u64,
     end_hash: &mut Sha256Hash,
-    event: Event,
-) -> Result<(), (Entry, ExitReason)> {
-    if let Event::UserDataKey(key) = event {
-        *end_hash = extend_and_hash(end_hash, &key);
-    }
+    event: Event<T>,
+) -> Result<(), (Entry<T>, ExitReason)> {
+    *end_hash = hash_event(end_hash, &event);
     let entry = Entry {
         end_hash: *end_hash,
         num_hashes: *num_hashes,
@@ -42,15 +42,15 @@ fn log_event(
     Ok(())
 }
 
-fn log_events(
-    receiver: &Receiver<Event>,
-    sender: &Sender<Entry>,
+fn log_events<T: Serialize + Clone + Debug>(
+    receiver: &Receiver<Event<T>>,
+    sender: &SyncSender<Entry<T>>,
     num_hashes: &mut u64,
     end_hash: &mut Sha256Hash,
     epoch: SystemTime,
     num_ticks: &mut u64,
     ms_per_tick: Option<u64>,
-) -> Result<(), (Entry, ExitReason)> {
+) -> Result<(), (Entry<T>, ExitReason)> {
     use std::sync::mpsc::TryRecvError;
     loop {
         if let Some(ms) = ms_per_tick {
@@ -62,7 +62,9 @@ fn log_events(
         }
         match receiver.try_recv() {
             Ok(event) => {
-                log_event(sender, num_hashes, end_hash, event)?;
+                if verify_event(&event) {
+                    log_event(sender, num_hashes, end_hash, event)?;
+                }
             }
             Err(TryRecvError::Empty) => {
                 return Ok(());
@@ -81,12 +83,12 @@ fn log_events(
 
 /// A background thread that will continue tagging received Event messages and
 /// sending back Entry messages until either the receiver or sender channel is closed.
-pub fn create_logger(
+pub fn create_logger<T: 'static + Serialize + Clone + Debug + Send>(
     start_hash: Sha256Hash,
     ms_per_tick: Option<u64>,
-    receiver: Receiver<Event>,
-    sender: Sender<Entry>,
-) -> JoinHandle<(Entry, ExitReason)> {
+    receiver: Receiver<Event<T>>,
+    sender: SyncSender<Entry<T>>,
+) -> JoinHandle<(Entry<T>, ExitReason)> {
     use std::thread;
     thread::spawn(move || {
         let mut end_hash = start_hash;
@@ -111,11 +113,11 @@ pub fn create_logger(
     })
 }
 
-impl Historian {
+impl<T: 'static + Serialize + Clone + Debug + Send> Historian<T> {
     pub fn new(start_hash: &Sha256Hash, ms_per_tick: Option<u64>) -> Self {
-        use std::sync::mpsc::channel;
-        let (sender, event_receiver) = channel();
-        let (entry_sender, receiver) = channel();
+        use std::sync::mpsc::sync_channel;
+        let (sender, event_receiver) = sync_channel(4000);
+        let (entry_sender, receiver) = sync_channel(4000);
         let thread_hdl = create_logger(*start_hash, ms_per_tick, event_receiver, entry_sender);
         Historian {
             sender,
@@ -139,7 +141,7 @@ mod tests {
 
         hist.sender.send(Event::Tick).unwrap();
         sleep(Duration::new(0, 1_000_000));
-        hist.sender.send(Event::UserDataKey(zero)).unwrap();
+        hist.sender.send(Event::Discovery { data: zero }).unwrap();
         sleep(Duration::new(0, 1_000_000));
         hist.sender.send(Event::Tick).unwrap();
 
@@ -159,7 +161,7 @@ mod tests {
     #[test]
     fn test_historian_closed_sender() {
         let zero = Sha256Hash::default();
-        let hist = Historian::new(&zero, None);
+        let hist = Historian::<u8>::new(&zero, None);
         drop(hist.receiver);
         hist.sender.send(Event::Tick).unwrap();
         assert_eq!(
@@ -173,7 +175,7 @@ mod tests {
         let zero = Sha256Hash::default();
         let hist = Historian::new(&zero, Some(20));
         sleep(Duration::from_millis(30));
-        hist.sender.send(Event::UserDataKey(zero)).unwrap();
+        hist.sender.send(Event::Discovery { data: zero }).unwrap();
         sleep(Duration::from_millis(15));
         drop(hist.sender);
         assert_eq!(
@@ -181,8 +183,28 @@ mod tests {
             ExitReason::RecvDisconnected
         );
 
-        let entries: Vec<Entry> = hist.receiver.iter().collect();
+        let entries: Vec<Entry<Sha256Hash>> = hist.receiver.iter().collect();
         assert!(entries.len() > 1);
         assert!(verify_slice(&entries, &zero));
     }
+
+    #[test]
+    fn test_bad_event_attack() {
+        let zero = Sha256Hash::default();
+        let hist = Historian::new(&zero, None);
+        let keypair = generate_keypair();
+        let event0 = Event::Claim {
+            key: get_pubkey(&keypair),
+            data: hash(b"goodbye cruel world"),
+            sig: sign_serialized(&hash(b"hello, world"), &keypair),
+        };
+        hist.sender.send(event0).unwrap();
+        drop(hist.sender);
+        assert_eq!(
+            hist.thread_hdl.join().unwrap().1,
+            ExitReason::RecvDisconnected
+        );
+        let entries: Vec<Entry<Sha256Hash>> = hist.receiver.iter().collect();
+        assert_eq!(entries.len(), 0);
+    }
 }

src/lib.rs

@@ -1,9 +1,15 @@
 #![cfg_attr(feature = "unstable", feature(test))]
 pub mod log;
 pub mod historian;
+pub mod accountant;
+pub mod accountant_skel;
+pub mod accountant_stub;
+extern crate bincode;
 extern crate generic_array;
 extern crate rayon;
+extern crate ring;
 extern crate serde;
 #[macro_use]
 extern crate serde_derive;
 extern crate sha2;
+extern crate untrusted;

src/log.rs

@@ -14,14 +14,19 @@
 /// was generated by the fastest processor at the time the entry was logged.
 
 use generic_array::GenericArray;
-use generic_array::typenum::U32;
+use generic_array::typenum::{U32, U64};
+use ring::signature::Ed25519KeyPair;
+use serde::Serialize;
+
 pub type Sha256Hash = GenericArray<u8, U32>;
+pub type PublicKey = GenericArray<u8, U32>;
+pub type Signature = GenericArray<u8, U64>;
 
 #[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)]
-pub struct Entry {
+pub struct Entry<T> {
     pub num_hashes: u64,
     pub end_hash: Sha256Hash,
-    pub event: Event,
+    pub event: Event<T>,
 }
 
 /// When 'event' is Tick, the event represents a simple clock tick, and exists for the
@@ -30,12 +35,25 @@ pub struct Entry {
 /// a hash alongside the tick, each tick and be verified in parallel using the 'end_hash'
 /// of the preceding tick to seed its hashing.
 #[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)]
-pub enum Event {
+pub enum Event<T> {
     Tick,
-    UserDataKey(Sha256Hash),
+    Discovery {
+        data: T,
+    },
+    Claim {
+        key: PublicKey,
+        data: T,
+        sig: Signature,
+    },
+    Transaction {
+        from: PublicKey,
+        to: PublicKey,
+        data: T,
+        sig: Signature,
+    },
 }
 
-impl Entry {
+impl<T> Entry<T> {
     /// Creates a Entry from the number of hashes 'num_hashes' since the previous event
     /// and that resulting 'end_hash'.
     pub fn new_tick(num_hashes: u64, end_hash: &Sha256Hash) -> Self {
@@ -45,14 +63,39 @@ impl Entry {
             event: Event::Tick,
         }
     }
-
-    /// Verifies self.end_hash is the result of hashing a 'start_hash' 'self.num_hashes' times.
-    /// If the event is a UserDataKey, then hash that as well.
-    pub fn verify(self: &Self, start_hash: &Sha256Hash) -> bool {
-        self.end_hash == next_hash(start_hash, self.num_hashes, &self.event)
-    }
 }
 
+/// Return a new ED25519 keypair
+pub fn generate_keypair() -> Ed25519KeyPair {
+    use ring::{rand, signature};
+    use untrusted;
+    let rng = rand::SystemRandom::new();
+    let pkcs8_bytes = signature::Ed25519KeyPair::generate_pkcs8(&rng).unwrap();
+    signature::Ed25519KeyPair::from_pkcs8(untrusted::Input::from(&pkcs8_bytes)).unwrap()
+}
+
+/// Return the public key for the given keypair
+pub fn get_pubkey(keypair: &Ed25519KeyPair) -> PublicKey {
+    GenericArray::clone_from_slice(keypair.public_key_bytes())
+}
+
+/// Return a signature for the given data using the private key from the given keypair.
+pub fn sign_serialized<T: Serialize>(data: &T, keypair: &Ed25519KeyPair) -> Signature {
+    use bincode::serialize;
+    let serialized = serialize(data).unwrap();
+    GenericArray::clone_from_slice(keypair.sign(&serialized).as_ref())
+}
+
+/// Return a signature for the given transaction data using the private key from the given keypair.
+pub fn sign_transaction_data<T: Serialize>(
+    data: &T,
+    keypair: &Ed25519KeyPair,
+    to: &PublicKey,
+) -> Signature {
+    sign_serialized(&(data, to), keypair)
+}
+
+/// Return a Sha256 hash for the given data.
 pub fn hash(val: &[u8]) -> Sha256Hash {
     use sha2::{Digest, Sha256};
     let mut hasher = Sha256::default();
@@ -61,25 +104,58 @@ pub fn hash(val: &[u8]) -> Sha256Hash {
 }
 
 /// Return the hash of the given hash extended with the given value.
-pub fn extend_and_hash(end_hash: &Sha256Hash, val: &[u8]) -> Sha256Hash {
+pub fn extend_and_hash(end_hash: &Sha256Hash, ty: u8, val: &[u8]) -> Sha256Hash {
     let mut hash_data = end_hash.to_vec();
+    hash_data.push(ty);
     hash_data.extend_from_slice(val);
     hash(&hash_data)
 }
 
-pub fn next_hash(start_hash: &Sha256Hash, num_hashes: u64, event: &Event) -> Sha256Hash {
+pub fn hash_event<T: Serialize>(end_hash: &Sha256Hash, event: &Event<T>) -> Sha256Hash {
+    use bincode::serialize;
+    match *event {
+        Event::Tick => *end_hash,
+        Event::Discovery { ref data } => extend_and_hash(end_hash, 1, &serialize(&data).unwrap()),
+        Event::Claim { key, ref data, sig } => {
+            let mut event_data = serialize(&data).unwrap();
+            event_data.extend_from_slice(&sig);
+            event_data.extend_from_slice(&key);
+            extend_and_hash(end_hash, 2, &event_data)
+        }
+        Event::Transaction {
+            from,
+            to,
+            ref data,
+            sig,
+        } => {
+            let mut event_data = serialize(&data).unwrap();
+            event_data.extend_from_slice(&sig);
+            event_data.extend_from_slice(&from);
+            event_data.extend_from_slice(&to);
+            extend_and_hash(end_hash, 2, &event_data)
+        }
+    }
+}
+
+/// Creates the hash 'num_hashes' after start_hash, plus an additional hash for any event data.
+pub fn next_hash<T: Serialize>(
+    start_hash: &Sha256Hash,
+    num_hashes: u64,
+    event: &Event<T>,
+) -> Sha256Hash {
     let mut end_hash = *start_hash;
     for _ in 0..num_hashes {
         end_hash = hash(&end_hash);
     }
-    if let Event::UserDataKey(key) = *event {
-        return extend_and_hash(&end_hash, &key);
-    }
-    end_hash
+    hash_event(&end_hash, event)
 }
 
 /// Creates the next Tick Entry 'num_hashes' after 'start_hash'.
-pub fn next_entry(start_hash: &Sha256Hash, num_hashes: u64, event: Event) -> Entry {
+pub fn next_entry<T: Serialize>(
+    start_hash: &Sha256Hash,
+    num_hashes: u64,
+    event: Event<T>,
+) -> Entry<T> {
     Entry {
         num_hashes,
         end_hash: next_hash(start_hash, num_hashes, &event),
@@ -88,36 +164,109 @@ pub fn next_entry(start_hash: &Sha256Hash, num_hashes: u64, event: Event) -> Ent
 }
 
 /// Creates the next Tick Entry 'num_hashes' after 'start_hash'.
-pub fn next_tick(start_hash: &Sha256Hash, num_hashes: u64) -> Entry {
+pub fn next_entry_mut<T: Serialize>(
+    start_hash: &mut Sha256Hash,
+    num_hashes: u64,
+    event: Event<T>,
+) -> Entry<T> {
+    let entry = next_entry(start_hash, num_hashes, event);
+    *start_hash = entry.end_hash;
+    entry
+}
+
+/// Creates the next Tick Entry 'num_hashes' after 'start_hash'.
+pub fn next_tick<T: Serialize>(start_hash: &Sha256Hash, num_hashes: u64) -> Entry<T> {
     next_entry(start_hash, num_hashes, Event::Tick)
 }
 
+pub fn verify_event<T: Serialize>(event: &Event<T>) -> bool {
+    use bincode::serialize;
+    if let Event::Claim { key, ref data, sig } = *event {
+        let mut claim_data = serialize(&data).unwrap();
+        if !verify_signature(&key, &claim_data, &sig) {
+            return false;
+        }
+    }
+    if let Event::Transaction {
+        from,
+        to,
+        ref data,
+        sig,
+    } = *event
+    {
+        let sign_data = serialize(&(&data, &to)).unwrap();
+        if !verify_signature(&from, &sign_data, &sig) {
+            return false;
+        }
+    }
+    true
+}
+
+/// Verifies self.end_hash is the result of hashing a 'start_hash' 'self.num_hashes' times.
+/// If the event is not a Tick, then hash that as well.
+pub fn verify_entry<T: Serialize>(entry: &Entry<T>, start_hash: &Sha256Hash) -> bool {
+    if !verify_event(&entry.event) {
+        return false;
+    }
+    entry.end_hash == next_hash(start_hash, entry.num_hashes, &entry.event)
+}
+
 /// Verifies the hashes and counts of a slice of events are all consistent.
-pub fn verify_slice(events: &[Entry], start_hash: &Sha256Hash) -> bool {
+pub fn verify_slice(events: &[Entry<Sha256Hash>], start_hash: &Sha256Hash) -> bool {
     use rayon::prelude::*;
     let genesis = [Entry::new_tick(Default::default(), start_hash)];
     let event_pairs = genesis.par_iter().chain(events).zip(events);
-    event_pairs.all(|(x0, x1)| x1.verify(&x0.end_hash))
+    event_pairs.all(|(x0, x1)| verify_entry(&x1, &x0.end_hash))
+}
+
+/// Verifies the hashes and counts of a slice of events are all consistent.
+pub fn verify_slice_u64(events: &[Entry<u64>], start_hash: &Sha256Hash) -> bool {
+    use rayon::prelude::*;
+    let genesis = [Entry::new_tick(Default::default(), start_hash)];
+    let event_pairs = genesis.par_iter().chain(events).zip(events);
+    event_pairs.all(|(x0, x1)| verify_entry(&x1, &x0.end_hash))
 }
 
 /// Verifies the hashes and events serially. Exists only for reference.
-pub fn verify_slice_seq(events: &[Entry], start_hash: &Sha256Hash) -> bool {
+pub fn verify_slice_seq<T: Serialize>(events: &[Entry<T>], start_hash: &Sha256Hash) -> bool {
     let genesis = [Entry::new_tick(0, start_hash)];
     let mut event_pairs = genesis.iter().chain(events).zip(events);
-    event_pairs.all(|(x0, x1)| x1.verify(&x0.end_hash))
+    event_pairs.all(|(x0, x1)| verify_entry(&x1, &x0.end_hash))
+}
+
+/// Verify a signed message with the given public key.
+pub fn verify_signature(peer_public_key_bytes: &[u8], msg_bytes: &[u8], sig_bytes: &[u8]) -> bool {
+    use untrusted;
+    use ring::signature;
+    let peer_public_key = untrusted::Input::from(peer_public_key_bytes);
+    let msg = untrusted::Input::from(msg_bytes);
+    let sig = untrusted::Input::from(sig_bytes);
+    signature::verify(&signature::ED25519, peer_public_key, msg, sig).is_ok()
+}
+
+pub fn create_entries<T: Serialize>(
+    start_hash: &Sha256Hash,
+    num_hashes: u64,
+    events: Vec<Event<T>>,
+) -> Vec<Entry<T>> {
+    let mut end_hash = *start_hash;
+    events
+        .into_iter()
+        .map(|event| next_entry_mut(&mut end_hash, num_hashes, event))
+        .collect()
 }
 
 /// Create a vector of Ticks of length 'len' from 'start_hash' hash and 'num_hashes'.
-pub fn create_ticks(start_hash: &Sha256Hash, num_hashes: u64, len: usize) -> Vec<Entry> {
+pub fn create_ticks(
+    start_hash: &Sha256Hash,
+    num_hashes: u64,
+    len: usize,
+) -> Vec<Entry<Sha256Hash>> {
     use std::iter;
     let mut end_hash = *start_hash;
     iter::repeat(Event::Tick)
         .take(len)
-        .map(|event| {
-            let entry = next_entry(&end_hash, num_hashes, event);
-            end_hash = entry.end_hash;
-            entry
-        })
+        .map(|event| next_entry_mut(&mut end_hash, num_hashes, event))
         .collect()
 }
 
@@ -129,19 +278,19 @@ mod tests {
     fn test_event_verify() {
         let zero = Sha256Hash::default();
         let one = hash(&zero);
-        assert!(Entry::new_tick(0, &zero).verify(&zero)); // base case
-        assert!(!Entry::new_tick(0, &zero).verify(&one)); // base case, bad
-        assert!(next_tick(&zero, 1).verify(&zero)); // inductive step
-        assert!(!next_tick(&zero, 1).verify(&one)); // inductive step, bad
+        assert!(verify_entry::<u8>(&Entry::new_tick(0, &zero), &zero)); // base case
+        assert!(!verify_entry::<u8>(&Entry::new_tick(0, &zero), &one)); // base case, bad
+        assert!(verify_entry::<u8>(&next_tick(&zero, 1), &zero)); // inductive step
+        assert!(!verify_entry::<u8>(&next_tick(&zero, 1), &one)); // inductive step, bad
     }
 
     #[test]
     fn test_next_tick() {
         let zero = Sha256Hash::default();
-        assert_eq!(next_tick(&zero, 1).num_hashes, 1)
+        assert_eq!(next_tick::<Sha256Hash>(&zero, 1).num_hashes, 1)
     }
 
-    fn verify_slice_generic(verify_slice: fn(&[Entry], &Sha256Hash) -> bool) {
+    fn verify_slice_generic(verify_slice: fn(&[Entry<Sha256Hash>], &Sha256Hash) -> bool) {
         let zero = Sha256Hash::default();
         let one = hash(&zero);
         assert!(verify_slice(&vec![], &zero)); // base case
@@ -161,7 +310,7 @@ mod tests {
 
     #[test]
     fn test_verify_slice_seq() {
-        verify_slice_generic(verify_slice_seq);
+        verify_slice_generic(verify_slice_seq::<Sha256Hash>);
     }
 
     #[test]
@@ -169,27 +318,100 @@ mod tests {
         let zero = Sha256Hash::default();
         let one = hash(&zero);
 
-        // First, verify UserData events
-        let mut end_hash = zero;
-        let events = [Event::UserDataKey(zero), Event::UserDataKey(one)];
-        let mut entries: Vec<Entry> = events
-            .iter()
-            .map(|event| {
-                let entry = next_entry(&end_hash, 0, event.clone());
-                end_hash = entry.end_hash;
-                entry
-            })
-            .collect();
-        assert!(verify_slice(&entries, &zero)); // inductive step
+        // First, verify Discovery events
+        let events = vec![
+            Event::Discovery { data: zero },
+            Event::Discovery { data: one },
+        ];
+        let mut entries = create_entries(&zero, 0, events);
+        assert!(verify_slice(&entries, &zero));
 
-        // Next, swap only two UserData events and ensure verification fails.
+        // Next, swap two Discovery events and ensure verification fails.
         let event0 = entries[0].event.clone();
         let event1 = entries[1].event.clone();
         entries[0].event = event1;
         entries[1].event = event0;
-        assert!(!verify_slice(&entries, &zero)); // inductive step
+        assert!(!verify_slice(&entries, &zero));
     }
 
+    #[test]
+    fn test_claim() {
+        let keypair = generate_keypair();
+        let data = hash(b"hello, world");
+        let event0 = Event::Claim {
+            key: get_pubkey(&keypair),
+            data,
+            sig: sign_serialized(&data, &keypair),
+        };
+        let zero = Sha256Hash::default();
+        let entries = create_entries(&zero, 0, vec![event0]);
+        assert!(verify_slice(&entries, &zero));
+    }
+
+    #[test]
+    fn test_wrong_data_claim_attack() {
+        let keypair = generate_keypair();
+        let event0 = Event::Claim {
+            key: get_pubkey(&keypair),
+            data: hash(b"goodbye cruel world"),
+            sig: sign_serialized(&hash(b"hello, world"), &keypair),
+        };
+        let zero = Sha256Hash::default();
+        let entries = create_entries(&zero, 0, vec![event0]);
+        assert!(!verify_slice(&entries, &zero));
+    }
+
+    #[test]
+    fn test_transfer() {
+        let keypair0 = generate_keypair();
+        let keypair1 = generate_keypair();
+        let pubkey1 = get_pubkey(&keypair1);
+        let data = hash(b"hello, world");
+        let event0 = Event::Transaction {
+            from: get_pubkey(&keypair0),
+            to: pubkey1,
+            data,
+            sig: sign_transaction_data(&data, &keypair0, &pubkey1),
+        };
+        let zero = Sha256Hash::default();
+        let entries = create_entries(&zero, 0, vec![event0]);
+        assert!(verify_slice(&entries, &zero));
+    }
+
+    #[test]
+    fn test_wrong_data_transfer_attack() {
+        let keypair0 = generate_keypair();
+        let keypair1 = generate_keypair();
+        let pubkey1 = get_pubkey(&keypair1);
+        let data = hash(b"hello, world");
+        let event0 = Event::Transaction {
+            from: get_pubkey(&keypair0),
+            to: pubkey1,
+            data: hash(b"goodbye cruel world"), // <-- attack!
+            sig: sign_transaction_data(&data, &keypair0, &pubkey1),
+        };
+        let zero = Sha256Hash::default();
+        let entries = create_entries(&zero, 0, vec![event0]);
+        assert!(!verify_slice(&entries, &zero));
+    }
+
+    #[test]
+    fn test_transfer_hijack_attack() {
+        let keypair0 = generate_keypair();
+        let keypair1 = generate_keypair();
+        let thief_keypair = generate_keypair();
+        let pubkey1 = get_pubkey(&keypair1);
+        let data = hash(b"hello, world");
+        let event0 = Event::Transaction {
+            from: get_pubkey(&keypair0),
+            to: get_pubkey(&thief_keypair), // <-- attack!
+            data: hash(b"goodbye cruel world"),
+            sig: sign_transaction_data(&data, &keypair0, &pubkey1),
+        };
+        let zero = Sha256Hash::default();
+        let entries = create_entries(&zero, 0, vec![event0]);
+        assert!(!verify_slice(&entries, &zero));
+    }
 }
 
 #[cfg(all(feature = "unstable", test))]