domani/src/service/gemini.rs

mod config;
mod proxy;

pub use config::*;

use crate::error::unexpected::{self, Mappable};
use crate::{domain, service, task_stack, util};

use core::time::Duration;
use std::sync;
use tokio_util::sync::CancellationToken;

pub struct Service {
    domain_manager: sync::Arc<dyn domain::manager::Manager>,
    cert_resolver: sync::Arc<dyn rustls::server::ResolvesServerCert>,
    config: Config,
}

#[derive(thiserror::Error, Debug)]
enum HandleConnError {
    #[error("client error: {0}")]
    ClientError(String),

    #[error(transparent)]
    Unexpected(#[from] unexpected::Error),
}

impl Service {
    pub fn new<CertResolver>(
        task_stack: &mut task_stack::TaskStack<unexpected::Error>,
        domain_manager: sync::Arc<dyn domain::manager::Manager>,
        cert_resolver: CertResolver,
        config: Config,
    ) -> sync::Arc<Service>
    where
        CertResolver: rustls::server::ResolvesServerCert + 'static,
    {
        let service = sync::Arc::new(Service {
            domain_manager,
            cert_resolver: sync::Arc::from(cert_resolver),
            config,
        });
        task_stack.push_spawn(|canceller| listen(service.clone(), canceller));
        service
    }

    async fn respond_conn<W>(
        &self,
        w: W,
        code: &str,
        meta: &str,
        body: Option<util::BoxByteStream>,
    ) -> unexpected::Result<()>
    where
        W: tokio::io::AsyncWrite + Unpin,
    {
        use tokio::io::{copy, AsyncWriteExt, BufWriter};
        let mut w = BufWriter::new(w);

        w.write_all(code.as_bytes()).await.or_unexpected()?;
        w.write_all(" ".as_bytes()).await.or_unexpected()?;
        w.write_all(meta.as_bytes()).await.or_unexpected()?;
        w.write_all("\r\n".as_bytes()).await.or_unexpected()?;

        if let Some(body) = body {
            let mut body = body.into_async_read();
            copy(&mut body, &mut w).await.or_unexpected()?;
        }

        w.flush().await.or_unexpected()?;

        Ok(())
    }

    async fn serve_conn<IO>(
        &self,
        settings: &domain::Settings,
        conn: IO,
    ) -> Result<(), HandleConnError>
    where
        IO: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
    {
        use tokio::io::*;

        let (r, w) = split(conn);
        let mut r = BufReader::new(r);

        let mut req = String::with_capacity(64);
        r.read_line(&mut req)
            .await
            .map_err(|e| HandleConnError::ClientError(format!("failed to read request: {e}")))?;

        let req = gemini::request::parse::request(req.as_bytes())
            .map(|(_, req)| req)
            .map_err(|e| HandleConnError::ClientError(format!("failed to parse request: {e}")))?
            .into_gemini_request()
            .map_err(|e| HandleConnError::ClientError(format!("failed to parse request: {e}")))?;

        let path = service::append_index_to_path(req.path(), "index.gmi");

        use domain::manager::GetFileError;
        let f = match self.domain_manager.get_file(settings, &path).await {
            Ok(f) => f,
            Err(GetFileError::FileNotFound) => {
                return Ok(self.respond_conn(w, "51", "File not found", None).await?)
            }
            Err(GetFileError::Unavailable) => {
                return Ok(self
                    .respond_conn(w, "43", "Content unavailable", None)
                    .await?)
            }
            Err(GetFileError::DescrNotSynced) => {
                return Err(unexpected::Error::from(
                    format!(
                        "Backend for {:?} has not yet been synced",
                        settings.origin_descr
                    )
                    .as_str(),
                )
                .into())
            }
            Err(GetFileError::PathIsDirectory) => {
                // redirect so that the path has '/' appended to it, which will cause the server to
                // check index.gmi within the path on the new page load.
                let mut path = path.into_owned();
                path.push('/');
                return Ok(self.respond_conn(w, "30", path.as_str(), None).await?);
            }
            Err(GetFileError::Unexpected(e)) => return Err(e.into()),
        };

        let content_type = service::guess_mime(&path);

        Ok(self
            .respond_conn(w, "20", content_type.as_str(), Some(f))
            .await?)
    }

    async fn proxy_conn<IO>(&self, proxy_addr: &str, mut conn: IO) -> unexpected::Result<()>
    where
        IO: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
    {
        let mut proxy_conn = tokio::net::TcpStream::connect(proxy_addr)
            .await
            .map_unexpected_while(|| format!("failed to connect to proxy {proxy_addr}"))?;

        _ = 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 teed_conn = {
            let (r, w) = tokio::io::split(conn);
            let r = proxy::AsyncTeeRead::with_capacity(r, 1024);
            proxy::AsyncReadWrite::new(r, w)
        };

        let acceptor =
            tokio_rustls::LazyConfigAcceptor::new(rustls::server::Acceptor::default(), teed_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}"
                    ))
                })?;

                use domain::manager::{GetSettingsError, GetSettingsResult};

                let get_settings_res = self.domain_manager.get_settings(&domain);

                // If the domain should be proxied, then proxy it. This case gets checked before
                // any of the others because the others require terminating the TLS stream to be
                // handled.
                if let Ok(GetSettingsResult::Proxied(ref config)) = get_settings_res {
                    if let Some(ref gemini_url) = config.gemini_url {
                        let prefixed_conn = proxy::teed_io_to_prefixed(start.into_inner());
                        self.proxy_conn(gemini_url.addr.as_str(), prefixed_conn)
                            .await?;
                        return Ok(());
                    }
                }

                // Terminate TLS stream
                let conn = start.into_stream(tls_config).await.or_unexpected()?;

                let settings = match get_settings_res {
                    Ok(GetSettingsResult::Stored(settings)) => settings,
                    Ok(GetSettingsResult::Builtin(config)) => config.settings,
                    Ok(GetSettingsResult::Proxied(_)) => {
                        panic!("proxied case already handled")
                    }
                    Ok(GetSettingsResult::Interface)
                    | Ok(GetSettingsResult::External(_))
                    | Err(GetSettingsError::NotFound) => {
                        return Ok(self
                            .respond_conn(conn, "51", "File not found", None)
                            .await?)
                    }
                    Err(GetSettingsError::Unexpected(e)) => return Err(e.into()),
                };

                self.serve_conn(&settings, conn).await
            }
            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 addr = &service
        .config
        .gemini_addr
        .expect("listen called with gemini_addr not set");

    let tls_config = sync::Arc::new(
        rustls::server::ServerConfig::builder()
            .with_safe_defaults()
            .with_no_client_auth()
            .with_cert_resolver(service.cert_resolver.clone()),
    );

    log::info!("Listening on gemini://{}", addr);

    let listener = tokio::net::TcpListener::bind(addr)
        .await
        .expect("failed to bind tcp socket");

    loop {
        let (conn, addr) = tokio::select! {
            res = listener.accept() => {
                if let Err(err) = res {
                    log::error!("Failed to accept new gemini request on {addr}: {err}");
                    tokio::time::sleep(Duration::from_secs(2)).await;
                    continue;
                }
                res.unwrap()
            },
            _ = 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}")
                }
            }
        });
    }
}