First commit

Author: madeye

.gitignore

@@ -0,0 +1,4 @@
+/target
+*.iml
+/.idea
+Cargo.lock

.rustfmt.toml

@@ -0,0 +1 @@
+use_try_shorthand = true

Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "qtun"
+version = "0.1.0"
+authors = ["Max Lv <[email protected]>"]
+repository = "https://github.com/madeye/qtun"
+license = "MIT"
+edition = "2018"
+
+[[bin]]
+name = "qtun-client"
+path = "src/client.rs"
+
+[[bin]]
+name = "qtun-server"
+path = "src/server.rs"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+tokio = { version = "^0.2.7", features = ["full"] }
+bytes = "0.5"
+futures = "0.3"
+quinn = "0.6"
+structopt = "0.3"
+anyhow = "1.0"
+tracing = "0.1"

README.md

@@ -0,0 +1,3 @@
+# qtun
+
+Yet another SIP003 plugin based on IETF-QUIC

src/client.rs

@@ -0,0 +1,84 @@
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use tokio::net::{TcpListener, TcpStream};
+use tokio::prelude::*;
+
+use anyhow::{anyhow, Result};
+use futures::future::try_join;
+use quinn::Endpoint;
+use structopt::{self, StructOpt};
+use tracing::info;
+
+#[derive(StructOpt, Debug)]
+#[structopt(name = "qtun-client")]
+struct Opt {
+    /// Address to listen on
+    #[structopt(long = "local", default_value = "0.0.0.0:4433")]
+    local: SocketAddr,
+    /// Address to listen on
+    #[structopt(long = "remote", default_value = "127.0.0.1:8138")]
+    remote: SocketAddr,
+    /// Override hostname used for certificate verification
+    #[structopt(long = "host", default_value = "bing.com")]
+    host: String,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let options = Opt::from_args();
+
+    let mut endpoint = quinn::Endpoint::builder();
+    let client_config = quinn::ClientConfigBuilder::default();
+    endpoint.default_client_config(client_config.build());
+
+    let (endpoint, _) = endpoint.bind(&"[::]:0".parse().unwrap())?;
+
+    let remote = Arc::<SocketAddr>::from(options.remote);
+    let host = Arc::<String>::from(options.host);
+    let endpoint = Arc::<Endpoint>::from(endpoint);
+
+    let mut listener = TcpListener::bind(options.local).await?;
+
+    while let Ok((inbound, _)) = listener.accept().await {
+        info!("connection incoming");
+
+        let remote = Arc::clone(&remote);
+        let host = Arc::clone(&host);
+        let endpoint = Arc::clone(&endpoint);
+
+        let transfer = transfer(remote, host, endpoint, inbound);
+        tokio::spawn(transfer);
+    }
+
+    Ok(())
+}
+
+async fn transfer(
+    remote: Arc<SocketAddr>,
+    host: Arc<String>,
+    endpoint: Arc<Endpoint>,
+    mut inbound: TcpStream,
+) -> Result<()> {
+    let new_conn = endpoint
+        .connect(&remote, &host)?
+        .await
+        .map_err(|e| anyhow!("failed to connect: {}", e))?;
+
+    let quinn::NewConnection {
+        connection: conn, ..
+    } = { new_conn };
+
+    let (mut ri, mut wi) = inbound.split();
+    let (mut wo, mut ro) = conn
+        .open_bi()
+        .await
+        .map_err(|e| anyhow!("failed to open stream: {}", e))?;
+
+    let client_to_server = io::copy(&mut ri, &mut wo);
+    let server_to_client = io::copy(&mut ro, &mut wi);
+
+    try_join(client_to_server, server_to_client).await?;
+
+    Ok(())
+}

src/server.rs

@@ -0,0 +1,150 @@
+use std::fs;
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use tokio::net::TcpStream;
+use tokio::prelude::*;
+
+use anyhow::{Context, Result};
+use futures::future::try_join;
+use futures::{StreamExt, TryFutureExt};
+use structopt::{self, StructOpt};
+use tracing::{error, info};
+
+#[derive(StructOpt, Debug)]
+#[structopt(name = "qtun-server")]
+struct Opt {
+    /// TLS private key in PEM format
+    #[structopt(
+        parse(from_os_str),
+        short = "k",
+        long = "key",
+        requires = "cert",
+        default_value = "key.der"
+    )]
+    key: PathBuf,
+    /// TLS certificate in PEM format
+    #[structopt(
+        parse(from_os_str),
+        short = "c",
+        long = "cert",
+        requires = "key",
+        default_value = "cert.der"
+    )]
+    cert: PathBuf,
+    /// Enable stateless retries
+    #[structopt(long = "stateless-retry")]
+    stateless_retry: bool,
+    /// Address to listen on
+    #[structopt(long = "local", default_value = "0.0.0.0:4433")]
+    local: SocketAddr,
+    /// Address to listen on
+    #[structopt(long = "remote", default_value = "127.0.0.1:8138")]
+    remote: SocketAddr,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let options = Opt::from_args();
+
+    let mut transport_config = quinn::TransportConfig::default();
+    transport_config.stream_window_uni(0);
+    let mut server_config = quinn::ServerConfig::default();
+    server_config.transport = Arc::new(transport_config);
+    let mut server_config = quinn::ServerConfigBuilder::new(server_config);
+
+    if options.stateless_retry {
+        server_config.use_stateless_retry(true);
+    }
+
+    // load certificates
+    let key_path = &options.key;
+    let cert_path = &options.cert;
+    let key = fs::read(key_path).context("failed to read private key")?;
+    let key = if key_path.extension().map_or(false, |x| x == "der") {
+        quinn::PrivateKey::from_der(&key)?
+    } else {
+        quinn::PrivateKey::from_pem(&key)?
+    };
+    let cert_chain = fs::read(cert_path).context("failed to read certificate chain")?;
+    let cert_chain = if cert_path.extension().map_or(false, |x| x == "der") {
+        quinn::CertificateChain::from_certs(quinn::Certificate::from_der(&cert_chain))
+    } else {
+        quinn::CertificateChain::from_pem(&cert_chain)?
+    };
+    server_config.certificate(cert_chain, key)?;
+
+    let mut endpoint = quinn::Endpoint::builder();
+    endpoint.listen(server_config.build());
+
+    let remote = Arc::<SocketAddr>::from(options.remote);
+
+    let mut incoming = {
+        let (endpoint, incoming) = endpoint.bind(&options.local)?;
+        info!("listening on {}", endpoint.local_addr()?);
+        incoming
+    };
+
+    while let Some(conn) = incoming.next().await {
+        info!("connection incoming");
+        tokio::spawn(
+            handle_connection(remote.clone(), conn).unwrap_or_else(move |e| {
+                error!("connection failed: {reason}", reason = e.to_string())
+            }),
+        );
+    }
+
+    Ok(())
+}
+
+async fn handle_connection(remote: Arc<SocketAddr>, conn: quinn::Connecting) -> Result<()> {
+    let quinn::NewConnection {
+        connection: _,
+        mut bi_streams,
+        ..
+    } = conn.await?;
+
+    async {
+        info!("established");
+
+        // Each stream initiated by the client constitutes a new request.
+        while let Some(stream) = bi_streams.next().await {
+            let stream = match stream {
+                Err(quinn::ConnectionError::ApplicationClosed { .. }) => {
+                    info!("connection closed");
+                    return Ok(());
+                }
+                Err(e) => {
+                    return Err(e);
+                }
+                Ok(s) => s,
+            };
+            tokio::spawn(
+                transfer(remote.clone(), stream)
+                    .unwrap_or_else(move |e| error!("failed: {reason}", reason = e.to_string())),
+            );
+        }
+        Ok(())
+    }
+    .await?;
+
+    Ok(())
+}
+
+async fn transfer(
+    remote: Arc<SocketAddr>,
+    inbound: (quinn::SendStream, quinn::RecvStream),
+) -> Result<()> {
+    let mut outbound = TcpStream::connect(remote.as_ref()).await?;
+
+    let (mut wi, mut ri) = inbound;
+    let (mut ro, mut wo) = outbound.split();
+
+    let client_to_server = io::copy(&mut ri, &mut wo);
+    let server_to_client = io::copy(&mut ro, &mut wi);
+
+    try_join(client_to_server, server_to_client).await?;
+
+    Ok(())
+}