PeachCloud
/
kuska-ssb
1
1
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

new handshake api

This commit is contained in:
adria0 2020-01-21 13:34:58 +01:00
parent 4e795b91cf
commit 11f767e3f5
33 changed files with 952 additions and 1280 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
Cargo.toml
View File

@ -1,6 +1,6 @@
[package]
name = "kuska-ssb"
version = "0.1.0"
version = "0.1.2"
authors = ["Dhole <dhole@riseup.net>", "Adria Massanet <adria@codecontext.io>"]
edition = "2018"
@ -8,27 +8,18 @@ edition = "2018"
name = "kuska_ssb"
[dependencies]
kuska-handshake = { git = "https://github.com/Kuska-ssb/kuska-handshake", branch = "feature/moderror" , features=["sync","async_std"] }
kuska-handshake = { git = "https://github.com/Kuska-ssb/kuska-handshake", branch = "master" , features=["sync","async_std"] }
sodiumoxide = { git = "https://github.com/Dhole/sodiumoxidez", branch = "extra" }
base64 = "0.11.0"
hex = "0.4.0"
async-std = { version = "1.1.0", features=["unstable","attributes"] }
async-std = { version = "1.4.0", features=["unstable","attributes"] }
crossbeam = "0.7.3"
log = "0.4.8"
env_logger = "0.7.1"
serde = "1.0.104"
serde_json = { version = "1.0.41", features=["preserve_order","arbitrary_precision"] }
serde_derive = "1.0.104"
serde = { version = "1.0.104", features = ["derive"] }
serde_json = { version = "1.0.48", features=["preserve_order","arbitrary_precision"] }
dirs = "2.0"
futures = "0.3.1"
snap = "0.2.5"
signal-hook = "0.1.12"
futures = "0.3.4"
lazy_static = "1.4.0"
rand = "0.7.2"
[dev-dependencies]
kuska-handshake = { git = "https://github.com/Kuska-ssb/kuska-handshake", branch = "feature/moderror", features=["sync","async_std","tokio_compat"] }
tokio = { version = "0.2.6", features=["full"] }
actix-rt = "1.0.0"
actix-web = "2.0.0"
rand = "0.7.3"

224
examples/iot-solar-wip.rs
View File

@ -1,224 +0,0 @@
#![allow(dead_code)]
#[macro_use]
extern crate lazy_static;
use async_std::sync::{Receiver,Sender,channel};
use async_std::task;
use std::fmt::Debug;
use std::cell::{RefCell};
use std::sync::{Arc,Mutex};
use std::time::Duration;
use signal_hook::{iterator::Signals, SIGTERM, SIGINT, SIGHUP, SIGQUIT};
use actix_web::{get, web, App, HttpServer, Responder};
use async_std::io::{Read,Write};
use kuska_handshake::async_std::{BoxStream,handshake_client,TokioCompatExt,TokioCompatExtRead,TokioCompatExtWrite};
use kuska_ssb::rpc::{Header,RequestNo,RpcClient};
use kuska_ssb::patchwork::*;
use tokio::net::TcpStream;
type AnyResult<T> = std::result::Result<T, Box<dyn std::error::Error>>;
async fn get_async<'a,R,W,T,F> (client: &mut ApiClient<R,W>, req_no : RequestNo, f : F) -> AnyResult<T>
where
R: Read+Unpin,
W: Write+Unpin,
F: Fn(&Header,&[u8])->Result<T>,
T: Debug
{
loop {
let (header,body) = client.rpc().recv().await?;
if header.req_no == req_no {
return f(&header,&body).map_err(|err| err.into());
}
}
}
async fn run_task<R:Read+Unpin,W:Write+Unpin>(api : &mut ApiClient<R,W>, _command: &str) -> AnyResult<bool> {
let req_id = api.send_whoami().await?;
let whoami = get_async(api,-req_id,parse_whoami).await?.id;
println!("{}",whoami);
Ok(false)
}
async fn sync_loop(command_receiver: Receiver<String>, stop_receiver : Receiver<bool>) -> AnyResult<()>{
/*
sync loop functionality
friend_list = []
discovered_peers = []
connected_peers = []
loop {
foreach peer in discoverd_peer not in conected_peers {
peer.handshake.createboxstream
peer.feed_callback = { |msg|
process_feed_message(msg)
}
for each friend in friend_list {
peer.createUserStream(friend)
}
conected_peers.push(peer)
}
}
*/
let IdentitySecret{pk,sk,..} = IdentitySecret::from_local_config()
.expect("read local secret");
let tokio_socket : TcpStream = TcpStream::connect("127.0.0.1:8008").await?;
let asyncstd_socket = TokioCompatExt::wrap(tokio_socket);
let (asyncstd_socket,handshake) = handshake_client(asyncstd_socket, ssb_net_id(), pk, sk.clone(), pk).await?;
let mut tokio_socket = asyncstd_socket.into_inner();
let (read,write) = tokio_socket.split();
let read = TokioCompatExtRead::wrap(read);
let write = TokioCompatExtWrite::wrap(write);
let (box_stream_read, box_stream_write) =
BoxStream::from_handhake(read, write, handshake, 0x8000)
.split_read_write();
let rpc = RpcClient::new(box_stream_read, box_stream_write);
let mut api = ApiClient::new(rpc);
let mut commands_queue : Vec<String> = Vec::new();
loop {
if !stop_receiver.is_empty() {
stop_receiver.recv().await;
println!("finished loop");
return Ok(());
}
// read all pending requests
while !command_receiver.is_empty() {
if let Some(msg) = command_receiver.recv().await {
commands_queue.push(msg);
}
}
if let Some(command) = commands_queue.pop() {
run_task(&mut api,&command).await?;
} else {
task::sleep(Duration::from_secs(1)).await;
println!("waiting!");
}
}
}
async fn command_loop(command_receiver: Receiver<String>, stop_receiver : Receiver<bool>) -> AnyResult<()>{
let IdentitySecret{pk,sk,..} = IdentitySecret::from_local_config()
.expect("read local secret");
let tokio_socket : TcpStream = TcpStream::connect("127.0.0.1:8008").await?;
let asyncstd_socket = TokioCompatExt::wrap(tokio_socket);
let (asyncstd_socket,handshake) = handshake_client(asyncstd_socket, ssb_net_id(), pk, sk.clone(), pk).await?;
println!("💃 handshake complete");
let mut tokio_socket = asyncstd_socket.into_inner();
let (read,write) = tokio_socket.split();
let read = TokioCompatExtRead::wrap(read);
let write = TokioCompatExtWrite::wrap(write);
let (box_stream_read, box_stream_write) =
BoxStream::from_handhake(read, write, handshake, 0x8000)
.split_read_write();
let rpc = RpcClient::new(box_stream_read, box_stream_write);
let mut api = ApiClient::new(rpc);
let mut commands_queue : Vec<String> = Vec::new();
loop {
if !stop_receiver.is_empty() {
stop_receiver.recv().await;
println!("finished loop");
return Ok(());
}
// read all pending requests
while !command_receiver.is_empty() {
if let Some(msg) = command_receiver.recv().await {
commands_queue.push(msg);
}
}
if let Some(command) = commands_queue.pop() {
run_task(&mut api,&command).await?;
} else {
task::sleep(Duration::from_secs(1)).await;
println!("waiting!");
}
}
}
lazy_static! {
static ref COMMAND_SENDER : Arc<Mutex<RefCell<Option<Sender<String>>>>> = Arc::new(Mutex::new(RefCell::new(None)));
}
#[get("/{id}/{name}/index.html")]
async fn index(info: web::Path<(u32, String)>) -> impl Responder {
COMMAND_SENDER.lock().unwrap().borrow().as_ref().unwrap().send("hola".to_owned()).await;
format!("Hello {}! id:{}", info.1, info.0)
}
async fn web_handler() -> std::io::Result<()> {
HttpServer::new(|| App::new().service(index))
.bind("127.0.0.1:8080")?
.run()
.await
}
async fn sigterm_handler(stop_sender : Sender<bool>, count : usize, ) {
let signals = Signals::new(&[SIGTERM, SIGHUP, SIGINT, SIGQUIT]).expect("cannot capture SIGTERM");
loop {
if signals.pending().next().is_some() {
for _ in 0..count {
stop_sender.send(true).await;
}
return;
}
task::sleep(Duration::from_secs(1)).await;
}
}
#[actix_rt::main]
async fn main() {
println!("started");
let (stop_sender, stop_receiver) = channel::<bool>(1);
let (command_sender, command_receiver) = channel::<String>(1);
COMMAND_SENDER.lock().unwrap().replace(Some(command_sender));
let future_sigterm = sigterm_handler(stop_sender,1);
let future_loop = command_loop(command_receiver,stop_receiver.clone());
let future_web = web_handler();
let (_,loop_res,_) = futures::join!(future_sigterm,future_loop,future_web);
loop_res.expect("main loop failed");
}

201
examples/ssb-cli.rs
View File

@ -6,52 +6,113 @@ extern crate crossbeam;
use std::fmt::Debug;
use async_std::io::{Read,Write};
use async_std::io::{Read, Write};
use async_std::net::TcpStream;
use kuska_handshake::async_std::{handshake_client,BoxStream};
use kuska_ssb::rpc::{Header,RequestNo,RpcClient};
use kuska_ssb::patchwork::*;
use kuska_ssb::feed::{is_privatebox,privatebox_decipher};
use kuska_handshake::async_std::{handshake_client, BoxStream};
use kuska_ssb::api::{
ApiHelper, CreateHistoryStreamArgs, CreateStreamArgs, LatestUserMessage, WhoAmI,
};
use kuska_ssb::discovery::ssb_net_id;
use kuska_ssb::feed::{is_privatebox, privatebox_decipher, Feed, Message};
use kuska_ssb::keystore::from_patchwork_local;
use kuska_ssb::keystore::OwnedIdentity;
use kuska_ssb::rpc::{RecvMsg, RequestNo, RpcStream};
type AnyResult<T> = std::result::Result<T, Box<dyn std::error::Error>>;
async fn get_async<'a,R,W,T,F> (client: &mut ApiClient<R,W>, req_no : RequestNo, f : F) -> AnyResult<T>
pub fn whoami_res_parse(body: &[u8]) -> AnyResult<WhoAmI> {
Ok(serde_json::from_slice(body)?)
}
pub fn message_res_parse(body: &[u8]) -> AnyResult<Message> {
Ok(Message::from_slice(body)?)
}
pub fn feed_res_parse(body: &[u8]) -> AnyResult<Feed> {
Ok(Feed::from_slice(&body)?)
}
pub fn latest_res_parse(body: &[u8]) -> AnyResult<LatestUserMessage> {
Ok(serde_json::from_slice(body)?)
}
#[derive(Debug)]
struct AppError {
message: String,
}
impl AppError {
pub fn new(message: String) -> Self {
AppError { message }
}
}
impl std::error::Error for AppError {
fn description(&self) -> &str {
&self.message
}
}
impl std::fmt::Display for AppError {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
write!(f, "{}", self.message)
}
}
async fn get_async<'a, R, W, T, F>(
client: &mut ApiHelper<R, W>,
req_no: RequestNo,
f: F,
) -> AnyResult<T>
where
R: Read+Unpin,
W: Write+Unpin,
F: Fn(&Header,&[u8])->Result<T>,
T: Debug
R: Read + Unpin,
W: Write + Unpin,
F: Fn(&[u8]) -> AnyResult<T>,
T: Debug,
{
loop {
let (header,body) = client.rpc().recv().await?;
if header.req_no == req_no {
return f(&header,&body).map_err(|err| err.into());
let (id, msg) = client.rpc().recv().await?;
if id == req_no {
match msg {
RecvMsg::BodyResponse(body) => {
return f(&body).map_err(|err| err.into());
}
RecvMsg::ErrorResponse(message) => {
return std::result::Result::Err(Box::new(AppError::new(message)));
}
_ => unreachable!(),
}
} else {
println!("discarded message {}", id);
}
}
}
async fn print_source_until_eof<'a,R,W,T,F> (client: &mut ApiClient<R,W>, req_no : RequestNo, f : F) -> AnyResult<()>
async fn print_source_until_eof<'a, R, W, T, F>(
client: &mut ApiHelper<R, W>,
req_no: RequestNo,
f: F,
) -> AnyResult<()>
where
R: Read+Unpin,
W: Write+Unpin,
F: Fn(&Header,&[u8])->Result<T>,
T: Debug+serde::Deserialize<'a>
R: Read + Unpin,
W: Write + Unpin,
F: Fn(&[u8]) -> AnyResult<T>,
T: Debug + serde::Deserialize<'a>,
{
loop {
let (header,body) = client.rpc().recv().await?;
if header.req_no == req_no {
if !header.is_end_or_error {
match f(&header,&body) {
Ok(res) => { println!("{:?}",res) },
Err(err) => println!(" 😢 Failed :( {:?} {}",err,String::from_utf8_lossy(&body)),
let (id, msg) = client.rpc().recv().await?;
if id == req_no {
match msg {
RecvMsg::BodyResponse(body) => {
let display = f(&body)?;
println!("{:?}", display);
}
} else {
println!("STREAM FINISHED");
return Ok(())
RecvMsg::ErrorResponse(message) => {
return std::result::Result::Err(Box::new(AppError::new(message)));
}
RecvMsg::CancelStreamRespose() => break,
_ => unreachable!(),
}
} else {
println!("discarded message {}", id);
}
}
Ok(())
}
#[async_std::main]
@ -59,97 +120,89 @@ async fn main() -> AnyResult<()> {
env_logger::init();
log::set_max_level(log::LevelFilter::max());
let IdentitySecret{pk,sk,..} = IdentitySecret::from_local_config()
.expect("read local secret");
let OwnedIdentity { pk, sk, .. } = from_patchwork_local().expect("read local secret");
let mut socket = TcpStream::connect("127.0.0.1:8008").await?;
let (_,handshake) = handshake_client(&mut socket, ssb_net_id(), pk, sk.clone(), pk).await?;
let mut socket = TcpStream::connect("127.0.0.1:8080").await?;
let handshake = handshake_client(&mut socket, ssb_net_id(), pk, sk.clone(), pk).await?;
println!("💃 handshake complete");
let (box_stream_read, box_stream_write) =
BoxStream::from_handhake(&socket, &socket, handshake, 0x8000)
.split_read_write();
BoxStream::from_handshake(&socket, &socket, handshake, 0x8000).split_read_write();
let mut client = ApiClient::new(RpcClient::new(box_stream_read, box_stream_write));
let mut client = ApiHelper::new(RpcStream::new(box_stream_read, box_stream_write));
let req_id = client.send_whoami().await?;
let whoami = get_async(&mut client,-req_id,parse_whoami).await?.id;
let req_id = client.whoami_req_send().await?;
let whoami = get_async(&mut client, req_id, whoami_res_parse).await?.id;
println!("😊 server says hello to {}",whoami);
println!("😊 server says hello to {}", whoami);
let mut line_buffer = String::new();
while let Ok(_) = std::io::stdin().read_line(&mut line_buffer) {
let args : Vec<String> = line_buffer
let args: Vec<String> = line_buffer
.replace("\n", "")
.split_whitespace()
.map(|arg| arg.to_string())
.collect();
match (args[0].as_str(), args.len()) {
("exit",1) => {
("exit", 1) => {
client.rpc().close().await?;
break;
}
("get",2) => {
("whoami", 1) => {
let req_id = client.whoami_req_send().await?;
let whoami = get_async(&mut client, req_id, whoami_res_parse).await?.id;
println!("{}", whoami);
}
("get", 2) => {
let msg_id = if args[1] == "any" {
"%TL34NIX8JpMJN+ubHWx6cRhIwEal8VqHdKVg2t6lFcg=.sha256".to_string()
} else {
args[1].clone()
};
let req_id = client.send_get(&msg_id).await?;
let msg = get_async(&mut client,-req_id,parse_message).await?;
println!("{:?}",msg);
let req_id = client.get_req_send(&msg_id).await?;
let msg = get_async(&mut client, req_id, message_res_parse).await?;
println!("{:?}", msg);
}
("user",2) => {
let user_id = if args[1] == "me" {
&whoami
} else {
&args[1]
};
("user", 2) => {
let user_id = if args[1] == "me" { &whoami } else { &args[1] };
let args = CreateHistoryStreamArgs::new(&user_id);
let req_id = client.send_create_history_stream(&args).await?;
print_source_until_eof(&mut client, -req_id, parse_feed).await?;
let args = CreateHistoryStreamArgs::new(user_id.clone());
let req_id = client.create_history_stream_req_send(&args).await?;
print_source_until_eof(&mut client, req_id, feed_res_parse).await?;
}
("feed",1) => {
("feed", 1) => {
let args = CreateStreamArgs::default();
let req_id = client.send_create_feed_stream(&args).await?;
print_source_until_eof(&mut client, -req_id, parse_feed).await?;
print_source_until_eof(&mut client, req_id, feed_res_parse).await?;
}
("latest",1) => {
("latest", 1) => {
let req_id = client.send_latest().await?;
print_source_until_eof(&mut client, -req_id, parse_latest).await?;
print_source_until_eof(&mut client, req_id, latest_res_parse).await?;
}
("private",2) => {
let user_id = if args[1] == "me" {
&whoami
} else {
&args[1]
};
("private", 2) => {
let user_id = if args[1] == "me" { &whoami } else { &args[1] };
let show_private = |header: &Header, body: &[u8]| {
let msg = parse_feed(header,body)?.into_message()?;
let show_private = |body: &[u8]| {
let msg = feed_res_parse(body)?.into_message()?;
if let serde_json::Value::String(content) = msg.content() {
if is_privatebox(&content) {
let ret = privatebox_decipher(&content, &sk)?
.unwrap_or("".to_string());
let ret = privatebox_decipher(&content, &sk)?.unwrap_or("".to_string());
return Ok(ret);
}
}
return Ok("".to_string());
};
};
let args = CreateHistoryStreamArgs::new(&user_id);
let req_id = client.send_create_history_stream(&args).await?;
let args = CreateHistoryStreamArgs::new(user_id.clone());
let req_id = client.create_history_stream_req_send(&args).await?;
print_source_until_eof(&mut client, -req_id, show_private).await?;
print_source_until_eof(&mut client, req_id, show_private).await?;
}
_ => println!("unknown command {}",line_buffer),
_ => println!("unknown command {}", line_buffer),
}
line_buffer.clear();
}
Ok(())
}
}

4
src/patchwork/error.rs → src/api/error.rs
View File

@ -54,6 +54,6 @@ impl std::fmt::Display for Error {
write!(f, "{:?}", self)
}
}
impl std::error::Error for Error { }
impl std::error::Error for Error {}
pub type Result<T> = std::result::Result<T, Error>;
pub type Result<T> = std::result::Result<T, Error>;

165
src/patchwork/api.rs → src/api/helper.rs
View File

@ -1,30 +1,27 @@
use crate::feed::Message;
use crate::rpc::{BodyType, RequestNo, RpcStream, RpcType};
use async_std::io::{Read, Write};
use serde_json;
use std::str::FromStr;
use crate::rpc::{RpcClient, Header, RequestNo, RpcType};
use crate::feed::Message;
use crate::feed::Feed;
use super::error::Result;
use super::error::{Error,Result};
#[derive(Debug, Deserialize)]
#[derive(Debug, Serialize, Deserialize)]
pub struct ErrorRes {
pub name: String,
pub message: String,
pub stack: String,
}
#[derive(Debug, Deserialize)]
#[derive(Debug, Serialize, Deserialize)]
pub struct WhoAmI {
pub id: String,
}
#[derive(Debug, Deserialize)]
#[derive(Debug, Serialize, Deserialize)]
pub struct LatestUserMessage {
pub id: String,
pub sequence: u64,
pub ts: u64,
pub ts: f64,
}
// https://github.com/ssbc/ssb-db/blob/master/api.md
@ -154,10 +151,10 @@ impl<K> CreateStreamArgs<K> {
}
}
#[derive(Debug, Serialize)]
pub struct CreateHistoryStreamArgs<'a> {
#[derive(Debug, Serialize, Deserialize)]
pub struct CreateHistoryStreamArgs {
// id (FeedID, required): The id of the feed to fetch.
pub id: &'a str,
pub id: String,
/// (number, default: 0): If seq > 0, then only stream messages with sequence numbers greater than seq.
#[serde(skip_serializing_if = "Option::is_none")]
@ -179,8 +176,8 @@ pub struct CreateHistoryStreamArgs<'a> {
pub limit: Option<u64>,
}
impl<'a> CreateHistoryStreamArgs<'a> {
pub fn new(id: &'a str) -> Self {
impl CreateHistoryStreamArgs {
pub fn new(id: String) -> Self {
Self {
id,
seq: None,
@ -217,95 +214,127 @@ impl<'a> CreateHistoryStreamArgs<'a> {
}
}
#[derive(Debug)]
pub enum ApiMethod {
WhoAmI,
Get,
CreateHistoryStream,
CreateFeedStream,
Latest,
}
fn parse_json<'a, T: serde::Deserialize<'a>>(
header: &'a Header,
body: &'a [u8],
) -> Result<T> {
if header.is_end_or_error {
let error: ErrorRes = serde_json::from_slice(&body[..])?;
Err(Error::ServerMessage(format!("{:?}", error)))
} else {
let res: T = serde_json::from_slice(&body[..])?;
Ok(res)
impl ApiMethod {
pub fn selector(&self) -> &'static [&'static str] {
use ApiMethod::*;
match self {
WhoAmI => &["whoami"],
Get => &["get"],
CreateHistoryStream => &["createHistoryStream"],
CreateFeedStream => &["createFeedStream"],
Latest => &["latest"],
}
}
pub fn from_selector(s: &[&str]) -> Option<Self> {
use ApiMethod::*;
match s {
["whoami"] => Some(WhoAmI),
["get"] => Some(Get),
["createHistoryStream"] => Some(CreateHistoryStream),
["createFeedStream"] => Some(CreateFeedStream),
["latest"] => Some(Latest),
_ => None,
}
}
}
pub fn parse_whoami(header: &Header, body: &[u8]) -> Result<WhoAmI> {
parse_json::<WhoAmI>(&header, body)
pub struct ApiHelper<R: Read + Unpin, W: Write + Unpin> {
rpc: RpcStream<R, W>,
}
pub fn parse_message(header: &Header, body: &[u8]) -> Result<Message> {
if header.is_end_or_error {
let error: ErrorRes = serde_json::from_slice(&body[..])?;
Err(Error::ServerMessage(format!("{:?}", error)))
} else {
Ok(Message::from_slice(body)?)
}
}
pub fn parse_feed(header: &Header, body: &[u8]) -> Result<Feed> {
if header.is_end_or_error {
let error: ErrorRes = serde_json::from_slice(&body[..])?;
Err(Error::ServerMessage(format!("{:?}", error)))
} else {
Ok(Feed::from_str(&String::from_utf8_lossy(&body))?)
}
}
pub fn parse_latest(header: &Header, body: &[u8]) -> Result<LatestUserMessage> {
parse_json::<LatestUserMessage>(&header, body)
}
pub struct ApiClient<R: Read + Unpin, W: Write + Unpin> {
rpc: RpcClient<R, W>,
}
impl<R: Read + Unpin, W: Write + Unpin> ApiClient<R, W> {
pub fn new(rpc: RpcClient<R, W>) -> Self {
impl<R: Read + Unpin, W: Write + Unpin> ApiHelper<R, W> {
pub fn new(rpc: RpcStream<R, W>) -> Self {
Self { rpc }
}
pub fn rpc(&mut self) -> &mut RpcClient<R, W> {
pub fn rpc(&mut self) -> &mut RpcStream<R, W> {
&mut self.rpc
}
// whoami: sync
// Get information about the current ssb-server user.
pub async fn send_whoami(&mut self) -> Result<RequestNo> {
pub async fn whoami_req_send(&mut self) -> Result<RequestNo> {
let args: [&str; 0] = [];
let req_no = self.rpc.send(&["whoami"], RpcType::Async, &args).await?;
let req_no = self
.rpc
.send_request(ApiMethod::WhoAmI.selector(), RpcType::Async, &args)
.await?;
Ok(req_no)
}
pub async fn whoami_res_send(&mut self, req_no: RequestNo, id: String) -> Result<()> {
let body = serde_json::to_string(&WhoAmI { id })?;
Ok(self
.rpc
.send_response(req_no, RpcType::Async, BodyType::JSON, body.as_bytes())
.await?)
}
// get: async
// Get a message by its hash-id. (sould start with %)
pub async fn send_get(&mut self, msg_id: &str) -> Result<RequestNo> {
let req_no = self.rpc.send(&["get"], RpcType::Async, &msg_id).await?;
pub async fn get_req_send(&mut self, msg_id: &str) -> Result<RequestNo> {
let req_no = self
.rpc
.send_request(ApiMethod::Get.selector(), RpcType::Async, &msg_id)
.await?;
Ok(req_no)
}
pub async fn get_res_send(&mut self, req_no: RequestNo, msg: &Message) -> Result<()> {
self.rpc
.send_response(
req_no,
RpcType::Async,
BodyType::JSON,
msg.to_string().as_bytes(),
)
.await?;
Ok(())
}
// createHistoryStream: source
// (hist) Fetch messages from a specific user, ordered by sequence numbers.
pub async fn send_create_history_stream<'a>(
pub async fn create_history_stream_req_send(
&mut self,
args: &'a CreateHistoryStreamArgs<'a>,
args: &CreateHistoryStreamArgs,
) -> Result<RequestNo> {
let req_no = self
.rpc
.send(&["createHistoryStream"], RpcType::Source, &args)
.send_request(
ApiMethod::CreateHistoryStream.selector(),
RpcType::Source,
&args,
)
.await?;
Ok(req_no)
}
pub async fn feed_res_send(&mut self, req_no: RequestNo, feed: &str) -> Result<()> {
self.rpc
.send_response(req_no, RpcType::Async, BodyType::JSON, feed.as_bytes())
.await?;
Ok(())
}
// createFeedStream: source
// (feed) Fetch messages ordered by their claimed timestamps.
pub async fn send_create_feed_stream<'a>(
&mut self,
args: &'a CreateStreamArgs<u64>,
args: &CreateStreamArgs<u64>,
) -> Result<RequestNo> {
let req_no = self
.rpc
.send(&["createFeedStream"], RpcType::Source, &args)
.send_request(
ApiMethod::CreateFeedStream.selector(),
RpcType::Source,
&args,
)
.await?;
Ok(req_no)
}
@ -314,8 +343,10 @@ impl<R: Read + Unpin, W: Write + Unpin> ApiClient<R, W> {
// Get the seq numbers of the latest messages of all users in the database.
pub async fn send_latest(&mut self) -> Result<RequestNo> {
let args: [&str; 0] = [];
let req_no = self.rpc.send(&["latest"], RpcType::Source, &args).await?;
let req_no = self
.rpc
.send_request(ApiMethod::Latest.selector(), RpcType::Async, &args)
.await?;
Ok(req_no)
}
}

8
src/api/mod.rs Normal file
View File

@ -0,0 +1,8 @@
mod error;
mod helper;
pub mod msgs;
pub use error::{Error, Result};
pub use helper::{
ApiHelper, ApiMethod, CreateHistoryStreamArgs, CreateStreamArgs, LatestUserMessage, WhoAmI,
};

32
src/patchwork/messagetypes.rs → src/api/msgs.rs
View File

@ -5,21 +5,33 @@ use std::collections::HashMap;
pub type SsbHash = String;
pub type SsbId = String;
#[derive(Debug, Deserialize)]
#[derive(Debug, Serialize, Deserialize)]
pub struct Mention {
pub link: SsbId,
pub name: Option<String>,
}
#[derive(Debug, Deserialize)]
pub struct MessageContent {
#[derive(Debug, Serialize, Deserialize)]
pub struct Post {
#[serde(rename = "type")]
pub xtype: String,
pub text: String,
pub mentions: Option<Vec<Mention>>,
}
impl Post {
pub fn new(text: String, mentions: Option<Vec<Mention>>) -> Self {
Post {
xtype: String::from("post"),
text,
mentions,
}
}
pub fn to_msg(&self) -> serde_json::Result<serde_json::Value> {
serde_json::to_value(self)
}
}
#[derive(Debug, Deserialize)]
pub struct PubAddress {
pub host: Option<String>,
@ -84,16 +96,7 @@ pub enum FeedTypedContent {
#[serde(rename = "pub")]
Pub { address: Option<PubAddress> },
#[serde(rename = "post")]
Post {
text: Option<String>,
post: Option<String>, // the same than text
channel: Option<String>,
mentions: Option<Mentions>,
root: Option<SsbHash>,
branch: Option<Branch>,
reply: Option<HashMap<SsbHash, SsbId>>,
recps: Option<String>,
},
Post,
#[serde(rename = "contact")]
Contact {
contact: Option<SsbId>,
@ -118,4 +121,3 @@ pub enum FeedTypedContent {
#[serde(rename = "vote")]
Vote { vote: Vote },
}

3
src/crypto/error.rs
View File

@ -18,7 +18,6 @@ impl std::fmt::Display for Error {
write!(f, "{:?}", self)
}
}
impl std::error::Error for Error { }
impl std::error::Error for Error {}
pub type Result<T> = std::result::Result<T, Error>;

8
src/crypto/mod.rs
View File

@ -1,5 +1,7 @@
mod sodium;
mod error;
mod sodium;
pub use sodium::ToSodiumObject;
pub use error::{Error,Result};
pub use error::{Error, Result};
pub use sodium::{
ToSodiumObject, ToSsbId, CURVE_ED25519_SUFFIX, ED25519_SIGNATURE_SUFFIX, SHA256_SUFFIX,
};

79
src/crypto/sodium.rs
View File

@ -1,69 +1,74 @@
use sodiumoxide::crypto::sign::ed25519;
use sodiumoxide::crypto::hash::sha256;
use base64;
use sodiumoxide::crypto::hash::sha256;
use sodiumoxide::crypto::sign::ed25519;
use super::error::{Error,Result};
use super::error::{Error, Result};
const CURVE_ED25519_SUFFIX : &str = ".ed25519";
const ED25519_SIGNATURE_SUFFIX : &str = ".sig.ed25519";
const SHA256_SUFFIX : &str = ".sha256";
pub const CURVE_ED25519_SUFFIX: &str = ".ed25519";
pub const ED25519_SIGNATURE_SUFFIX: &str = ".sig.ed25519";
pub const SHA256_SUFFIX: &str = ".sha256";
pub trait ToSodiumObject {
fn to_ed25519_pk(&self) -> Result<ed25519::PublicKey>;
fn to_ed25519_sk(&self) -> Result<ed25519::SecretKey>;
fn to_ed25519_sk_no_suffix(&self) -> Result<ed25519::SecretKey>;
fn to_ed25519_signature(&self) -> Result<ed25519::Signature>;
fn to_ed25519_signature(&self) -> Result<ed25519::Signature>;
fn to_sha256(&self) -> Result<sha256::Digest>;
}
pub trait ToSsbId {
fn to_ssb_id(&self) -> String;
}
impl<'a> ToSsbId for ed25519::PublicKey {
fn to_ssb_id(&self) -> String {
format!("{}{}", base64::encode(self), CURVE_ED25519_SUFFIX)
}
}
impl ToSodiumObject for str {
fn to_ed25519_pk(self : &str) -> Result<ed25519::PublicKey> {
fn to_ed25519_pk(self: &str) -> Result<ed25519::PublicKey> {
if !self.ends_with(CURVE_ED25519_SUFFIX) {
return Err(Error::InvalidSuffix);
}
let key_len = self.len()-CURVE_ED25519_SUFFIX.len();
let key_len = self.len() - CURVE_ED25519_SUFFIX.len();
let bytes = base64::decode(&self[..key_len])?;
ed25519::PublicKey::from_slice(&bytes)
.ok_or_else(|| Error::BadPublicKey)
}
fn to_ed25519_sk(self : &str) -> Result<ed25519::SecretKey> {
if !self.ends_with(CURVE_ED25519_SUFFIX) {
return Err(Error::InvalidSuffix);
}
let key_len = self.len()-CURVE_ED25519_SUFFIX.len();
let bytes = base64::decode(&self[..key_len])?;
ed25519::SecretKey::from_slice(&bytes)
.ok_or_else(|| Error::BadSecretKey)
ed25519::PublicKey::from_slice(&bytes).ok_or_else(|| Error::BadPublicKey)
}
fn to_ed25519_sk_no_suffix(self : &str) -> Result<ed25519::SecretKey> {
fn to_ed25519_sk(self: &str) -> Result<ed25519::SecretKey> {
if !self.ends_with(CURVE_ED25519_SUFFIX) {
return Err(Error::InvalidSuffix);
}
let key_len = self.len() - CURVE_ED25519_SUFFIX.len();
let bytes = base64::decode(&self[..key_len])?;
ed25519::SecretKey::from_slice(&bytes).ok_or_else(|| Error::BadSecretKey)
}
fn to_ed25519_sk_no_suffix(self: &str) -> Result<ed25519::SecretKey> {
let bytes = base64::decode(&self[..])?;
ed25519::SecretKey::from_slice(&bytes)
.ok_or_else(|| Error::BadSecretKey)
ed25519::SecretKey::from_slice(&bytes).ok_or_else(|| Error::BadSecretKey)
}
fn to_sha256(self : &str) -> Result<sha256::Digest> {
fn to_sha256(self: &str) -> Result<sha256::Digest> {
if !self.ends_with(SHA256_SUFFIX) {
return Err(Error::InvalidSuffix);
}
let key = base64::decode(&self[..self.len()-SHA256_SUFFIX.len()])?;
let key = base64::decode(&self[..self.len() - SHA256_SUFFIX.len()])?;
sha256::Digest::from_slice(&key)
.ok_or(Error::InvalidDigest)
sha256::Digest::from_slice(&key).ok_or(Error::InvalidDigest)
}
fn to_ed25519_signature(self : &str) -> Result<ed25519::Signature> {
fn to_ed25519_signature(self: &str) -> Result<ed25519::Signature> {
if !self.ends_with(ED25519_SIGNATURE_SUFFIX) {
return Err(Error::InvalidSuffix);
}
let signature = base64::decode(&self[..self.len()-ED25519_SIGNATURE_SUFFIX.len()])?;
let signature = base64::decode(&self[..self.len() - ED25519_SIGNATURE_SUFFIX.len()])?;
ed25519::Signature::from_slice(&signature)
.ok_or(Error::CannotCreateSignature)
ed25519::Signature::from_slice(&signature).ok_or(Error::CannotCreateSignature)
}
}

324
src/db/feeds.rs
View File

@ -1,324 +0,0 @@
use std::io;
use std::path::PathBuf;
use std::fs::{File,OpenOptions};
use std::io::SeekFrom;
use std::io::prelude::*;
use super::error::{Error,Result};
pub struct FeedsStorage {
path : PathBuf
}
impl FeedsStorage {
fn filename(&self, user_id : &str) -> PathBuf {
let name = user_id.chars().map(|ch| match ch {
'+' => '-',
'/' => '_',
_ => ch,
}).collect::<String>();
let mut path = PathBuf::new();
path.push(&self.path);
path.push(name);
path
}
pub fn new(path : PathBuf) -> Self{
FeedsStorage { path }
}
pub fn user(&self, user_id : String) -> FeedStorage {
FeedStorage {
path : self.filename(&user_id)
}
}
}
pub struct FeedStorage {
path : PathBuf,
}
impl FeedStorage {
/*
raw feed storage structure:
- last sequence in feed - 32 bits be
- * | message-len 32 bits be
| message
| message-len 32 bits be
*/
pub fn append(&self, seq_no: u32, feed: &str) -> Result<()> {
let mut file = OpenOptions::new()
.create(true)
.read(true)
.write(true)
.open(&self.path)?;
// check and update feed sequence number
let created = file.seek(SeekFrom::End(0))? == 0;
if created {
if seq_no != 1{
return Err(Error::InvalidSequenceNo);
}
} else {
let file_seq_no = self.get_last_seq(&mut file)?;
if file_seq_no + 1 != seq_no {
return Err(Error::InvalidSequenceNo);
}
}
self.set_last_seq(&mut file,seq_no)?;
file.seek(SeekFrom::End(0))?;
// write feed size dummy
file.write_all( &(0 as u32).to_be_bytes()[..] )?;
// write compressed feed
let offset = file.seek(SeekFrom::Current(0))?;
let mut wtr = snap::Writer::new(file);
io::copy(&mut feed.as_bytes(), &mut wtr)?;
let mut file = wtr.into_inner().map_err(
|err| Error::CompressionError(format!("{:?}",err))
)?;
let len = file.seek(SeekFrom::Current(0))?- offset;
// write feed size
file.write_all( &(len as u32).to_be_bytes()[..] )?;
file.seek(SeekFrom::End(-((8+len) as i64)))?;
file.write_all( &(len as u32).to_be_bytes()[..] )?;
Ok(())
}
pub fn last_seq(&self) -> Result<u32> {
if !self.path.exists() {
return Ok(0);
}
let mut file = OpenOptions::new()
.read(true)
.open(&self.path)?;
self.get_last_seq(&mut file)
}
fn get_last_seq(&self, file: &mut File) -> Result<u32> {
let mut file_seq_no = [0u8;4];
file.seek(SeekFrom::Start(0))?;
file.read_exact(&mut file_seq_no[..])?;
Ok(u32::from_be_bytes(file_seq_no))
}
fn set_last_seq(&self, file: &mut File, seq_no: u32) -> Result<()> {
file.seek(SeekFrom::Start(0))?;
file.write_all( &seq_no.to_be_bytes()[..] )?;
Ok(())
}
pub fn iter(&self) -> Result<FeedStorageIterator> {
let mut file = OpenOptions::new()
.read(true)
.open(&self.path)?;
let last_seq_no = self.get_last_seq(&mut file)?;
Ok(FeedStorageIterator{
file,
current_seq_no : 0,
last_seq_no,
})
}
pub fn rev_iter(&self) -> Result<FeedStorageReverseIterator> {
let mut file = OpenOptions::new()
.read(true)
.open(&self.path)?;
let last_seq_no = self.get_last_seq(&mut file)?;
file.seek(SeekFrom::End(-4))?;
Ok(FeedStorageReverseIterator{
file,
current_seq_no : last_seq_no,
})
}
}
#[derive(PartialEq,Debug)]
pub struct Feed {
pub seq_no : u32,
pub value : String,
}
pub struct FeedStorageIterator {
file : File,
current_seq_no : u32,
last_seq_no : u32,
}
impl Iterator for FeedStorageIterator {
type Item = Result<Feed>;
// next() is the only required method
fn next(&mut self) -> Option<Self::Item> {
if self.current_seq_no >= self.last_seq_no {
return None;
}
// read compressed size
let mut size_buf = [0u8;4];
if let Err(err) = self.file.read_exact(&mut size_buf[..]) {
return Some(Err(Error::Io(err)));
}
let size = u32::from_be_bytes(size_buf);
// read compressed data
let mut compressed = vec![0; size as usize];
if let Err(err) = self.file.read_exact(&mut compressed[..]) {
return Some(Err(Error::Io(err)));
}
let mut rdr = snap::Reader::new(&compressed[..]);
let mut plaintext : Vec<u8> = Vec::new();
if let Err(err) = io::copy(&mut rdr, &mut plaintext) {
return Some(Err(Error::Io(err)));
}
// read compresed size again
if let Err(err) = self.file.read_exact(&mut size_buf[..]) {
return Some(Err(Error::Io(err)));
}
if u32::from_be_bytes(size_buf) != size {
return Some(Err(Error::MismatchReadingSecondSize));
}
self.current_seq_no += 1;
let ret = match String::from_utf8(plaintext) {
Err(err) => Err(Error::Utf8(err)),
Ok(value) => Ok(Feed { seq_no : self.current_seq_no, value })
};
Some(ret)
}
}
pub struct FeedStorageReverseIterator {
file : File,
current_seq_no : u32,
}
impl Iterator for FeedStorageReverseIterator {
type Item = Result<Feed>;
// next() is the only required method
fn next(&mut self) -> Option<Self::Item> {
if self.current_seq_no == 0 {
return None;
}
// read compressed size
let mut size_buf = [0u8;4];
if let Err(err) = self.file.read_exact(&mut size_buf[..]) {
return Some(Err(Error::Io(err)));
}
let size = u32::from_be_bytes(size_buf);
if let Err(err) = self.file.seek(SeekFrom::Current(-((size+8) as i64))) {
return Some(Err(Error::Io(err)));
}
// read compresed size again
if let Err(err) = self.file.read_exact(&mut size_buf[..]) {
return Some(Err(Error::Io(err)));
}
if u32::from_be_bytes(size_buf) != size {
return Some(Err(Error::MismatchReadingSecondSize));
}
// read compressed data
let mut compressed = vec![0u8; size as usize];
if let Err(err) = self.file.read_exact(&mut compressed[..]) {
return Some(Err(Error::Io(err)));
}
let mut rdr = snap::Reader::new(&compressed[..]);
let mut plaintext : Vec<u8> = Vec::new();
if let Err(err) = io::copy(&mut rdr, &mut plaintext) {
return Some(Err(Error::Io(err)));
}
// prepare offset for the next read
if let Err(err) = self.file.seek(SeekFrom::Current(-((size+8) as i64))) {
return Some(Err(Error::Io(err)));
}
let ret = match String::from_utf8(plaintext) {
Err(err) => Err(Error::Utf8(err)),
Ok(value) => Ok(Feed { seq_no : self.current_seq_no, value })
};
self.current_seq_no -= 1;
Some(ret)
}
}
#[cfg(test)]
mod test {
use super::*;
use rand::distributions::Alphanumeric;
use rand::{thread_rng, Rng};
use std::iter;
fn rand_folder() -> Result<PathBuf> {
let mut rng = thread_rng();
let name: String = iter::repeat(())
.map(|()| rng.sample(Alphanumeric))
.take(12)
.collect();
let mut tmp_folder = std::env::temp_dir();
tmp_folder.push(name);
std::fs::create_dir(&tmp_folder)?;
Ok(tmp_folder)
}
#[test]
fn test_db_feeds() -> Result<()> {
let user_id = "@ZFWw+UclcUgYi081/C8lhgH+KQ9s7YJRoOYGnzxW/JQ=.ed25519";
let feeds = FeedsStorage::new(rand_folder()?);
let feed = feeds.user(user_id.to_owned());
let f1 = Feed{seq_no:1, value:"123".to_string()};
let f2 = Feed{seq_no:2, value:"8181".to_string()};
let f3 = Feed{seq_no:3, value:"182881".to_string()};
assert_eq!(0,feed.last_seq()?);
feed.append(f1.seq_no, &f1.value)?;
assert_eq!(1,feed.last_seq()?);
feed.append(f2.seq_no, &f2.value)?;
assert_eq!(2,feed.last_seq()?);
feed.append(f3.seq_no, &f3.value)?;
assert_eq!(3,feed.last_seq()?);
let mut it = feed.iter()?;
assert_eq!(it.next().unwrap()?,f1);
assert_eq!(it.next().unwrap()?,f2);
assert_eq!(it.next().unwrap()?,f3);
assert_eq!(it.next().is_none(),true);
let mut rev_it = feed.rev_iter()?;
assert_eq!(rev_it.next().unwrap()?,f3);
assert_eq!(rev_it.next().unwrap()?,f2);
assert_eq!(rev_it.next().unwrap()?,f1);
assert_eq!(rev_it.next().is_none(),true);
Ok(())
}
}

4
src/db/mod.rs
View File

@ -1,4 +0,0 @@
mod error;
pub mod feeds;
pub use error::{Error,Result};

27
src/discovery/error.rs Normal file
View File

@ -0,0 +1,27 @@
#[derive(Debug)]
pub enum Error {
ParseInt(std::num::ParseIntError),
InvalidInviteCode,
CryptoFormat(crate::crypto::Error),
}
impl From<crate::crypto::Error> for Error {
fn from(err: crate::crypto::Error) -> Self {
Error::CryptoFormat(err)
}
}
impl From<std::num::ParseIntError> for Error {
fn from(err: std::num::ParseIntError) -> Self {
Error::ParseInt(err)
}
}
impl std::fmt::Display for Error {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
write!(f, "{:?}", self)
}
}
impl std::error::Error for Error {}
pub type Result<T> = std::result::Result<T, Error>;

6
src/discovery/mod.rs Normal file
View File

@ -0,0 +1,6 @@
mod error;
mod network;
mod pubs;
pub use network::ssb_net_id;
pub use pubs::Invite;

6
src/discovery/network.rs Normal file
View File

@ -0,0 +1,6 @@
use sodiumoxide::crypto::auth;
pub const SSB_NET_ID: &str = "d4a1cb88a66f02f8db635ce26441cc5dac1b08420ceaac230839b755845a9ffb";
pub fn ssb_net_id() -> auth::Key {
auth::Key::from_slice(&hex::decode(SSB_NET_ID).unwrap()).unwrap()
}

22
src/patchwork/pubs.rs → src/discovery/pubs.rs
View File

@ -1,25 +1,25 @@
use sodiumoxide::crypto::sign::ed25519;
use crate::crypto::ToSodiumObject;
use sodiumoxide::crypto::sign::ed25519;
use super::error::{Error,Result};
use super::error::{Error, Result};
pub struct Invite {
pub domain : String,
pub port : u16,
pub domain: String,
pub port: u16,
pub pub_pk: ed25519::PublicKey,
pub invite_sk: ed25519::SecretKey,
}
impl Invite {
pub fn from_code(code : &str) -> Result<Self> {
let domain_port_keys : Vec<_> = code.split(':').collect();
pub fn from_code(code: &str) -> Result<Self> {
let domain_port_keys: Vec<_> = code.split(':').collect();
if domain_port_keys.len() != 3 {
return Err(Error::InvalidInviteCode);
}
let domain = domain_port_keys[0].to_string();
let port = domain_port_keys[1].parse::<u16>()?;
let pk_sk :Vec<_> = domain_port_keys[2].split('~').collect();
let pk_sk: Vec<_> = domain_port_keys[2].split('~').collect();
if pk_sk.len() != 2 {
return Err(Error::InvalidInviteCode);
@ -27,9 +27,13 @@ impl Invite {
let pub_pk = pk_sk[0][1..].to_ed25519_pk()?;
let invite_sk = pk_sk[1][..].to_ed25519_sk_no_suffix()?;
Ok(Invite { domain, port, pub_pk, invite_sk })
Ok(Invite {
domain,
port,
pub_pk,
invite_sk,
})
}
}
#[cfg(test)]

46
src/feed/base.rs
View File

@ -1,12 +1,12 @@
use std::str::FromStr;
use serde_json::Value;
use super::error::{Error, Result};
use super::message::Message;
use super::ssb_sha256;
use super::error::{Error,Result};
use std::time::SystemTime;
use std::time::UNIX_EPOCH;
#[derive(Debug,Serialize,Deserialize)]
#[derive(Debug, Serialize, Deserialize)]
pub struct Feed {
pub key: String,
pub value: Value,
@ -14,18 +14,32 @@ pub struct Feed {
pub rts: Option<f64>,
}
impl Feed {
pub fn into_message(self) -> Result<Message> {
Message::from_value(self.value)
impl ToString for Feed {
fn to_string(&self) -> String {
serde_json::to_string(self).unwrap()
}
}
impl FromStr for Feed {
type Err = Error;
fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
let feed : Feed = serde_json::from_str(&s)?;
let digest = format!("%{}.sha256",base64::encode(&ssb_sha256(&feed.value)?));
impl Feed {
pub fn into_message(self) -> Result<Message> {
Message::from_value(self.value)
}
pub fn new(m: Message) -> Self {
let key = format!("%{}.sha256", base64::encode(&ssb_sha256(&m.value).unwrap()));
let timestamp = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs_f64();
Feed {
key,
value: m.value,
timestamp,
rts: None,
}
}
pub fn from_slice(s: &[u8]) -> Result<Self> {
let feed: Feed = serde_json::from_slice(&s)?;
let digest = format!("%{}.sha256", base64::encode(&ssb_sha256(&feed.value)?));
if digest != feed.key {
return Err(Error::FeedDigestMismatch);
@ -38,11 +52,11 @@ impl FromStr for Feed {
#[cfg(test)]
mod test {
use super::*;
#[test]
fn test_verify_feed_integrity() -> Result<()> {
let feed = r#"{"key":"%Cg0ZpZ8cV85G8UIIropgBOvM8+Srlv9LSGDNGnpdK44=.sha256","value":{"previous":"%seUEAo7PTyA7vNwnOrmGIsUFfpyRzOvzGVv1QCb/Fz8=.sha256","author":"@BIbVppzlrNiRJogxDYz3glUS7G4s4D4NiXiPEAEzxdE=.ed25519","sequence":37,"timestamp":1439392020612,"hash":"sha256","content":{"type":"post","text":"@paul real time replies didn't work.","repliesTo":"%xWKunF6nXD7XMC+D4cjwDMZWmBnmRu69w9T25iLNa1Q=.sha256","mentions":["%7UKRfZb2u8al4tYWHqM55R9xpE/KKVh9U0M6BdugGt4=.sha256"],"recps":[{"link":"@hxGxqPrplLjRG2vtjQL87abX4QKqeLgCwQpS730nNwE=.ed25519","name":"paul"}]},"signature":"gGxSPdBJZxp6x5f3HzQGoQSeSdh/C5AtymIn+miWa+lcC6DdqpRSgaeH9KHeLf+/CKhU6REYIpWaLr4CKDMfCg==.sig.ed25519"},"timestamp":1573574678194,"rts":1439392020612}"#;
Feed::from_str(&feed)?;
Feed::from_slice(feed.as_bytes())?;
Ok(())
}
}
}

17
src/feed/encoding.rs
View File

@ -1,6 +1,6 @@
use super::error::Result;
use serde_json::Value;
use sodiumoxide::crypto::hash::sha256;
use super::error::Result;
pub fn ssb_sha256(v: &Value) -> Result<sha256::Digest> {
let v8encoding = stringify_json(&v)?
@ -60,8 +60,8 @@ pub fn stringify_json(v: &Value) -> Result<String> {
Value::Number(value) => {
let mut as_str = value.to_string();
if as_str.contains('e') && !as_str.contains("e-") {
as_str = as_str.replace("e","e+")
}
as_str = as_str.replace("e", "e+")
}
buffer.push_str(&as_str);
}
Value::Bool(value) => {
@ -82,7 +82,7 @@ pub fn stringify_json(v: &Value) -> Result<String> {
mod test {
use super::*;
const JSON : &str = r#"{"a":0,"b":1.1,"c":null,"d":true,"f":false,"g":{},"h":{"h1":1},"i":[],"j":[1],"k":[1,2]}"#;
const JSON: &str = r#"{"a":0,"b":1.1,"c":null,"d":true,"f":false,"g":{},"h":{"h1":1},"i":[],"j":[1],"k":[1,2]}"#;
#[test]
fn test_json_stringify() -> Result<()> {
let v: Value = serde_json::from_str(JSON)?;
@ -124,8 +124,8 @@ mod test {
fn test_msg_with_float_mantissa() -> Result<()> {
let expected = "RUcldndjJUkEcZ5hX6zAj/xLlnh0n4BZ6ThJOW5RvIk=";
let message = r#"{"previous":"%gbem82xZNVHbOM2pyOlxymsAfstdMFfGSoawWQtObX8=.sha256","author":"@TXKFQehlyoSn8UJAIVP/k2BjFINC591MlBC2e2d24mA=.ed25519","sequence":1557,"timestamp":1495245157893,"hash":"sha256","content":{"type":"post","transactionHash":9.691449834862513e+76,"address":7.073631810716965e+46,"event":"ActionAdded","text":"{\"actionID\":\"1\",\"amount\":\"0\",\"description\":\"Bind Ethereum events to Secure Scuttlebutt posts\"}}"},"signature":"/Qvm9ozEfl0Thyvs+mnwhLDReZ8xeKXA3hSXOxm53SFkLEnnJ+IF0l7LSqc56Y3vl8FwarJ6k0PGmcU3U8FMAw==.sig.ed25519"}"#;
let message_value: Value = serde_json::from_str(&message)?;
let current = base64::encode(&ssb_sha256(&message_value)?);
let message_value: Value = serde_json::from_str(&message)?;
let current = base64::encode(&ssb_sha256(&message_value)?);
assert_eq!(expected, current);
Ok(())
}
@ -134,10 +134,9 @@ mod test {
fn test_msg_with_float_precision() -> Result<()> {
let expected = "BUtTVIJyN5fUXzQy2uQfCCzlAg0s6laQQqFIu+kGnFM=";
let message = r#"{"previous":"%ButTjV+H9VfONhX+lLbJb5LR+W14SFqbmjOfdMPZ5+4=.sha256","sequence":15034,"author":"@6ilZq3kN0F+dXFHAPjAwMm87JEb/VdB+LC9eIMW3sa0=.ed25519","timestamp":1567190273951.0159,"hash":"sha256","content":{"type":"vote","channel":null,"vote":{"link":"%GvtUsekEwsCj1cQ6+4Gihkm+ek99BhB537g1xUKjhsA=.sha256","value":1,"expression":"Like"}},"signature":"UkVfqDmBhHrDfMvFT8iUhEispAku/zbdXKCyRVlxYp2wNtJ4okwKE7hTkKhbiMVA7sGIV5dzHZyMotXCL46iDw==.sig.ed25519"}"#;
let message_value: Value = serde_json::from_str(&message)?;
let current = base64::encode(&ssb_sha256(&message_value)?);
let message_value: Value = serde_json::from_str(&message)?;
let current = base64::encode(&ssb_sha256(&message_value)?);
assert_eq!(expected, current);
Ok(())
}
}

3
src/feed/error.rs
View File

@ -44,7 +44,6 @@ impl std::fmt::Display for Error {
write!(f, "{:?}", self)
}
}
impl std::error::Error for Error { }
impl std::error::Error for Error {}
pub type Result<T> = std::result::Result<T, Error>;

178
src/feed/message.rs
View File

@ -1,29 +1,30 @@
use std::time::SystemTime;
use std::str::FromStr;
use std::time::SystemTime;
use sodiumoxide::crypto::{sign::ed25519};
use serde_json::Value;
use sodiumoxide::crypto::sign::ed25519;
use super::error::{Error, Result};
use super::{ssb_sha256, stringify_json};
use crate::crypto::ToSodiumObject;
use crate::patchwork::IdentitySecret;
use super::{stringify_json,ssb_sha256};
use super::error::{Error,Result};
use crate::keystore::OwnedIdentity;
use sodiumoxide::crypto::hash::sha256;
const MSG_PREVIOUS : &str = "previous";
const MSG_AUTHOR : &str = "author";
const MSG_SEQUENCE : &str = "sequence";
const MSG_TIMESTAMP : &str = "timestamp";
const MSG_HASH : &str = "hash";
const MSG_CONTENT : &str = "content";
const MSG_SIGNATURE : &str = "signature";
const MSG_PREVIOUS: &str = "previous";
const MSG_AUTHOR: &str = "author";
const MSG_SEQUENCE: &str = "sequence";
const MSG_TIMESTAMP: &str = "timestamp";
const MSG_HASH: &str = "hash";
const MSG_CONTENT: &str = "content";
const MSG_SIGNATURE: &str = "signature";
macro_rules! cast {
($input:expr,$pth:path) => {
match $input {
Some($pth(x)) => Ok(x),
_ => Err(Error::InvalidJson),
_ => Err(Error::InvalidJson),
};
}
};
}
macro_rules! cast_opt {
@ -32,88 +33,93 @@ macro_rules! cast_opt {
None => Ok(None),
Some(Value::Null) => Ok(None),
Some($pth(x)) => Ok(Some(x)),
_ => Err(Error::InvalidJson)
_ => Err(Error::InvalidJson),
};
};
}
pub struct MessageId(sha256::Digest);
impl ToString for MessageId {
fn to_string(&self) -> String {
let digest = base64::encode(&self.0);
format!("%{}.sha256", digest)
}
}
impl AsRef<[u8]> for MessageId {
fn as_ref(&self) -> &[u8] {
self.0.as_ref()
}
}
#[derive(Debug)]
#[derive(Debug, Deserialize)]
pub struct Message {
value: serde_json::Value,
pub value: serde_json::Value,
}
impl Message {
pub fn new(prev : Option<&Message>, identity: &IdentitySecret, content: Value) -> Result<Self> {
let mut value : serde_json::Map<String,Value> = serde_json::Map::new();
pub fn sign(prev: Option<&Message>, identity: &OwnedIdentity, content: Value) -> Result<Self> {
let mut value: serde_json::Map<String, Value> = serde_json::Map::new();
if let Some(prev) = prev {
value.insert(MSG_PREVIOUS.to_string(), Value::String(prev.id()?));
value.insert(
MSG_PREVIOUS.to_string(),
Value::String(prev.id().to_string()),
);
value.insert(
MSG_SEQUENCE.to_string(),
Value::Number(serde_json::Number::from(prev.sequence()+1))
Value::Number(serde_json::Number::from(prev.sequence() + 1)),
);
} else {
value.insert(
MSG_SEQUENCE.to_string(),
Value::Number(serde_json::Number::from(1))
Value::Number(serde_json::Number::from(1)),
);
}
let timestamp = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)?
let timestamp = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)?
.as_millis() as u64;
let timestamp = Value::Number(serde_json::Number::from(timestamp));
value.insert(
MSG_AUTHOR.to_string(),
Value::String(identity.id.clone())
);
value.insert(
MSG_TIMESTAMP.to_string(),
timestamp
);
value.insert(
MSG_HASH.to_string(),
Value::String("sha256".to_string())
);
value.insert(
MSG_CONTENT.to_string(),
content
);
value.insert(MSG_AUTHOR.to_string(), Value::String(identity.id.clone()));
value.insert(MSG_TIMESTAMP.to_string(), timestamp);
value.insert(MSG_HASH.to_string(), Value::String("sha256".to_string()));
value.insert(MSG_CONTENT.to_string(), content);
let value = Value::Object(value);
let to_sign_text = stringify_json(&value)?;
let mut value = cast!(Some(value),Value::Object)?;
let mut value = cast!(Some(value), Value::Object)?;
let signature = ed25519::sign_detached(to_sign_text.as_bytes(), &identity.sk);
value.insert(
MSG_SIGNATURE.to_string(),
Value::String(format!("{}.sig.ed25519",base64::encode(&signature)))
Value::String(format!("{}.sig.ed25519", base64::encode(&signature))),
);
Ok(Message { value : Value::Object(value) })
Ok(Message {
value: Value::Object(value),
})
}
pub fn from_slice(s : &[u8]) -> Result<Self> {
pub fn from_slice(s: &[u8]) -> Result<Self> {
Self::from_value(serde_json::from_slice(&s)?)
}
pub fn from_value(v : Value) -> Result<Self> {
pub fn from_value(v: Value) -> Result<Self> {
let mut v = cast!(Some(v), Value::Object)?;
let mut v = cast!(Some(v),Value::Object)?;
// check if ok
cast_opt!(v.get(MSG_PREVIOUS),Value::String)?;
cast!(v.get(MSG_SEQUENCE),Value::Number)?;
cast!(v.get(MSG_SEQUENCE),Value::Number)?;
cast!(v.get(MSG_TIMESTAMP),Value::Number)?;
cast!(v.get(MSG_HASH),Value::String)?;
// check if ok
cast_opt!(v.get(MSG_PREVIOUS), Value::String)?;
cast!(v.get(MSG_SEQUENCE), Value::Number)?;
cast!(v.get(MSG_SEQUENCE), Value::Number)?;
cast!(v.get(MSG_TIMESTAMP), Value::Number)?;
cast!(v.get(MSG_HASH), Value::String)?;
v.get(MSG_CONTENT).ok_or(Error::InvalidJson)?;
// verify signature
let signature = cast!(v.remove(MSG_SIGNATURE),Value::String)?;
let author = cast!(v.get(MSG_AUTHOR),Value::String)?;
let signature = cast!(v.remove(MSG_SIGNATURE), Value::String)?;
let author = cast!(v.get(MSG_AUTHOR), Value::String)?;
let sig = signature.to_ed25519_signature()?;
let signer = author[1..].to_ed25519_pk()?;
@ -122,42 +128,43 @@ impl Message {
if !ed25519::verify_detached(&sig, &signed_text.as_ref(), &signer) {
return Err(Error::InvalidSignature);
}
// put signature back
let mut v = cast!(Some(value),Value::Object)?;
let mut v = cast!(Some(value), Value::Object)?;
v.insert(MSG_SIGNATURE.to_string(), Value::String(signature));
Ok(Message { value : Value::Object(v) })
Ok(Message {
value: Value::Object(v),
})
}
pub fn id(&self) -> Result<String> {
let digest = base64::encode(&ssb_sha256(&self.value)?);
Ok(format!("%{}.sha256",digest))
pub fn id(&self) -> MessageId {
MessageId(ssb_sha256(&self.value).unwrap())
}
pub fn previous(&self) -> Option<&String> {
cast_opt!(self.value.get(MSG_PREVIOUS),Value::String)
.unwrap()
cast_opt!(self.value.get(MSG_PREVIOUS), Value::String).unwrap()
}
pub fn author(&self) -> &String {
cast!(self.value.get(MSG_AUTHOR),Value::String)
cast!(self.value.get(MSG_AUTHOR), Value::String).unwrap()
}
pub fn sequence(&self) -> u64 {
cast!(self.value.get(MSG_SEQUENCE), Value::Number)
.unwrap()
.as_u64()
.unwrap()
}
pub fn sequence(&self) -> u64 {
cast!(self.value.get(MSG_SEQUENCE),Value::Number)
.unwrap().as_u64().unwrap()
}
pub fn timestamp(&self) -> f64 {
cast!(self.value.get(MSG_TIMESTAMP),Value::Number)
.unwrap().as_f64().unwrap()
cast!(self.value.get(MSG_TIMESTAMP), Value::Number)
.unwrap()
.as_f64()
.unwrap()
}
pub fn hash(&self) -> &String {
cast!(self.value.get(MSG_HASH),Value::String)
.unwrap()
cast!(self.value.get(MSG_HASH), Value::String).unwrap()
}
pub fn content(&self) -> &Value {
@ -165,10 +172,8 @@ impl Message {
}
pub fn signature(&self) -> &String {
cast!(self.value.get(MSG_SIGNATURE),Value::String)
.unwrap()
cast!(self.value.get(MSG_SIGNATURE), Value::String).unwrap()
}
}
impl FromStr for Message {
@ -188,25 +193,24 @@ impl ToString for Message {
#[cfg(test)]
mod test {
use super::*;
#[test]
fn test_verify_known_msg_integrity() -> Result<()> {
let message_id ="%Cg0ZpZ8cV85G8UIIropgBOvM8+Srlv9LSGDNGnpdK44=.sha256";
let message_id = "%Cg0ZpZ8cV85G8UIIropgBOvM8+Srlv9LSGDNGnpdK44=.sha256";
let message = r#"{"previous":"%seUEAo7PTyA7vNwnOrmGIsUFfpyRzOvzGVv1QCb/Fz8=.sha256","author":"@BIbVppzlrNiRJogxDYz3glUS7G4s4D4NiXiPEAEzxdE=.ed25519","sequence":37,"timestamp":1439392020612,"hash":"sha256","content":{"type":"post","text":"@paul real time replies didn't work.","repliesTo":"%xWKunF6nXD7XMC+D4cjwDMZWmBnmRu69w9T25iLNa1Q=.sha256","mentions":["%7UKRfZb2u8al4tYWHqM55R9xpE/KKVh9U0M6BdugGt4=.sha256"],"recps":[{"link":"@hxGxqPrplLjRG2vtjQL87abX4QKqeLgCwQpS730nNwE=.ed25519","name":"paul"}]},"signature":"gGxSPdBJZxp6x5f3HzQGoQSeSdh/C5AtymIn+miWa+lcC6DdqpRSgaeH9KHeLf+/CKhU6REYIpWaLr4CKDMfCg==.sig.ed25519"}"#;
let msg = Message::from_str(&message)?;
assert_eq!(msg.id()?,message_id);
assert_eq!(msg.id().to_string(), message_id);
Ok(())
}
#[test]
fn test_sign_verify() -> Result<()> {
let content = Value::String("thistest".to_string());
let id = IdentitySecret::new();
let msg1 = Message::new(None,&id,content.clone())?.to_string();
let id = OwnedIdentity::create();
let msg1 = Message::sign(None, &id, content.clone())?.to_string();
let msg1 = Message::from_str(&msg1)?;
let msg2 = Message::new(Some(&msg1),&id,content)?.to_string();
let msg2 = Message::sign(Some(&msg1), &id, content)?.to_string();
Message::from_str(&msg2)?;
Ok(())
}
}
}

12
src/feed/mod.rs
View File

@ -1,11 +1,11 @@
mod encoding;
mod base;
mod encoding;
mod error;
mod message;
mod privatebox;
mod error;
pub use message::Message;
pub use base::Feed;
pub use privatebox::{is_privatebox,privatebox_cipher,privatebox_decipher};
pub use encoding::{ssb_sha256,stringify_json};
pub use error::{Error,Result};
pub use encoding::{ssb_sha256, stringify_json};
pub use error::{Error, Result};
pub use message::Message;
pub use privatebox::{is_privatebox, privatebox_cipher, privatebox_decipher};

108
src/feed/privatebox.rs
View File

@ -1,58 +1,52 @@
use crate::crypto::ToSodiumObject;
use sodiumoxide::crypto::sign::SecretKey;
use sodiumoxide::crypto::{sign::ed25519, scalarmult::curve25519, secretbox};
use sodiumoxide::crypto::{scalarmult::curve25519, secretbox, sign::ed25519};
use super::error::{Error,Result};
use super::error::{Error, Result};
pub const SUFFIX : &str = ".box";
pub const SUFFIX: &str = ".box";
pub const MAX_RECIPIENTS : u8 = 7;
pub const MAX_RECIPIENTS: u8 = 7;
const RECIPIENT_COUNT_LEN : usize = 1;
const ENCRYPTED_HEADER_LEN : usize =
RECIPIENT_COUNT_LEN
+ secretbox::KEYBYTES
+ secretbox::MACBYTES;
const RECIPIENT_COUNT_LEN: usize = 1;
const ENCRYPTED_HEADER_LEN: usize = RECIPIENT_COUNT_LEN + secretbox::KEYBYTES + secretbox::MACBYTES;
pub fn is_privatebox(text : &str) -> bool {
text.ends_with(SUFFIX)
pub fn is_privatebox(text: &str) -> bool {
text.ends_with(SUFFIX)
}
pub fn privatebox_cipher(plaintext : &str, recipients: &[&str]) -> Result<String> {
let recipients : crate::crypto::Result<Vec<_>> = recipients
pub fn privatebox_cipher(plaintext: &str, recipients: &[&str]) -> Result<String> {
let recipients: crate::crypto::Result<Vec<_>> = recipients
.iter()
.map(|id| {
id[1..].to_ed25519_pk()
}).collect();
.map(|id| id[1..].to_ed25519_pk())
.collect();
let recipients = recipients?;
let recipients_ref : Vec<_> =
recipients.iter().map(|r| r).collect();
let recipients_ref: Vec<_> = recipients.iter().map(|r| r).collect();
let ciphertext = cipher(plaintext.as_bytes(),&recipients_ref[..])?;
let ciphertext = cipher(plaintext.as_bytes(), &recipients_ref[..])?;
Ok(format!("{}{}",base64::encode(&ciphertext),SUFFIX))
}
Ok(format!("{}{}", base64::encode(&ciphertext), SUFFIX))
}
pub fn privatebox_decipher(ciphertext : &str, sk : &SecretKey) -> Result<Option<String>> {
let msg = &ciphertext.as_bytes()[..ciphertext.len()-SUFFIX.len()];
pub fn privatebox_decipher(ciphertext: &str, sk: &SecretKey) -> Result<Option<String>> {
let msg = &ciphertext.as_bytes()[..ciphertext.len() - SUFFIX.len()];
let msg = base64::decode(msg)?;
let plaintext = decipher(&msg, &sk)?
.map(|msg| String::from_utf8_lossy(&msg).to_string());
let plaintext = decipher(&msg, &sk)?.map(|msg| String::from_utf8_lossy(&msg).to_string());
Ok(plaintext)
}
}
fn cipher(plaintext : &[u8], recipients : &[&ed25519::PublicKey]) -> Result<Box<[u8]>> {
fn cipher(plaintext: &[u8], recipients: &[&ed25519::PublicKey]) -> Result<Box<[u8]>> {
// Precondition checks
if plaintext.is_empty() {
return Err(Error::EmptyPlaintext);
}
if recipients.is_empty() || recipients.len() > MAX_RECIPIENTS as usize{
if recipients.is_empty() || recipients.len() > MAX_RECIPIENTS as usize {
return Err(Error::BadRecipientCount);
}
@ -61,7 +55,7 @@ fn cipher(plaintext : &[u8], recipients : &[&ed25519::PublicKey]) -> Result<Box<
// Generated random 32-byte secret key used to encrypt the message body
let y = sodiumoxide::crypto::secretbox::gen_key();
// Encrypt the plaintext with y, with a random nonce
let nonce = secretbox::gen_nonce();
let cipher_message = secretbox::seal(plaintext, &nonce, &y);
@ -69,15 +63,16 @@ fn cipher(plaintext : &[u8], recipients : &[&ed25519::PublicKey]) -> Result<Box<
// The sender uses scalar multiplication to derive a shared secret for each recipient,
// and encrypts (number_of_recipents || y) for each one
let h_sk_scalar = &h_sk.to_curve25519();
let mut plain_header = [0u8;RECIPIENT_COUNT_LEN+secretbox::KEYBYTES];
plain_header[0]=recipients.len() as u8;
let mut plain_header = [0u8; RECIPIENT_COUNT_LEN + secretbox::KEYBYTES];
plain_header[0] = recipients.len() as u8;
plain_header[RECIPIENT_COUNT_LEN..].copy_from_slice(&y[..]);
let mut buffer : Vec<u8> = Vec::with_capacity(
secretbox::NONCEBYTES
+ ed25519::PUBLICKEYBYTES
+ ENCRYPTED_HEADER_LEN*recipients.len()
+ secretbox::MACBYTES+plaintext.len()
let mut buffer: Vec<u8> = Vec::with_capacity(
secretbox::NONCEBYTES
+ ed25519::PUBLICKEYBYTES
+ ENCRYPTED_HEADER_LEN * recipients.len()
+ secretbox::MACBYTES
+ plaintext.len(),
);
buffer.extend_from_slice(&nonce[..]);
@ -87,20 +82,18 @@ fn cipher(plaintext : &[u8], recipients : &[&ed25519::PublicKey]) -> Result<Box<
let key = curve25519::scalarmult(&h_sk_scalar, &recipient.to_curve25519())
.map_err(|_| Error::CryptoScalarMultFailed)?;
let key = secretbox::Key::from_slice(&key[..])
.ok_or(Error::CryptoKeyFromGrupFailed)?;
let key = secretbox::Key::from_slice(&key[..]).ok_or(Error::CryptoKeyFromGrupFailed)?;
buffer.extend_from_slice(&secretbox::seal(&plain_header[..],&nonce, &key));
buffer.extend_from_slice(&secretbox::seal(&plain_header[..], &nonce, &key));
}
buffer.extend_from_slice(&cipher_message[..]);
Ok(buffer.into_boxed_slice())
}
fn decipher(ciphertext : &[u8], sk : &SecretKey) -> Result<Option<Vec<u8>>> {
fn decipher(ciphertext: &[u8], sk: &SecretKey) -> Result<Option<Vec<u8>>> {
let mut cursor = ciphertext;
let nonce = secretbox::Nonce::from_slice(&cursor[..secretbox::NONCEBYTES])
.ok_or(Error::CannotReadNonce)?;
cursor = &cursor[secretbox::NONCEBYTES..];
@ -108,17 +101,16 @@ fn decipher(ciphertext : &[u8], sk : &SecretKey) -> Result<Option<Vec<u8>>> {
let h_pk = curve25519::GroupElement::from_slice(&cursor[..ed25519::PUBLICKEYBYTES])
.ok_or(Error::CannotReadNonce)?;
cursor = &cursor[ed25519::PUBLICKEYBYTES..];
let key = curve25519::scalarmult(&sk.to_curve25519(), &h_pk)
.and_then(|key| secretbox::Key::from_slice(&key[..]).ok_or(()))
.map_err(|_| Error::CannotCreateKey)?;
let mut header_no = 0;
while header_no < MAX_RECIPIENTS && cursor.len() > ENCRYPTED_HEADER_LEN+secretbox::MACBYTES {
while header_no < MAX_RECIPIENTS && cursor.len() > ENCRYPTED_HEADER_LEN + secretbox::MACBYTES {
if let Ok(header) = secretbox::open(&cursor[..ENCRYPTED_HEADER_LEN], &nonce, &key) {
let encrypted_message_offset = ENCRYPTED_HEADER_LEN*(header[0]-header_no) as usize;
let y = secretbox::Key::from_slice(&header[1..])
.ok_or(Error::CannotCreateKey)?;
let encrypted_message_offset = ENCRYPTED_HEADER_LEN * (header[0] - header_no) as usize;
let y = secretbox::Key::from_slice(&header[1..]).ok_or(Error::CannotCreateKey)?;
let plaintext = secretbox::open(&cursor[encrypted_message_offset..], &nonce, &y)
.map_err(|_| Error::FailedToDecipher)?;
return Ok(Some(plaintext));
@ -133,26 +125,26 @@ fn decipher(ciphertext : &[u8], sk : &SecretKey) -> Result<Option<Vec<u8>>> {
#[cfg(test)]
mod test {
use super::*;
use crate::patchwork::IdentitySecret;
use crate::keystore::OwnedIdentity;
#[test]
fn test_msg_cipher_to_one() -> Result<()> {
let (u1_pk, u1_sk) = ed25519::gen_keypair();
let plaintext = "hola".as_bytes();
let ciphertext = cipher(plaintext, &[&u1_pk])?;
let plaintext_1 = decipher(&ciphertext, &u1_sk)?.unwrap();
assert_eq!(plaintext.to_vec(),plaintext_1);
assert_eq!(plaintext.to_vec(), plaintext_1);
Ok(())
}
#[test]
fn test_msg_cipher_to_one_helper() -> Result<()> {
let id = IdentitySecret::new();
let id = OwnedIdentity::create();
let plaintext = "holar";
let ciphertext = privatebox_cipher(plaintext, &[&id.id])?;
assert_eq!(is_privatebox(&ciphertext),true);
assert_eq!(is_privatebox(&ciphertext), true);
let plaintext_1 = privatebox_decipher(&ciphertext, &id.sk)?.unwrap();
assert_eq!(plaintext,plaintext_1);
assert_eq!(plaintext, plaintext_1);
Ok(())
}
@ -163,21 +155,21 @@ mod test {
let plaintext = "hola".as_bytes();
let ciphertext = cipher(plaintext, &[&u1_pk])?;
let plaintext_1 = decipher(&ciphertext, &u1_sk)?;
assert_eq!(None,plaintext_1);
assert_eq!(None, plaintext_1);
Ok(())
}
#[test]
fn test_msg_cipher_to_multiple() -> Result<()> {
let u = (0..7).map(|_| ed25519::gen_keypair()).collect::<Vec<_>>();
let u_pk = u.iter().map(|(pk,_)| pk).collect::<Vec<_>>();
let u_pk = u.iter().map(|(pk, _)| pk).collect::<Vec<_>>();
let plaintext = "hola".as_bytes();
let ciphertext = cipher(plaintext, &u_pk)?;
for (_,sk) in u.iter() {
for (_, sk) in u.iter() {
let plaintext_1 = decipher(&ciphertext, sk)?.unwrap();
assert_eq!(plaintext.to_vec(),plaintext_1);
assert_eq!(plaintext.to_vec(), plaintext_1);
}
Ok(())

22
src/db/error.rs → src/keystore/error.rs
View File

@ -1,11 +1,15 @@
#[derive(Debug)]
pub enum Error {
InvalidSequenceNo,
MismatchReadingSecondSize,
CompressionError(String),
Utf8(std::string::FromUtf8Error),
HomeNotFound,
InvalidConfig,
CryptoFormat(crate::crypto::Error),
Io(std::io::Error),
}
impl From<crate::crypto::Error> for Error {
fn from(err: crate::crypto::Error) -> Self {
Error::CryptoFormat(err)
}
}
impl From<std::io::Error> for Error {
fn from(err: std::io::Error) -> Self {
@ -13,17 +17,11 @@ impl From<std::io::Error> for Error {
}
}
impl From<std::string::FromUtf8Error> for Error {
fn from(err: std::string::FromUtf8Error) -> Self {
Error::Utf8(err)
}
}
impl std::fmt::Display for Error {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
write!(f, "{:?}", self)
}
}
impl std::error::Error for Error { }
pub type Result<T> = std::result::Result<T, Error>;
impl std::error::Error for Error {}
pub type Result<T> = std::result::Result<T, Error>;

20
src/keystore/identity.rs Normal file
View File

@ -0,0 +1,20 @@
use crate::crypto::CURVE_ED25519_SUFFIX;
use sodiumoxide::crypto::sign::ed25519;
#[derive(Debug, Clone)]
pub struct OwnedIdentity {
pub id: String,
pub pk: ed25519::PublicKey,
pub sk: ed25519::SecretKey,
}
impl OwnedIdentity {
pub fn create() -> OwnedIdentity {
let (pk, sk) = ed25519::gen_keypair();
OwnedIdentity {
pk,
sk,
id: format!("@{}{}", base64::encode(&pk), CURVE_ED25519_SUFFIX),
}
}
}

6
src/keystore/mod.rs Normal file
View File

@ -0,0 +1,6 @@
mod error;
mod identity;
pub mod patchwork;
pub use identity::OwnedIdentity;
pub use patchwork::{from_patchwork_config, from_patchwork_local};

51
src/keystore/patchwork.rs Normal file
View File

@ -0,0 +1,51 @@
use std::io;
use std::string::ToString;
use crate::crypto::ToSodiumObject;
use super::error::{Error, Result};
use super::OwnedIdentity;
pub const CURVE_ED25519: &str = "ed25519";
#[derive(Deserialize)]
struct JsonSSBSecret {
id: String,
curve: String,
public: String,
private: String,
}
fn to_ioerr<T: ToString>(err: T) -> io::Error {
io::Error::new(io::ErrorKind::Other, err.to_string())
}
pub fn from_patchwork_local() -> Result<OwnedIdentity> {
let home_dir = dirs::home_dir().ok_or(Error::HomeNotFound)?;
let local_key_file = format!("{}/.ssb/secret", home_dir.to_string_lossy());
let content = std::fs::read_to_string(local_key_file)?;
Ok(from_patchwork_config(content)?)
}
pub fn from_patchwork_config<T: AsRef<str>>(config: T) -> Result<OwnedIdentity> {
// strip all comments
let json = config
.as_ref()
.lines()
.filter(|line| !line.starts_with('#'))
.collect::<Vec<_>>()
.join("");
// parse json
let secret: JsonSSBSecret = serde_json::from_str(json.as_ref()).map_err(to_ioerr)?;
if secret.curve != CURVE_ED25519 {
return Err(Error::InvalidConfig);
}
Ok(OwnedIdentity {
id: secret.id,
pk: secret.public.to_ed25519_pk()?,
sk: secret.private.to_ed25519_sk()?,
})
}

10
src/lib.rs
View File

@ -1,11 +1,13 @@
extern crate kuska_handshake;
#[macro_use]
extern crate serde;
extern crate serde_json;
extern crate async_std;
extern crate serde_json;
pub mod rpc;
pub mod api;
pub mod crypto;
pub mod patchwork;
pub mod discovery;
pub mod feed;
pub mod db;
pub mod keystore;
pub mod rpc;

80
src/patchwork/config.rs
View File

@ -1,80 +0,0 @@
use std::io;
use std::string::ToString;
use sodiumoxide::crypto::sign::ed25519;
use sodiumoxide::crypto::auth;
use crate::crypto::ToSodiumObject;
use super::error::{Error,Result};
const CURVE_ED25519 : &str = "ed25519";
pub const SSB_NET_ID : &str = "d4a1cb88a66f02f8db635ce26441cc5dac1b08420ceaac230839b755845a9ffb";
#[derive(Debug)]
pub struct IdentitySecret {
pub id: String,
pub pk: ed25519::PublicKey,
pub sk: ed25519::SecretKey,
}
#[derive(Deserialize)]
struct JsonSSBSecret {
id: String,
curve: String,
public: String,
private: String,
}
pub fn ssb_net_id() -> auth::Key {
auth::Key::from_slice(&hex::decode(SSB_NET_ID).unwrap()).unwrap()
}
fn to_ioerr<T: ToString>(err: T) -> io::Error {
io::Error::new(io::ErrorKind::Other, err.to_string())
}
#[allow(clippy::new_without_default)]
impl IdentitySecret {
pub fn new() -> IdentitySecret {
let (pk, sk) = ed25519::gen_keypair();
IdentitySecret {
pk, sk,
id : format!("@{}.{}",base64::encode(&pk),CURVE_ED25519),
}
}
pub fn from_local_config() -> Result<IdentitySecret> {
let home_dir = dirs::home_dir().ok_or(Error::HomeNotFound)?;
let local_key_file = format!("{}/.ssb/secret",home_dir.to_string_lossy());
let content = std::fs::read_to_string(local_key_file)?;
Ok(IdentitySecret::from_config(content)?)
}
pub fn from_config<T : AsRef<str>>(config: T) -> Result<IdentitySecret> {
// strip all comments
let json = config.as_ref()
.lines()
.filter(|line| !line.starts_with('#'))
.collect::<Vec<_>>()
.join("");
// parse json
let secret : JsonSSBSecret = serde_json::from_str(json.as_ref())
.map_err(to_ioerr)?;
if secret.curve != CURVE_ED25519 {
return Err(Error::InvalidConfig);
}
Ok(IdentitySecret {
id : secret.id,
pk : secret.public.to_ed25519_pk()?,
sk : secret.private.to_ed25519_sk()?,
})
}
}

12
src/patchwork/mod.rs
View File

@ -1,12 +0,0 @@
mod api;
mod config;
mod messagetypes;
mod error;
pub mod pubs;
pub use api::{
parse_feed, parse_latest, parse_message, parse_whoami, ApiClient, CreateHistoryStreamArgs,
CreateStreamArgs,
};
pub use config::{ssb_net_id, IdentitySecret};
pub use error::{Error,Result};

212
src/rpc/client.rs
View File

@ -1,212 +0,0 @@
use super::error::{Error,Result};
use async_std::io;
use async_std::prelude::*;
use kuska_handshake::async_std::{
BoxStreamRead,
BoxStreamWrite
};
pub type RequestNo = i32;
const HEADER_SIZE : usize = 9;
const RPC_HEADER_STREAM_FLAG : u8 = 1 << 3;
const RPC_HEADER_END_OR_ERROR_FLAG : u8 = 1 << 2;
const RPC_HEADER_BODY_TYPE_MASK : u8 = 0b11;
#[derive(Debug,PartialEq)]
pub enum BodyType {
Binary,
UTF8,
JSON,
}
#[derive(Debug,PartialEq)]
pub enum RpcType {
Async,
Source,
}
impl RpcType {
pub fn rpc_id(&self) -> &'static str {
match self {
RpcType::Async => "async",
RpcType::Source => "source",
}
}
}
#[derive(Debug,PartialEq)]
pub struct Header {
pub req_no : RequestNo,
pub is_stream : bool,
pub is_end_or_error : bool,
pub body_type : BodyType,
pub body_len : u32,
}
impl Header {
pub fn from_slice(bytes: &[u8]) -> Result<Header> {
if bytes.len() < HEADER_SIZE {
return Err(Error::HeaderSizeTooSmall);
}
let is_stream = (bytes[0] & RPC_HEADER_STREAM_FLAG) == RPC_HEADER_STREAM_FLAG;
let is_end_or_error = (bytes[0] & RPC_HEADER_END_OR_ERROR_FLAG) == RPC_HEADER_END_OR_ERROR_FLAG;
let body_type = match bytes[0] & RPC_HEADER_BODY_TYPE_MASK {
0 => BodyType::Binary,
1 => BodyType::UTF8,
2 => BodyType::JSON,
_ => return Err(Error::InvalidBodyType),
};
let mut body_len_buff = [0u8;4];
body_len_buff.copy_from_slice(&bytes[1..5]);
let body_len = u32::from_be_bytes(body_len_buff);
let mut reqno_buff = [0u8;4];
reqno_buff.copy_from_slice(&bytes[5..9]);
let req_no = i32::from_be_bytes(reqno_buff);
Ok(Header{
req_no, is_stream, is_end_or_error, body_type, body_len
})
}
pub fn to_array(&self) -> [u8;9] {
let mut flags : u8 = 0;
if self.is_end_or_error {
flags |= RPC_HEADER_END_OR_ERROR_FLAG;
}
if self.is_stream {
flags |= RPC_HEADER_STREAM_FLAG;
}
flags |= match self.body_type {
BodyType::Binary => 0,
BodyType::UTF8 => 1,
BodyType::JSON => 2,
};
let len = self.body_len.to_be_bytes();
let req_no = self.req_no.to_be_bytes();
let mut encoded = [0u8;9];
encoded[0] = flags;
encoded[1..5].copy_from_slice(&len[..]);
encoded[5..9].copy_from_slice(&req_no[..]);
encoded
}
}
pub struct RpcClient<R : io::Read + Unpin, W : io::Write + Unpin> {
box_reader : BoxStreamRead<R>,
box_writer : BoxStreamWrite<W>,
req_no : RequestNo,
}
impl<R:io::Read+Unpin , W:io::Write+Unpin> RpcClient<R,W> {
pub fn new(box_reader :BoxStreamRead<R>, box_writer :BoxStreamWrite<W>) -> RpcClient<R,W> {
RpcClient { box_reader, box_writer, req_no : 0 }
}
pub async fn recv(&mut self) -> Result<(Header,Vec<u8>)> {
let mut rpc_header_raw = [0u8;9];
self.box_reader.read_exact(&mut rpc_header_raw[..]).await?;
let rpc_header = Header::from_slice(&rpc_header_raw[..])?;
let mut rpc_body : Vec<u8> = vec![0;rpc_header.body_len as usize];
self.box_reader.read_exact(&mut rpc_body[..]).await?;
Ok((rpc_header,rpc_body))
}
pub async fn send<T:serde::Serialize>(&mut self, name : &[&str], rpc_type: RpcType, args :&T) -> Result<RequestNo>{
self.req_no+=1;
let mut body = String::from("{\"name\":");
body.push_str(&serde_json::to_string(&name)?);
body.push_str(",\"type\":\"");
body.push_str(rpc_type.rpc_id());
body.push_str("\",\"args\":[");
body.push_str(&serde_json::to_string(&args)?);
body.push_str("]}");
let rpc_header = Header {
req_no : self.req_no,
is_stream : rpc_type == RpcType::Source,
is_end_or_error : false,
body_type : BodyType::JSON,
body_len : body.len() as u32,
}.to_array();
self.box_writer.write_all(&rpc_header[..]).await?;
self.box_writer.write_all(body.as_bytes()).await?;
self.box_writer.flush().await?;
Ok(self.req_no)
}
pub async fn send_cancel_stream(&mut self, req_no: RequestNo) -> Result<()> {
let body_bytes = b"true";
let rpc_header = Header {
req_no,
is_stream : true,
is_end_or_error : true,
body_type : BodyType::JSON,
body_len : body_bytes.len() as u32,
}.to_array();
self.box_writer.write_all(&rpc_header[..]).await?;
self.box_writer.write_all(&body_bytes[..]).await?;
Ok(())
}
pub async fn close(&mut self) -> Result<()> {
self.box_writer.goodbye().await?;
Ok(())
}
}
#[cfg(test)]
mod test {
use super::{Header,BodyType};
#[test]
fn test_header_encoding_1() {
let h = Header::from_slice(&(Header{
req_no : 5,
is_stream : true,
is_end_or_error : false,
body_type : BodyType::JSON,
body_len : 123,
}.to_array())[..]).unwrap();
assert_eq!(h.req_no,5);
assert_eq!(h.is_stream, true);
assert_eq!(h.is_end_or_error, false);
assert_eq!(h.body_type, BodyType::JSON);
assert_eq!(h.body_len, 123);
}
#[test]
fn test_header_encoding_2() {
let h = Header::from_slice(&(Header{
req_no : -5,
is_stream : false,
is_end_or_error : true,
body_type : BodyType::Binary,
body_len : 2123,
}.to_array())[..]).unwrap();
assert_eq!(h.req_no,-5);
assert_eq!(h.is_stream, false);
assert_eq!(h.is_end_or_error, true);
assert_eq!(h.body_type, BodyType::Binary);
assert_eq!(h.body_len, 2123);
}
}

3
src/rpc/error.rs
View File

@ -23,7 +23,6 @@ impl std::fmt::Display for Error {
write!(f, "{:?}", self)
}
}
impl std::error::Error for Error { }
impl std::error::Error for Error {}
pub type Result<T> = std::result::Result<T, Error>;

6
src/rpc/mod.rs
View File

@ -1,5 +1,5 @@
mod client;
mod error;
mod stream;
pub use client::{RpcClient,Header, RequestNo, RpcType};
pub use error::{Error,Result};
pub use error::{Error, Result};
pub use stream::{Body, BodyType, RecvMsg, RequestNo, RpcStream, RpcType};

310
src/rpc/stream.rs Normal file
View File

@ -0,0 +1,310 @@
use super::error::{Error, Result};
use async_std::io;
use async_std::prelude::*;
use kuska_handshake::async_std::{BoxStreamRead, BoxStreamWrite};
pub type RequestNo = i32;
const HEADER_SIZE: usize = 9;
const RPC_HEADER_STREAM_FLAG: u8 = 1 << 3;
const RPC_HEADER_END_OR_ERROR_FLAG: u8 = 1 << 2;
const RPC_HEADER_BODY_TYPE_MASK: u8 = 0b11;
#[derive(Copy, Clone, Debug, PartialEq)]
pub enum BodyType {
Binary,
UTF8,
JSON,
}
#[derive(Deserialize)]
pub struct Body {
pub name: Vec<String>,
#[serde(rename = "type")]
pub rpc_type: RpcType,
pub args: serde_json::Value,
}
#[derive(Serialize)]
pub struct BodyRef<'a, T: serde::Serialize> {
pub name: &'a [&'a str],
#[serde(rename = "type")]
pub rpc_type: RpcType,
pub args: &'a T,
}
#[derive(Clone, Copy, Serialize, Deserialize, Debug, PartialEq)]
pub enum RpcType {
#[serde(rename = "async")]
Async,
#[serde(rename = "source")]
Source,
}
#[derive(Debug, PartialEq)]
pub struct Header {
pub req_no: RequestNo,
pub is_stream: bool,
pub is_end_or_error: bool,
pub body_type: BodyType,
pub body_len: u32,
}
#[derive(Serialize, Deserialize)]
struct ErrorMessage<'a> {
name: &'a str,
message: &'a str,
}
impl Header {
pub fn from_slice(bytes: &[u8]) -> Result<Header> {
if bytes.len() < HEADER_SIZE {
return Err(Error::HeaderSizeTooSmall);
}
let is_stream = (bytes[0] & RPC_HEADER_STREAM_FLAG) == RPC_HEADER_STREAM_FLAG;
let is_end_or_error =
(bytes[0] & RPC_HEADER_END_OR_ERROR_FLAG) == RPC_HEADER_END_OR_ERROR_FLAG;
let body_type = match bytes[0] & RPC_HEADER_BODY_TYPE_MASK {
0 => BodyType::Binary,
1 => BodyType::UTF8,
2 => BodyType::JSON,
_ => return Err(Error::InvalidBodyType),
};
let mut body_len_buff = [0u8; 4];
body_len_buff.copy_from_slice(&bytes[1..5]);
let body_len = u32::from_be_bytes(body_len_buff);
let mut reqno_buff = [0u8; 4];
reqno_buff.copy_from_slice(&bytes[5..9]);
let req_no = i32::from_be_bytes(reqno_buff);
Ok(Header {
req_no,
is_stream,
is_end_or_error,
body_type,
body_len,
})
}
pub fn to_array(&self) -> [u8; 9] {
let mut flags: u8 = 0;
if self.is_end_or_error {
flags |= RPC_HEADER_END_OR_ERROR_FLAG;
}
if self.is_stream {
flags |= RPC_HEADER_STREAM_FLAG;
}
flags |= match self.body_type {
BodyType::Binary => 0,
BodyType::UTF8 => 1,
BodyType::JSON => 2,
};
let len = self.body_len.to_be_bytes();
let req_no = self.req_no.to_be_bytes();
let mut encoded = [0u8; 9];
encoded[0] = flags;
encoded[1..5].copy_from_slice(&len[..]);
encoded[5..9].copy_from_slice(&req_no[..]);
encoded
}
}
pub struct RpcStream<R: io::Read + Unpin, W: io::Write + Unpin> {
box_reader: BoxStreamRead<R>,
box_writer: BoxStreamWrite<W>,
req_no: RequestNo,
}
pub enum RecvMsg {
Request(Body),
ErrorResponse(String),
CancelStreamRespose(),
BodyResponse(Vec<u8>),
}
impl<R: io::Read + Unpin, W: io::Write + Unpin> RpcStream<R, W> {
pub fn new(box_reader: BoxStreamRead<R>, box_writer: BoxStreamWrite<W>) -> RpcStream<R, W> {
RpcStream {
box_reader,
box_writer,
req_no: 0,
}
}
pub async fn recv(&mut self) -> Result<(RequestNo, RecvMsg)> {
let mut rpc_header_raw = [0u8; 9];
self.box_reader.read_exact(&mut rpc_header_raw[..]).await?;
let rpc_header = Header::from_slice(&rpc_header_raw[..])?;
let mut body_raw: Vec<u8> = vec![0; rpc_header.body_len as usize];
self.box_reader.read_exact(&mut body_raw[..]).await?;
if rpc_header.req_no > 0 {
let rpc_body = serde_json::from_slice(&body_raw)?;
Ok((rpc_header.req_no, RecvMsg::Request(rpc_body)))
} else if rpc_header.is_end_or_error {
if rpc_header.is_stream {
Ok((-rpc_header.req_no, RecvMsg::CancelStreamRespose()))
} else {
let err: ErrorMessage = serde_json::from_slice(&body_raw)?;
Ok((
-rpc_header.req_no,
RecvMsg::ErrorResponse(err.message.to_string()),
))
}
} else {
Ok((-rpc_header.req_no, RecvMsg::BodyResponse(body_raw)))
}
}
pub async fn send_request<T: serde::Serialize>(
&mut self,
name: &[&str],
rpc_type: RpcType,
args: &T,
) -> Result<RequestNo> {
self.req_no += 1;
let body_str = serde_json::to_string(&BodyRef {
name,
rpc_type,
args: &[&args],
})?;
let rpc_header = Header {
req_no: self.req_no,
is_stream: rpc_type == RpcType::Source,
is_end_or_error: false,
body_type: BodyType::JSON,
body_len: body_str.as_bytes().len() as u32,
}
.to_array();
self.box_writer.write_all(&rpc_header[..]).await?;
self.box_writer.write_all(body_str.as_bytes()).await?;
self.box_writer.flush().await?;
Ok(self.req_no)
}
pub async fn send_response(
&mut self,
req_no: RequestNo,
rpc_type: RpcType,
body_type: BodyType,
body: &[u8],
) -> Result<()> {
let rpc_header = Header {
req_no: -req_no,
is_stream: rpc_type == RpcType::Source,
is_end_or_error: false,
body_type,
body_len: body.len() as u32,
}
.to_array();
self.box_writer.write_all(&rpc_header[..]).await?;
self.box_writer.write_all(body).await?;
self.box_writer.flush().await?;
Ok(())
}
pub async fn send_error(&mut self, req_no: RequestNo, message: &str) -> Result<()> {
let body_bytes = serde_json::to_string(&ErrorMessage {
name: "Error",
message,
})?;
let rpc_header = Header {
req_no: -req_no,
is_stream: false,
is_end_or_error: true,
body_type: BodyType::UTF8,
body_len: body_bytes.as_bytes().len() as u32,
}
.to_array();
self.box_writer.write_all(&rpc_header[..]).await?;
self.box_writer.write_all(body_bytes.as_bytes()).await?;
self.box_writer.flush().await?;
Ok(())
}
pub async fn send_stream_eof(&mut self, req_no: RequestNo) -> Result<()> {
let body_bytes = b"true";
let rpc_header = Header {
req_no: -req_no,
is_stream: true,
is_end_or_error: true,
body_type: BodyType::JSON,
body_len: body_bytes.len() as u32,
}
.to_array();
self.box_writer.write_all(&rpc_header[..]).await?;
self.box_writer.write_all(&body_bytes[..]).await?;
self.box_writer.flush().await?;
Ok(())
}
pub async fn close(&mut self) -> Result<()> {
self.box_writer.goodbye().await?;
Ok(())
}
}
#[cfg(test)]
mod test {
use super::{BodyType, Header};
#[test]
fn test_header_encoding_1() {
let h = Header::from_slice(
&(Header {
req_no: 5,
is_stream: true,
is_end_or_error: false,
body_type: BodyType::JSON,
body_len: 123,
}
.to_array())[..],
)
.unwrap();
assert_eq!(h.req_no, 5);
assert_eq!(h.is_stream, true);
assert_eq!(h.is_end_or_error, false);
assert_eq!(h.body_type, BodyType::JSON);
assert_eq!(h.body_len, 123);
}
#[test]
fn test_header_encoding_2() {
let h = Header::from_slice(
&(Header {
req_no: -5,
is_stream: false,
is_end_or_error: true,
body_type: BodyType::Binary,
body_len: 2123,
}
.to_array())[..],
)
.unwrap();
assert_eq!(h.req_no, -5);
assert_eq!(h.is_stream, false);
assert_eq!(h.is_end_or_error, true);
assert_eq!(h.body_type, BodyType::Binary);
assert_eq!(h.body_len, 2123);
}
}