initial commit

.gitignore

@@ -0,0 +1 @@
+/target

Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "golgi"
+version = "0.1.0"
+authors = ["glyph <glyph@mycelial.technology>"]
+edition = "2021"
+
+[dependencies]
+async-std = "1.10.0"
+base64 = "0.13.0"
+futures = "0.3.18"
+hex = "0.4.3"
+kuska-handshake = { version = "0.2.0", features = ["async_std"] }
+kuska-sodiumoxide = "0.2.5-0"
+# waiting for a pr merge upstream
+kuska-ssb = { path = "../ssb" }
+# try to replace with miniserde
+serde = "1"
+serde_json = "1"

README.md

@@ -0,0 +1,16 @@
+# golgi
+
+_A Scuttlebutt client written in Rust_
+
+An experimental client which uses the [kuska-ssb](https://github.com/Kuska-ssb) libraries and aims to provide a high-level API for interacting with an sbot instance. Development efforts are currently oriented towards [go-sbot](https://github.com/cryptoscope/ssb) interoperability.
+
+## Example Usage
+
+```rust
+pub async fn run() -> Result<(), GolgiError> {
+    let mut sbot_client = Sbot::init(None, None).await?;
+
+    let id = sbot_client.whoami().await?;
+    println!("{}", id);
+}
+```

src/error.rs

@@ -0,0 +1,106 @@
+/*
+use async_std::{io::Read, net::TcpStream};
+use std::fmt::Debug;
+
+use kuska_handshake::async_std::BoxStream;
+use kuska_sodiumoxide::crypto::sign::ed25519;
+use kuska_ssb::api::{dto::WhoAmIOut, ApiCaller};
+use kuska_ssb::discovery;
+use kuska_ssb::keystore;
+use kuska_ssb::keystore::OwnedIdentity;
+use kuska_ssb::rpc::{RecvMsg, RequestNo, RpcReader, RpcWriter};
+
+type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;
+
+pub fn whoami_res_parse(body: &[u8]) -> Result<WhoAmIOut> {
+    Ok(serde_json::from_slice(body)?)
+}
+*/
+
+#[derive(Debug)]
+pub enum GolgiError {
+    DecodeBase64(base64::DecodeError),
+    Io {
+        source: std::io::Error,
+        context: String,
+    },
+    Handshake(kuska_handshake::async_std::Error),
+    KuskaApi(kuska_ssb::api::Error),
+    KuskaFeed(kuska_ssb::feed::Error),
+    KuskaRpc(kuska_ssb::rpc::Error),
+    // error message returned from the go-sbot
+    Sbot(String),
+    SerdeJson(serde_json::Error),
+    WhoAmI(String),
+}
+
+impl std::error::Error for GolgiError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match *self {
+            GolgiError::DecodeBase64(ref err) => Some(err),
+            GolgiError::Io { ref source, .. } => Some(source),
+            GolgiError::Handshake(_) => None,
+            GolgiError::KuskaApi(ref err) => Some(err),
+            GolgiError::KuskaFeed(ref err) => Some(err),
+            GolgiError::KuskaRpc(ref err) => Some(err),
+            GolgiError::Sbot(_) => None,
+            GolgiError::SerdeJson(ref err) => Some(err),
+            GolgiError::WhoAmI(_) => None,
+        }
+    }
+}
+
+impl std::fmt::Display for GolgiError {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        match *self {
+            // TODO: add context (what were we trying to decode?)
+            GolgiError::DecodeBase64(_) => write!(f, "Failed to decode base64"),
+            GolgiError::Io { ref context, .. } => write!(f, "IO error: {}", context),
+            GolgiError::Handshake(ref err) => write!(f, "{}", err),
+            GolgiError::KuskaApi(_) => write!(f, "SSB API failure"),
+            GolgiError::KuskaFeed(_) => write!(f, "SSB feed error"),
+            // TODO: improve this variant with a context message
+            // then have the core display msg be: "SSB RPC error: {}", context
+            GolgiError::KuskaRpc(_) => write!(f, "SSB RPC failure"),
+            GolgiError::Sbot(ref err) => write!(f, "Sbot returned an error response: {}", err),
+            GolgiError::SerdeJson(_) => write!(f, "Failed to serialize JSON slice"),
+            GolgiError::WhoAmI(ref err) => write!(f, "{}", err),
+        }
+    }
+}
+
+impl From<base64::DecodeError> for GolgiError {
+    fn from(err: base64::DecodeError) -> Self {
+        GolgiError::DecodeBase64(err)
+    }
+}
+
+impl From<kuska_handshake::async_std::Error> for GolgiError {
+    fn from(err: kuska_handshake::async_std::Error) -> Self {
+        GolgiError::Handshake(err)
+    }
+}
+
+impl From<kuska_ssb::api::Error> for GolgiError {
+    fn from(err: kuska_ssb::api::Error) -> Self {
+        GolgiError::KuskaApi(err)
+    }
+}
+
+impl From<kuska_ssb::feed::Error> for GolgiError {
+    fn from(err: kuska_ssb::feed::Error) -> Self {
+        GolgiError::KuskaFeed(err)
+    }
+}
+
+impl From<kuska_ssb::rpc::Error> for GolgiError {
+    fn from(err: kuska_ssb::rpc::Error) -> Self {
+        GolgiError::KuskaRpc(err)
+    }
+}
+
+impl From<serde_json::Error> for GolgiError {
+    fn from(err: serde_json::Error) -> Self {
+        GolgiError::SerdeJson(err)
+    }
+}

src/lib.rs

@@ -0,0 +1,106 @@
+mod error;
+mod utils;
+
+use async_std::net::TcpStream;
+//use futures::io::{self, AsyncRead as Read, AsyncWrite as Write};
+use kuska_handshake::async_std::{BoxStream, BoxStreamRead, BoxStreamWrite};
+use kuska_handshake::HandshakeComplete;
+use kuska_sodiumoxide::crypto::{auth, sign::ed25519};
+use kuska_ssb::api::{dto::CreateHistoryStreamIn, ApiCaller};
+use kuska_ssb::discovery;
+use kuska_ssb::keystore;
+use kuska_ssb::keystore::OwnedIdentity;
+use kuska_ssb::rpc::{RpcReader, RpcWriter};
+
+use crate::error::GolgiError;
+
+struct Sbot {
+    id: String,
+    public_key: ed25519::PublicKey,
+    private_key: ed25519::SecretKey,
+    address: String,
+    // aka caps key (scuttleverse identifier)
+    network_id: auth::Key,
+    client: ApiCaller<TcpStream>,
+    rpc_reader: RpcReader<TcpStream>,
+}
+
+impl Sbot {
+    async fn init(ip_port: Option<String>, net_id: Option<String>) -> Result<Sbot, GolgiError> {
+        let address;
+        if ip_port.is_none() {
+            address = "127.0.0.1:8008".to_string();
+        } else {
+            address = ip_port.unwrap();
+        }
+
+        let network_id;
+        if net_id.is_none() {
+            network_id = discovery::ssb_net_id();
+        } else {
+            network_id = auth::Key::from_slice(&hex::decode(net_id.unwrap()).unwrap()).unwrap();
+        }
+
+        let OwnedIdentity { pk, sk, id } = keystore::from_gosbot_local()
+            .await
+            .expect("couldn't read local secret");
+
+        let socket = TcpStream::connect(&address)
+            .await
+            .map_err(|source| GolgiError::Io {
+                source,
+                context: "socket error; failed to initiate tcp stream connection".to_string(),
+            })?;
+
+        let handshake = kuska_handshake::async_std::handshake_client(
+            &mut &socket,
+            network_id.clone(),
+            pk,
+            sk.clone(),
+            pk,
+        )
+        .await
+        .map_err(GolgiError::Handshake)?;
+
+        let (box_stream_read, box_stream_write) =
+            BoxStream::from_handshake(socket.clone(), socket, handshake, 0x8000).split_read_write();
+
+        let rpc_reader = RpcReader::new(box_stream_read);
+        let client = ApiCaller::new(RpcWriter::new(box_stream_write));
+
+        Ok(Self {
+            id,
+            public_key: pk,
+            private_key: sk,
+            address,
+            network_id,
+            client: client,
+            rpc_reader: rpc_reader,
+        })
+    }
+
+    async fn whoami(&mut self) -> Result<String, GolgiError> {
+        let req_id = self.client.whoami_req_send().await?;
+
+        utils::get_async(&mut self.rpc_reader, req_id, utils::whoami_res_parse)
+            .await
+            .map(|whoami| whoami.id)
+    }
+
+    async fn create_history_stream(&mut self, id: String) -> Result<(), GolgiError> {
+        let args = CreateHistoryStreamIn::new(id);
+        let req_id = self.client.create_history_stream_req_send(&args).await?;
+        utils::print_source_until_eof(&mut self.rpc_reader, req_id, utils::feed_res_parse).await
+    }
+}
+
+pub async fn run() -> Result<(), GolgiError> {
+    let mut sbot_client = Sbot::init(None, None).await?;
+
+    let id = sbot_client.whoami().await?;
+    println!("{}", id);
+
+    sbot_client.create_history_stream(id).await?;
+
+    Ok(())
+}

src/lib.rs_lifetimes

@@ -0,0 +1,161 @@
+mod error;
+mod utils;
+
+use async_std::net::TcpStream;
+//use futures::io::{self, AsyncRead as Read, AsyncWrite as Write};
+use kuska_handshake::async_std::{BoxStream, BoxStreamRead, BoxStreamWrite};
+use kuska_handshake::HandshakeComplete;
+use kuska_sodiumoxide::crypto::{auth, sign::ed25519};
+use kuska_ssb::api::{dto::CreateHistoryStreamIn, ApiCaller};
+use kuska_ssb::discovery;
+use kuska_ssb::keystore;
+use kuska_ssb::keystore::OwnedIdentity;
+use kuska_ssb::rpc::{RpcReader, RpcWriter};
+
+use crate::error::GolgiError;
+
+//struct Sbot<R: Read + std::marker::Unpin, W: Write + std::marker::Unpin> {
+struct Sbot<'a> {
+    id: String,
+    public_key: ed25519::PublicKey,
+    private_key: ed25519::SecretKey,
+    address: String,
+    // aka caps key (scuttleverse identifier)
+    network_id: auth::Key,
+    //socket: TcpStream,
+    //client: ApiCaller<W>,
+    client: ApiCaller<TcpStream>,
+    //rpc_reader: RpcReader<R>,
+    rpc_reader: RpcReader<&'a TcpStream>,
+}
+
+//impl<R: Read + std::marker::Unpin, W: Write + std::marker::Unpin> Sbot<R, W> {
+//impl Sbot<'static> {
+impl Sbot<'_> {
+    async fn init(
+        ip_port: Option<String>,
+        net_id: Option<String>,
+    ) -> Result<Sbot<'static>, GolgiError> {
+        let address;
+        if ip_port.is_none() {
+            // set default
+            address = "127.0.0.1:8008".to_string();
+        } else {
+            address = ip_port.unwrap().to_string();
+        }
+
+        let network_id;
+        if net_id.is_none() {
+            network_id = discovery::ssb_net_id();
+        } else {
+            network_id = auth::Key::from_slice(&hex::decode(net_id.unwrap()).unwrap()).unwrap();
+        }
+
+        let OwnedIdentity { pk, sk, id } = keystore::from_gosbot_local()
+            .await
+            .expect("couldn't read local secret");
+
+        let socket = TcpStream::connect(&address)
+            .await
+            .map_err(|source| GolgiError::Io {
+                source,
+                context: "socket error; failed to initiate tcp stream connection".to_string(),
+            })?;
+
+        let handshake = kuska_handshake::async_std::handshake_client(
+            &mut &socket,
+            network_id.clone(),
+            pk,
+            sk.clone(),
+            pk,
+        )
+        .await
+        .map_err(|err| GolgiError::Handshake(err))?;
+
+        let (box_stream_read, box_stream_write) =
+            BoxStream::from_handshake(socket, socket, handshake, 0x8000).split_read_write();
+
+        let rpc_reader = RpcReader::new(box_stream_read);
+        let client = ApiCaller::new(RpcWriter::new(box_stream_write));
+
+        Ok(Self {
+            id,
+            public_key: pk,
+            private_key: sk,
+            address,
+            network_id,
+            client: client,
+            rpc_reader: rpc_reader,
+        })
+    }
+}
+
+pub async fn run() -> Result<(), GolgiError> {
+    let sbot_client = Sbot::init(None, None);
+    /*
+    // read go-sbot id details from `/.ssb-go/secret
+    let OwnedIdentity { pk, sk, id } = keystore::from_gosbot_local()
+        .await
+        .expect("read local secret");
+    println!("connecting with identity {}", id);
+
+    // set the public key, ip and port for the sbot instance
+    let sbot_public_key = "a0SsCiZkBu6qaQ6tWVvzQPDSzvO0JqMAqPXt0LBIl30=".to_string();
+    let server_pk =
+        ed25519::PublicKey::from_slice(&base64::decode(&sbot_public_key)?).expect("bad public key");
+    let server_ipport = "127.0.0.1:8008".to_string();
+
+    // connect to the local go-sbot instance
+    let mut socket = TcpStream::connect(server_ipport)
+        .await
+        .map_err(|source| GolgiError::Io {
+            source,
+            context: "socket error; failed to initiate tcp stream connection".to_string(),
+        })?;
+
+    // initiate secret handshake
+    let handshake = kuska_handshake::async_std::handshake_client(
+        &mut socket,
+        discovery::ssb_net_id(),
+        pk,
+        sk.clone(),
+        server_pk,
+    )
+    .await
+    .expect("handshake error");
+
+    println!("💃 handshake complete");
+
+    // call `whoami`
+    let (box_stream_read, box_stream_write) =
+        BoxStream::from_handshake(&socket, &socket, handshake, 0x8000).split_read_write();
+
+    let mut rpc_reader = RpcReader::new(box_stream_read);
+    let mut client = ApiCaller::new(RpcWriter::new(box_stream_write));
+
+    let req_id = client.whoami_req_send().await?;
+    let whoami = match utils::get_async(&mut rpc_reader, req_id, utils::whoami_res_parse).await {
+        Ok(res) => {
+            println!("😊 server says hello to {}", res.id);
+            id
+        }
+        Err(err) => {
+            if !err
+                .to_string()
+                .contains("method:whoami is not in list of allowed methods")
+            {
+                println!("Cannot ask for whoami {}", err);
+            }
+            id
+        }
+    };
+
+    // call `createhistorystream`
+    let args = CreateHistoryStreamIn::new(whoami.clone());
+    // TODO: this should return an error if the args are wrong but it doesn't?
+    let req_id = client.create_history_stream_req_send(&args).await?;
+    utils::print_source_until_eof(&mut rpc_reader, req_id, utils::feed_res_parse).await?;
+    */
+
+    Ok(())
+}

src/main.rs

@@ -0,0 +1,9 @@
+use std::process;
+
+#[async_std::main]
+async fn main() {
+    if let Err(e) = golgi::run().await {
+        eprintln!("Application error: {}", e);
+        process::exit(1);
+    }
+}

src/utils.rs

@@ -0,0 +1,80 @@
+/*
+
+use kuska_handshake::async_std::BoxStream;
+use kuska_sodiumoxide::crypto::sign::ed25519;
+use kuska_ssb::discovery;
+use kuska_ssb::keystore;
+use kuska_ssb::keystore::OwnedIdentity;
+*/
+use std::fmt::Debug;
+
+use async_std::io::Read;
+
+use kuska_ssb::api::dto::WhoAmIOut;
+use kuska_ssb::feed::Feed;
+use kuska_ssb::rpc::{RecvMsg, RequestNo, RpcReader};
+
+use crate::error::GolgiError;
+
+pub fn feed_res_parse(body: &[u8]) -> Result<Feed, GolgiError> {
+    Ok(Feed::from_slice(body)?)
+}
+
+pub fn whoami_res_parse(body: &[u8]) -> Result<WhoAmIOut, GolgiError> {
+    Ok(serde_json::from_slice(body)?)
+}
+
+pub async fn get_async<'a, R, T, F>(
+    rpc_reader: &mut RpcReader<R>,
+    req_no: RequestNo,
+    f: F,
+) -> Result<T, GolgiError>
+where
+    R: Read + Unpin,
+    F: Fn(&[u8]) -> Result<T, GolgiError>,
+    T: Debug,
+{
+    loop {
+        let (id, msg) = rpc_reader.recv().await?;
+        if id == req_no {
+            match msg {
+                RecvMsg::RpcResponse(_type, body) => {
+                    return f(&body).map_err(|err| err);
+                }
+                RecvMsg::ErrorResponse(message) => {
+                    return Err(GolgiError::Sbot(message));
+                }
+                _ => {}
+            }
+        }
+    }
+}
+
+pub async fn print_source_until_eof<'a, R, T, F>(
+    rpc_reader: &mut RpcReader<R>,
+    req_no: RequestNo,
+    f: F,
+) -> Result<(), GolgiError>
+where
+    R: Read + Unpin,
+    F: Fn(&[u8]) -> Result<T, GolgiError>,
+    T: Debug + serde::Deserialize<'a>,
+{
+    loop {
+        let (id, msg) = rpc_reader.recv().await?;
+        if id == req_no {
+            match msg {
+                RecvMsg::RpcResponse(_type, body) => {
+                    let display = f(&body)?;
+                    println!("{:?}", display);
+                }
+                RecvMsg::ErrorResponse(message) => {
+                    return Err(GolgiError::Sbot(message));
+                }
+                RecvMsg::CancelStreamRespose() => break,
+                _ => {}
+            }
+        }
+    }
+    Ok(())
+}