Got gemini proxy working, via a custom tokio_rustls branch

.dev-config.yml

@@ -21,6 +21,10 @@ service:
             value: hi
           - name: user-agent
             value: ""
+  gemini:
+    proxied_domains:
+      mediocregopher.com:
+        url: gemini://127.0.0.1:1965
   passphrase: foobar
   dns_records:
     - kind: A

Cargo.lock

@@ -2963,9 +2963,8 @@ dependencies = [
 
 [[package]]
 name = "tokio-rustls"
-version = "0.24.0"
+version = "0.24.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
+source = "git+https://code.betamike.com/micropelago/tokio-rustls.git?branch=transparent-acceptor#18fd688b335430e17e054e15ff7d6ce073db2419"
-checksum = "e0d409377ff5b1e3ca6437aa86c1eb7d40c134bfec254e44c830defa92669db5"
 dependencies = [
  "rustls",
  "tokio",

Cargo.toml

@@ -20,7 +20,7 @@ serde_json = "1.0.96"
 trust-dns-client = "0.22.0"
 mockall = "0.11.4"
 thiserror = "1.0.40"
-tokio = { version = "1.28.1", features = [ "full" ]}
+tokio = { version = "1.28.1", features = [ "full", "net" ]}
 signal-hook = "0.3.15"
 futures = "0.3.28"
 signal-hook-tokio = { version = "0.3.1", features = [ "futures-v0_3" ]}
@@ -45,3 +45,6 @@ serde_yaml = "0.9.22"
 rand = "0.8.5"
 reqwest = "0.11.18"
 hyper-reverse-proxy = "0.5.1"
+
+[patch.crates-io]
+tokio-rustls = { git = "https://code.betamike.com/micropelago/tokio-rustls.git", branch = "transparent-acceptor" }

flake.nix

@@ -44,6 +44,8 @@
             pkgs.stdenv.cc
             pkgs.openssl
             toolchain
+
+            pkgs.nmap # ncat
           ];
           shellHook = ''
             source $(pwd)/.env.dev

src/domain/manager.rs

@@ -1,7 +1,6 @@
 use crate::domain::{self, acme, checker, store};
 use crate::error::unexpected::{self, Mappable};
-use crate::origin;
+use crate::{origin, task_stack, util};
-use crate::util;
 
 use std::sync;
 use tokio_util::sync::CancellationToken;
@@ -187,7 +186,7 @@ impl ManagerImpl {
         DomainStore: store::Store + Send + Sync + 'static,
         AcmeManager: acme::manager::Manager + Send + Sync + 'static,
     >(
-        task_stack: &mut util::TaskStack<unexpected::Error>,
+        task_stack: &mut task_stack::TaskStack<unexpected::Error>,
         origin_store: OriginStore,
         domain_store: DomainStore,
         domain_checker: checker::DNSChecker,

src/lib.rs

@@ -4,8 +4,10 @@
 
 pub mod config;
 pub mod domain;
-pub mod error;
 pub mod origin;
 pub mod service;
+pub mod task_stack;
 pub mod token;
-pub mod util;
+
+mod error;
+mod util;

src/main.rs

@@ -128,7 +128,7 @@ async fn main() {
         None
     };
 
-    let mut task_stack = domani::util::TaskStack::new();
+    let mut task_stack = domani::task_stack::TaskStack::new();
 
     let domain_manager = domani::domain::manager::ManagerImpl::new(
         &mut task_stack,
@@ -142,6 +142,12 @@ async fn main() {
         &mut task_stack,
         domain_manager.clone(),
         domain_manager.clone(),
+        config.service.clone(),
+    );
+
+    let _ = domani::service::gemini::Service::new(
+        &mut task_stack,
+        domain_manager.clone(),
         config.service,
     );
 

src/service.rs

@@ -1,5 +1,5 @@
 mod config;
+pub mod gemini;
 pub mod http;
-mod util;
 
 pub use config::*;

src/service/config.rs

@@ -24,7 +24,7 @@ impl From<ConfigDNSRecord> for domain::checker::DNSRecord {
     }
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Clone)]
 pub struct Config {
     pub passphrase: String,
     pub dns_records: Vec<ConfigDNSRecord>,
@@ -32,4 +32,6 @@ pub struct Config {
     pub primary_domain: domain::Name,
     #[serde(default)]
     pub http: service::http::Config,
+    #[serde(default)]
+    pub gemini: service::gemini::Config,
 }

src/service/gemini.rs

@@ -0,0 +1,144 @@
+mod config;
+
+pub use config::*;
+
+use crate::error::unexpected::{self, Mappable};
+use crate::{domain, service, task_stack};
+
+use std::sync;
+use tokio_util::sync::CancellationToken;
+
+pub struct Service {
+    cert_resolver: sync::Arc<dyn rustls::server::ResolvesServerCert>,
+    config: service::Config,
+}
+
+#[derive(thiserror::Error, Debug)]
+enum HandleConnError {
+    #[error("client error: {0}")]
+    ClientError(String),
+
+    #[error(transparent)]
+    Unexpected(#[from] unexpected::Error),
+}
+
+impl Service {
+    pub fn new(
+        task_stack: &mut task_stack::TaskStack<unexpected::Error>,
+        cert_resolver: sync::Arc<dyn rustls::server::ResolvesServerCert>,
+        config: service::Config,
+    ) -> sync::Arc<Service> {
+        let service = sync::Arc::new(Service {
+            cert_resolver,
+            config,
+        });
+        task_stack.push_spawn(|canceller| listen(service.clone(), canceller));
+        service
+    }
+
+    async fn proxy_conn<IO>(
+        &self,
+        proxied_domain: &ConfigProxiedDomain,
+        mut conn: IO,
+    ) -> unexpected::Result<()>
+    where
+        IO: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+    {
+        let mut proxy_conn = tokio::net::TcpStream::connect(&proxied_domain.url.addr)
+            .await
+            .map_unexpected_while(|| {
+                format!("failed to connect to proxy {}", proxied_domain.url.url,)
+            })?;
+
+        _ = tokio::io::copy_bidirectional(&mut conn, &mut proxy_conn).await;
+
+        Ok(())
+    }
+
+    async fn handle_conn(
+        &self,
+        conn: tokio::net::TcpStream,
+        _tls_config: sync::Arc<rustls::ServerConfig>,
+    ) -> Result<(), HandleConnError> {
+        let acceptor =
+            tokio_rustls::TransparentConfigAcceptor::new(rustls::server::Acceptor::default(), conn);
+        futures::pin_mut!(acceptor);
+
+        match acceptor.as_mut().await {
+            Ok(start) => {
+                let client_hello = start.client_hello();
+
+                let domain = client_hello.server_name().ok_or_else(|| {
+                    HandleConnError::ClientError("missing SNI in ClientHello".to_string())
+                })?;
+
+                let domain: domain::Name = domain.parse().map_err(|e| {
+                    HandleConnError::ClientError(format!(
+                        "parsing domain {domain}, provided in SNI: {e}"
+                    ))
+                })?;
+
+                // If the domain should be proxied, then proxy it
+                if let Some(proxied_domain) = self.config.gemini.proxied_domains.get(&domain) {
+                    let conn = start.into_original_stream();
+                    self.proxy_conn(proxied_domain, conn).await?;
+                    return Ok(());
+                }
+
+                return Err(HandleConnError::ClientError(format!(
+                    "unknown domain {domain}"
+                )));
+            }
+            Err(err) => {
+                return Err(unexpected::Error::from(
+                    format!("failed to accept TLS connection: {err}").as_str(),
+                )
+                .into())
+            }
+        }
+    }
+}
+
+async fn listen(
+    service: sync::Arc<Service>,
+    canceller: CancellationToken,
+) -> unexpected::Result<()> {
+    let tls_config = sync::Arc::new(
+        rustls::server::ServerConfig::builder()
+            .with_safe_defaults()
+            .with_no_client_auth() // TODO maybe this isn't right?
+            .with_cert_resolver(service.cert_resolver.clone()),
+    );
+
+    log::info!(
+        "Listening on gemini://{}:{}",
+        &service.config.primary_domain.clone(),
+        &service.config.gemini.gemini_addr.port(),
+    );
+
+    let listener = tokio::net::TcpListener::bind(service.config.gemini.gemini_addr)
+        .await
+        .or_unexpected_while("binding tcp socket")?;
+
+    loop {
+        let (conn, addr) = tokio::select! {
+        res = listener.accept() => res.or_unexpected_while("accepting connection")?,
+        _ = canceller.cancelled() => return Ok(()),
+        };
+
+        let service = service.clone();
+        let tls_config = tls_config.clone();
+
+        tokio::spawn(async move {
+            match service.handle_conn(conn, tls_config).await {
+                Ok(_) => (),
+                Err(HandleConnError::ClientError(e)) => {
+                    log::warn!("Bad request from connection {addr}: {e}")
+                }
+                Err(HandleConnError::Unexpected(e)) => {
+                    log::error!("Server error handling connection {addr}: {e}")
+                }
+            }
+        });
+    }
+}

src/service/gemini/config.rs

@@ -0,0 +1,79 @@
+use crate::domain;
+use crate::error::unexpected::{self, Mappable};
+
+use serde::{Deserialize, Serialize};
+use serde_with::{serde_as, TryFromInto};
+
+use std::{collections, net, str::FromStr};
+
+fn default_gemini_addr() -> net::SocketAddr {
+    net::SocketAddr::from_str("[::]:3965").unwrap()
+}
+
+#[derive(Deserialize, Serialize, Clone)]
+pub struct ConfigProxiedDomainUrl {
+    pub url: String,
+    pub addr: String,
+}
+
+impl From<ConfigProxiedDomainUrl> for String {
+    fn from(url: ConfigProxiedDomainUrl) -> Self {
+        url.url
+    }
+}
+
+impl TryFrom<String> for ConfigProxiedDomainUrl {
+    type Error = unexpected::Error;
+
+    fn try_from(url: String) -> Result<Self, Self::Error> {
+        // use http's implementation, should be the same
+        let parsed = http::Uri::from_str(url.as_str())
+            .map_unexpected_while(|| format!("parsing proxy url {url}"))?;
+
+        let scheme = parsed.scheme().map_unexpected_while(|| {
+            format!("expected a scheme of gemini in the proxy url {url}")
+        })?;
+
+        if scheme != "gemini" {
+            return Err(unexpected::Error::from(
+                format!("scheme of proxy url {url} should be 'gemini'",).as_str(),
+            ));
+        }
+
+        match parsed.authority() {
+            None => Err(unexpected::Error::from(
+                format!("proxy url {url} should have a host",).as_str(),
+            )),
+            Some(authority) => {
+                let port = authority.port().map(|p| p.as_u16()).unwrap_or(1965);
+                Ok(ConfigProxiedDomainUrl {
+                    url: url,
+                    addr: format!("{}:{port}", authority.host()),
+                })
+            }
+        }
+    }
+}
+
+#[serde_as]
+#[derive(Deserialize, Serialize, Clone)]
+pub struct ConfigProxiedDomain {
+    #[serde_as(as = "TryFromInto<String>")]
+    pub url: ConfigProxiedDomainUrl,
+}
+
+#[derive(Deserialize, Serialize, Clone)]
+pub struct Config {
+    #[serde(default = "default_gemini_addr")]
+    pub gemini_addr: net::SocketAddr,
+    pub proxied_domains: collections::HashMap<domain::Name, ConfigProxiedDomain>,
+}
+
+impl Default for Config {
+    fn default() -> Self {
+        Self {
+            gemini_addr: default_gemini_addr(),
+            proxied_domains: Default::default(),
+        }
+    }
+}

src/service/http.rs

@@ -2,6 +2,7 @@ mod config;
 mod proxy;
 mod tasks;
 mod tpl;
+mod util;
 
 pub use config::*;
 
@@ -13,7 +14,7 @@ use std::str::FromStr;
 use std::{future, net, sync};
 
 use crate::error::unexpected;
-use crate::{domain, service, util};
+use crate::{domain, service, task_stack};
 
 pub struct Service {
     domain_manager: sync::Arc<dyn domain::manager::Manager>,
@@ -23,7 +24,7 @@ pub struct Service {
 }
 
 pub fn new(
-    task_stack: &mut util::TaskStack<unexpected::Error>,
+    task_stack: &mut task_stack::TaskStack<unexpected::Error>,
     domain_manager: sync::Arc<dyn domain::manager::Manager>,
     cert_resolver: sync::Arc<dyn rustls::server::ResolvesServerCert>,
     config: service::Config,
@@ -64,7 +65,7 @@ struct DomainInitArgs {
     domain: domain::Name,
 
     #[serde(flatten)]
-    flat_domain_settings: service::util::FlatDomainSettings,
+    url_encoded_domain_settings: util::UrlEncodedDomainSettings,
 }
 
 #[derive(Deserialize)]
@@ -73,7 +74,7 @@ struct DomainSyncArgs {
     passphrase: String,
 
     #[serde(flatten)]
-    flat_domain_settings: service::util::FlatDomainSettings,
+    url_encoded_domain_settings: util::UrlEncodedDomainSettings,
 }
 
 impl<'svc> Service {
@@ -267,7 +268,7 @@ impl<'svc> Service {
         #[derive(Serialize)]
         struct Data<'a> {
             domain: domain::Name,
-            flat_domain_settings: service::util::FlatDomainSettings,
+            url_encoded_domain_settings: util::UrlEncodedDomainSettings,
             dns_records: &'a [service::ConfigDNSRecord],
             challenge_token: String,
 
@@ -275,7 +276,7 @@ impl<'svc> Service {
             dns_records_have_cname: bool,
         }
 
-        let settings: domain::Settings = match args.flat_domain_settings.try_into() {
+        let settings: domain::Settings = match args.url_encoded_domain_settings.try_into() {
             Ok(settings) => settings,
             Err(e) => {
                 return self
@@ -298,7 +299,7 @@ impl<'svc> Service {
             _ => false,
         });
 
-        let flat_domain_settings = match settings.try_into() {
+        let url_encoded_domain_settings = match settings.try_into() {
             Ok(s) => s,
             Err(e) => {
                 return self
@@ -310,7 +311,7 @@ impl<'svc> Service {
             "/domain_init.html",
             Data {
                 domain: args.domain,
-                flat_domain_settings,
+                url_encoded_domain_settings,
                 dns_records: &self.config.dns_records,
                 challenge_token: settings_hash,
 
@@ -325,7 +326,7 @@ impl<'svc> Service {
             return self.render_error_page(401, "Incorrect passphrase");
         }
 
-        let settings: domain::Settings = match args.flat_domain_settings.try_into() {
+        let settings: domain::Settings = match args.url_encoded_domain_settings.try_into() {
             Ok(settings) => settings,
             Err(e) => {
                 return self

src/service/http/config.rs

@@ -7,10 +7,10 @@ use serde_with::{serde_as, TryFromInto};
 use std::{collections, net, str::FromStr};
 
 fn default_http_addr() -> net::SocketAddr {
-    net::SocketAddr::from_str("[::]:3030").unwrap()
+    net::SocketAddr::from_str("[::]:3080").unwrap()
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Clone)]
 pub enum ConfigFormMethod {
     GET,
     POST,
@@ -60,7 +60,7 @@ impl TryFrom<String> for ConfigProxiedDomainUrl {
 
     fn try_from(url: String) -> Result<Self, Self::Error> {
         let parsed = http::Uri::from_str(url.as_str())
-            .or_unexpected_while("parsing proxy url {proxy_url}")?;
+            .map_unexpected_while(|| format!("parsing proxy url {url}"))?;
 
         let scheme = parsed
             .scheme()
@@ -76,7 +76,7 @@ impl TryFrom<String> for ConfigProxiedDomainUrl {
     }
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Clone)]
 pub struct ConfigProxiedDomainRequestHeader {
     pub name: String,
     pub value: String,
@@ -129,7 +129,7 @@ impl TryFrom<Vec<ConfigProxiedDomainRequestHeader>> for ConfigProxiedDomainReque
 }
 
 #[serde_as]
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Clone)]
 pub struct ConfigProxiedDomain {
     #[serde_as(as = "TryFromInto<String>")]
     pub url: ConfigProxiedDomainUrl,
@@ -139,7 +139,7 @@ pub struct ConfigProxiedDomain {
     pub request_headers: ConfigProxiedDomainRequestHeaders,
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Clone)]
 pub struct Config {
     #[serde(default = "default_http_addr")]
     pub http_addr: net::SocketAddr,

src/service/http/util.rs

@@ -7,7 +7,7 @@ use crate::{domain, error::unexpected, origin};
 
 #[serde_as]
 #[derive(Serialize, Deserialize, Default, Debug)]
-pub struct FlatDomainSettings {
+pub struct UrlEncodedDomainSettings {
     domain_setting_origin_descr_kind: String,
 
     domain_setting_origin_descr_git_url: Option<String>,
@@ -20,10 +20,10 @@ pub struct FlatDomainSettings {
     domain_setting_add_path_prefix: Option<String>,
 }
 
-impl TryFrom<FlatDomainSettings> for domain::Settings {
+impl TryFrom<UrlEncodedDomainSettings> for domain::Settings {
     type Error = String;
 
-    fn try_from(v: FlatDomainSettings) -> Result<Self, Self::Error> {
+    fn try_from(v: UrlEncodedDomainSettings) -> Result<Self, Self::Error> {
         let origin_descr = match v.domain_setting_origin_descr_kind.as_str() {
             "git" => Ok(origin::Descr::Git {
                 url: v
@@ -44,11 +44,11 @@ impl TryFrom<FlatDomainSettings> for domain::Settings {
     }
 }
 
-impl TryFrom<domain::Settings> for FlatDomainSettings {
+impl TryFrom<domain::Settings> for UrlEncodedDomainSettings {
     type Error = unexpected::Error;
 
     fn try_from(v: domain::Settings) -> Result<Self, Self::Error> {
-        let mut res = FlatDomainSettings::default();
+        let mut res = UrlEncodedDomainSettings::default();
 
         match v.origin_descr {
             origin::Descr::Git { url, branch_name } => {

src/task_stack.rs

@@ -0,0 +1,66 @@
+use crate::util::BoxFuture;
+use std::error;
+use tokio_util::sync::CancellationToken;
+
+pub struct TaskStack<E>
+where
+    E: error::Error + Send + 'static,
+{
+    wait_group: Vec<BoxFuture<'static, Result<(), E>>>,
+}
+
+impl<E> TaskStack<E>
+where
+    E: error::Error + Send + 'static,
+{
+    pub fn new() -> TaskStack<E> {
+        TaskStack {
+            wait_group: Vec::new(),
+        }
+    }
+
+    /// push adds the given Future to the stack, to be executed once stop is called.
+    pub fn push<Fut>(&mut self, f: Fut)
+    where
+        Fut: futures::Future<Output = Result<(), E>> + Send + 'static,
+    {
+        self.wait_group.push(Box::pin(f))
+    }
+
+    /// push_spawn will spawn the given closure in a tokio task. Once the CancellationToken is
+    /// cancelled the closure is expected to return.
+    pub fn push_spawn<F, Fut>(&mut self, mut f: F)
+    where
+        Fut: futures::Future<Output = Result<(), E>> + Send + 'static,
+        F: FnMut(CancellationToken) -> Fut,
+    {
+        let canceller = CancellationToken::new();
+        let handle = tokio::spawn(f(canceller.clone()));
+        self.push(async move {
+            canceller.cancel();
+            handle.await.expect("failed to join task")
+        });
+    }
+
+    /// stop will process all operations which have been pushed onto the stack in the reverse order
+    /// they were pushed.
+    pub async fn stop(mut self) -> Result<(), E> {
+        // reverse wait_group in place, so we stop the most recently added first. Since this method
+        // consumes self this is fine.
+        self.wait_group.reverse();
+
+        for fut in self.wait_group {
+            fut.await?;
+        }
+        Ok(())
+    }
+}
+
+impl<E> Default for TaskStack<E>
+where
+    E: error::Error + Send + 'static,
+{
+    fn default() -> Self {
+        Self::new()
+    }
+}

src/util.rs

@@ -1,6 +1,4 @@
-use std::{error, fs, io, path, pin};
+use std::{fs, io, path, pin};
-
-use tokio_util::sync::CancellationToken;
 
 pub fn open_file(path: &path::Path) -> io::Result<Option<fs::File>> {
     match fs::File::open(path) {
@@ -15,66 +13,3 @@ pub fn open_file(path: &path::Path) -> io::Result<Option<fs::File>> {
 pub type BoxByteStream = futures::stream::BoxStream<'static, io::Result<Vec<u8>>>;
 
 pub type BoxFuture<'a, O> = pin::Pin<Box<dyn futures::Future<Output = O> + Send + 'a>>;
-
-pub struct TaskStack<E>
-where
-    E: error::Error + Send + 'static,
-{
-    wait_group: Vec<BoxFuture<'static, Result<(), E>>>,
-}
-
-impl<E> TaskStack<E>
-where
-    E: error::Error + Send + 'static,
-{
-    pub fn new() -> TaskStack<E> {
-        TaskStack {
-            wait_group: Vec::new(),
-        }
-    }
-
-    /// push adds the given Future to the stack, to be executed once stop is called.
-    pub fn push<Fut>(&mut self, f: Fut)
-    where
-        Fut: futures::Future<Output = Result<(), E>> + Send + 'static,
-    {
-        self.wait_group.push(Box::pin(f))
-    }
-
-    /// push_spawn will spawn the given closure in a tokio task. Once the CancellationToken is
-    /// cancelled the closure is expected to return.
-    pub fn push_spawn<F, Fut>(&mut self, mut f: F)
-    where
-        Fut: futures::Future<Output = Result<(), E>> + Send + 'static,
-        F: FnMut(CancellationToken) -> Fut,
-    {
-        let canceller = CancellationToken::new();
-        let handle = tokio::spawn(f(canceller.clone()));
-        self.push(async move {
-            canceller.cancel();
-            handle.await.expect("failed to join task")
-        });
-    }
-
-    /// stop will process all operations which have been pushed onto the stack in the reverse order
-    /// they were pushed.
-    pub async fn stop(mut self) -> Result<(), E> {
-        // reverse wait_group in place, so we stop the most recently added first. Since this method
-        // consumes self this is fine.
-        self.wait_group.reverse();
-
-        for fut in self.wait_group {
-            fut.await?;
-        }
-        Ok(())
-    }
-}
-
-impl<E> Default for TaskStack<E>
-where
-    E: error::Error + Send + 'static,
-{
-    fn default() -> Self {
-        Self::new()
-    }
-}