add example program: quasselproxyclient

and inspection proxy for quassel

3 files changed, 280 insertions, 0 deletions

diff --git a/examples/quasselproxy/.gitignore b/examples/quasselproxy/.gitignore
new file mode 100644
index 0000000..ea8c4bf
--- /dev/null
+++ b/examples/quasselproxy/.gitignore
@@ -0,0 +1 @@
+/target
diff --git a/examples/quasselproxy/Cargo.toml b/examples/quasselproxy/Cargo.toml
new file mode 100644
index 0000000..c7d78a8
--- /dev/null
+++ b/examples/quasselproxy/Cargo.toml
@@ -0,0 +1,28 @@
+[package]
+name = "quasselproxy"
+version = "0.1.0"
+authors = ["Max Audron <audron@cocaine.farm>"]
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+tracing = "0.1"
+tracing-subscriber = "0.2"
+tracing-futures = "0.2"
+byteorder = "1.3.2"
+failure = "0.1"
+either = "1.5"
+time = "0.2"
+libquassel = { path = "../../", features = ["framing", "all-quassel-features"] }
+
+bytes = { version = "1.0" }
+flate2 = { version = "1.0", features = ["tokio"] }
+tokio = { version = "1.0", features = ["full"]}
+tokio-util = { version = "0.6", features = ["codec"] }
+tokio-rustls = { version = "0.22" }
+webpki-roots = { version = "0.21" }
+futures-util = { version = "0.3", features = ["std"] }
+futures = { version = "0.3" }
+log = "*"
+pretty_env_logger = { version = "0.4" }
diff --git a/examples/quasselproxy/src/main.rs b/examples/quasselproxy/src/main.rs
new file mode 100644
index 0000000..16c0fa9
--- /dev/null
+++ b/examples/quasselproxy/src/main.rs
@@ -0,0 +1,251 @@
+use failure::Error;
+
+use std::result::Result;
+use std::vec::Vec;
+
+use core::marker::Unpin;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use tokio::net::TcpListener;
+use tokio::net::TcpStream;
+use tokio_util::codec::Framed;
+
+use futures::{
+    stream::{SplitSink, SplitStream},
+    SinkExt, StreamExt,
+};
+
+use libquassel::{
+    frame::QuasselCodec,
+    message::{ConnAck, HandshakeMessage},
+};
+
+use log::{debug, error, info, trace};
+
+#[tokio::main]
+async fn main() -> Result<(), Error> {
+    pretty_env_logger::init();
+
+    let host = std::env::args().nth(1).expect("no host given");
+    let host: Vec<&str> = host.split(':').collect();
+    let username = std::env::args().nth(2).expect("no username given");
+    let password = std::env::args().nth(3).expect("no password given");
+
+    let server = Server::new(
+        "test",
+        ServerSettings {
+            tls: false,
+            compression: false,
+            host: host[0].to_string(),
+            port: host[1].parse().unwrap(),
+            username,
+            password,
+        },
+    );
+
+    //
+    // Start Server Connection
+    //
+
+    let mut s_server =
+        TcpStream::connect(format!("{}:{}", server.settings.host, server.settings.port)).await?;
+
+    info!(target: "init", "Establishing Connection");
+    let connack = Server::init(
+        &mut s_server,
+        server.settings.tls,
+        server.settings.compression,
+    )
+    .await?;
+
+    debug!(target: "init", "{:?}", connack);
+
+    let codec = QuasselCodec::builder().compression(false).new_codec();
+    let framed = Framed::new(s_server, codec);
+    let (s_sink, s_stream) = framed.split();
+
+    //
+    // Accept first listerner
+    //
+
+    let listener = TcpListener::bind("0.0.0.0:4243").await.unwrap();
+    let (mut client, _) = listener.accept().await.unwrap();
+
+    //
+    // Setup Listener
+    //
+
+    {
+        let (mut c_stream, mut c_sink) = client.split();
+
+        let mut init = [0; 12];
+        let n = c_stream.peek(&mut init).await.unwrap();
+        c_stream.read(&mut init[..n]).await.unwrap();
+        let init = libquassel::message::Init::parse(&init);
+        debug!("{:?}", init);
+
+        c_sink.write(&[0x0, 0x0, 0x0, 0x2]).await.unwrap();
+    }
+
+    let codec = QuasselCodec::builder().compression(false).new_codec();
+    let framed = Framed::new(client, codec);
+    let (c_sink, c_stream) = framed.split();
+
+    //
+    // Start Processing
+    //
+
+    let s_state = ClientState::Handshake;
+    let c_state = ClientState::Handshake;
+
+    tokio::join!(
+        Server::run(s_stream, c_sink, s_state, "server -> client"),
+        Server::run(c_stream, s_sink, c_state, "client -> server")
+    );
+
+    Ok(())
+}
+
+#[derive(Clone, Debug)]
+pub struct ServerSettings {
+    pub tls: bool,
+    pub compression: bool,
+    pub host: String,
+    pub port: u32,
+    pub username: String,
+    pub password: String,
+}
+
+pub struct Server {
+    server_name: String,
+    settings: ServerSettings,
+}
+
+impl std::fmt::Debug for Server {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let mut fmt = f.debug_struct("Server");
+        fmt.field("settings", &self.settings);
+        fmt.field("name", &self.server_name).finish()
+    }
+}
+
+#[derive(Clone, Debug)]
+pub enum ClientState {
+    Handshake,
+    Connected,
+}
+
+impl Server {
+    fn new(name: &str, settings: ServerSettings) -> Self {
+        Server {
+            server_name: name.to_string(),
+            settings,
+        }
+    }
+
+    // Send the initialization message to the stream
+    async fn init(
+        stream: &mut (impl AsyncRead + AsyncWrite + Unpin),
+        tls: bool,
+        compression: bool,
+    ) -> Result<ConnAck, Error> {
+        use libquassel::Deserialize;
+
+        // Buffer for our initialization
+        let mut init: Vec<u8> = vec![];
+
+        // The handshake message
+        let mut handshake: u32 = 0x42b33f00;
+
+        // If TLS is enabled set the TLS bit on the handshake
+        if tls {
+            info!(target: "init", "Enabled TLS");
+            handshake |= 0x01;
+        }
+
+        // If COMPRESSION is enabled set the COMPRESSION bit on the handshake
+        if compression {
+            info!(target: "init", "Enabled Compression");
+            handshake |= 0x02;
+        }
+
+        // Select Protocol 2: Datastream
+        let mut proto: u32 = 0x00000002;
+
+        // Flag proto as the last protocol
+        let fin: u32 = 0x80000000;
+        proto |= fin;
+
+        // Add handshake and protocol to our buffer
+        init.extend(handshake.to_be_bytes().iter());
+        init.extend(proto.to_be_bytes().iter());
+
+        // Send Buffer
+        stream.write(&init).await?;
+
+        // Read Response
+        let mut buf = [0; 4];
+        stream.read(&mut buf).await?;
+
+        let (_, connack) = ConnAck::parse(&buf)?;
+        Ok(connack)
+    }
+
+    pub async fn run(
+        mut stream: SplitStream<Framed<TcpStream, QuasselCodec>>,
+        mut sink: SplitSink<Framed<TcpStream, QuasselCodec>, Vec<u8>>,
+        mut state: ClientState,
+        direction: &str,
+    ) {
+        // Start event loop
+        while let Some(msg) = stream.next().await {
+            let msg = msg.unwrap();
+            sink.send(msg.to_vec()).await.unwrap();
+            match state {
+                ClientState::Handshake => Server::handle_login_message(&msg, &mut state, direction)
+                    .await
+                    .unwrap(),
+                ClientState::Connected => Server::handle_message(&msg, direction).await.unwrap(),
+            }
+        }
+    }
+
+    #[tracing::instrument]
+    async fn handle_login_message(
+        buf: &[u8],
+        state: &mut ClientState,
+        direction: &str,
+    ) -> Result<(), Error> {
+        use libquassel::HandshakeDeserialize;
+
+        trace!(target: "message", "Received bytes: {:x?}", buf);
+        match HandshakeMessage::parse(buf) {
+            Ok((_size, res)) => {
+                info!("{}: {:#?}", direction, res);
+
+                match res {
+                    HandshakeMessage::SessionInit(_) => *state = ClientState::Connected,
+                    HandshakeMessage::ClientLogin(_) => *state = ClientState::Connected,
+                    _ => {}
+                }
+            }
+            Err(e) => error!("failed to parse handshake message {}", e),
+        }
+
+        Ok(())
+    }
+
+    #[tracing::instrument]
+    async fn handle_message(buf: &[u8], direction: &str) -> Result<(), Error> {
+        use libquassel::message::Message;
+        use libquassel::Deserialize;
+
+        trace!(target: "message", "Received bytes: {:x?}", buf);
+
+        match Message::parse(buf) {
+            Ok((_size, res)) => info!("{}: {:#?}", direction, res),
+            Err(e) => error!("failed to parse message {}", e),
+        }
+
+        return Ok(());
+    }
+}