mumble-web2/proxy/src/main.rs

use color_eyre::eyre::{anyhow, bail, Context, Result};
use mumble_web2_common::{ping_server, ProxyOverrides, ServerStatus};
use salvo::conn::rustls::{Keycert, RustlsConfig};
use salvo::cors::{AllowOrigin, Cors};
use salvo::logging::Logger;
use salvo::prelude::*;
use salvo::proto::quic::BidiStream;
use serde::{Deserialize, Serialize};
use std::net::{SocketAddr, ToSocketAddrs};
use std::path::PathBuf;
use std::sync::Arc;
use tokio::fs;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::pin;
use tokio_rustls::rustls::client::danger::{HandshakeSignatureValid, ServerCertVerifier};
use tokio_rustls::rustls::pki_types::{CertificateDer, ServerName, UnixTime};
use tokio_rustls::rustls::{ClientConfig, DigitallySignedStruct};
use tokio_rustls::{rustls, TlsConnector};
use tracing::info;
use tracing::info_span;
use tracing::Instrument;
use tracing::{error, instrument};
use tracing_subscriber::filter::LevelFilter;
use tracing_subscriber::EnvFilter;
use url::Url;

fn default_cert_alt_names() -> Vec<String> {
    vec!["localhost".into()]
}

#[derive(Debug, Deserialize, Serialize)]
struct Config {
    proxy_url: Option<Url>,
    https_listen_address: SocketAddr,
    http_listen_address: Option<SocketAddr>,
    cert_path: Option<PathBuf>,
    key_path: Option<PathBuf>,
    #[serde(default = "default_cert_alt_names")]
    cert_alt_names: Vec<String>,
    mumble_server_url: String,
    mumble_server_address: Option<SocketAddr>,
    gui_path: Option<PathBuf>,
}

fn init_config() -> Result<Config> {
    let mut config: Config = toml::from_str(
        &std::fs::read_to_string("./config.toml")
            .context("reading config.toml (try making a copy of config.toml.example)")?,
    )?;
    let mumble_server_addr = config
        .mumble_server_url
        .to_socket_addrs()
        .context(format!(
            "parsing mumble_server_url={}",
            config.mumble_server_url
        ))?
        .next()
        .ok_or(anyhow!(
            "no socket addrs in mumble_server_url={}",
            config.mumble_server_url
        ))?;
    config.mumble_server_address = Some(mumble_server_addr);
    Ok(config)
}

#[tokio::main]
async fn main() -> Result<()> {
    init_logging();
    let server_config = Arc::new(init_config()?);
    info!("config:\n{}", toml::to_string_pretty(&*server_config)?);

    rustls::crypto::aws_lc_rs::default_provider()
        .install_default()
        .map_err(|e| anyhow!("could not install crypto provider {e:?}"))?;

    let mut overrides = ProxyOverrides {
        proxy_url: match &server_config.proxy_url {
            Some(url) => Some(url.to_string()),
            None => None,
        },
        cert_hash: None,
        any_server: false,
    };

    let (cert, key) = match (&server_config.cert_path, &server_config.key_path) {
        (None, None) => {
            info!("generating self-signed cert");

            // FIXME: redo every <14 days
            let mut dname = rcgen::DistinguishedName::new();
            dname.push(rcgen::DnType::CommonName, "mumble-web self-signed");
            let key_pair = rcgen::KeyPair::generate_for(&rcgen::PKCS_ECDSA_P256_SHA256)?;
            let mut cert_params =
                rcgen::CertificateParams::new(server_config.cert_alt_names.clone())?;
            cert_params.distinguished_name = dname;
            cert_params.not_before = time::OffsetDateTime::now_utc();
            cert_params.not_after = cert_params.not_before + time::Duration::days(12);
            let cert = cert_params.self_signed(&key_pair)?;

            let hash = hmac_sha256::Hash::hash(cert.der().as_ref());
            overrides.cert_hash = Some(hash.into());

            (cert.pem().into(), key_pair.serialize_pem().into())
        }
        (Some(cert_path), Some(key_path)) => {
            // Read server certs
            let cert = fs::read(cert_path)
                .await
                .context(format!("reading cert {}", cert_path.display()))?;
            let key = fs::read(key_path)
                .await
                .context(format!("reading key {}", key_path.display()))?;
            (cert, key)
        }
        _ => {
            bail!("please supply both cert_path and key_path (or neither to generate a self-signed cert)")
        }
    };
    let rustls_config = RustlsConfig::new(Keycert::new().cert(cert.as_slice()).key(key.as_slice()));

    info!("proxy overrides:\n{}", toml::to_string_pretty(&overrides)?);

    let config_craft = ConfigCraft {
        server_config: server_config.clone(),
        overrides,
    };

    let status_craft = StatusCraft {
        mumble_server_address: server_config.mumble_server_address.unwrap().clone(),
    };

    // Server routing
    let mut router = Router::new()
        .push(Router::with_path("/proxy").goal(config_craft.connect_proxy()))
        .push(Router::with_path("/overrides").get(config_craft.get_overrides()))
        .push(Router::with_path("/status").get(status_craft.get_status()))
        .hoop(Logger::new());
    if let Some(gui_path) = server_config.gui_path.clone() {
        router =
            router.push(Router::with_path("/").get(StaticFile::new(gui_path.join("index.html"))));
        router = router.push(Router::with_path("/<*+rest>").get(StaticDir::new(gui_path)));
    }

    let cors = Cors::new().allow_origin(AllowOrigin::any()).into_handler();

    let service = Service::new(router).hoop(cors);

    // Create http listeners
    let http_listener = server_config.http_listen_address.map(TcpListener::new);
    let https_listener =
        TcpListener::new(server_config.https_listen_address).rustls(rustls_config.clone());
    let http3_listener = QuinnListener::new(rustls_config, server_config.https_listen_address);

    // Start server
    match (http_listener, https_listener, http3_listener) {
        (Some(a), b, c) => {
            let accepter = a.join(b).join(c).bind().await;
            Server::new(accepter).serve(service).await;
        }
        (None, b, c) => {
            let accepter = b.join(c).bind().await;
            Server::new(accepter).serve(service).await;
        }
    }

    Ok(())
}

#[derive(Clone)]
pub struct StatusCraft {
    mumble_server_address: SocketAddr,
}

#[craft]
impl StatusCraft {
    #[craft(handler)]
    async fn get_status(&self) -> Json<ServerStatus> {
        let addr = self.mumble_server_address;
        match ping_server(&addr.ip().to_string(), addr.port()).await {
            Ok(status) => Json(status),
            Err(e) => {
                error!("ping failed: {e:#}");
                Json(ServerStatus::default())
            }
        }
    }
}

#[derive(Clone)]
pub struct ConfigCraft {
    server_config: Arc<Config>,
    overrides: ProxyOverrides,
}

#[craft]
impl ConfigCraft {
    #[craft(handler)]
    async fn get_overrides(&self) -> Json<ProxyOverrides> {
        Json(self.overrides.clone())
    }

    #[craft(handler)]
    async fn connect_proxy(&self, req: &mut Request, res: &mut Response) {
        info!("received proxy request");
        let mumble_server_address = self.server_config.mumble_server_address.unwrap();
        let wt = match req.web_transport_mut().await {
            Ok(wt) => wt,
            Err(err) => {
                res.status_code(StatusCode::BAD_REQUEST);
                res.render(format!("error with webtransport: {err:?}"));
                return;
            }
        };
        info!("got webtransport for connection");

        use salvo::webtransport::server::AcceptedBi;
        let (id, bi) = match wt.accept_bi().await {
            Ok(Some(AcceptedBi::BidiStream(id, bi))) => (id, bi),
            Ok(Some(AcceptedBi::Request(req, _))) => {
                res.status_code(StatusCode::BAD_REQUEST);
                res.render(format!(
                    "expected webtransport stream but got request {req:?}"
                ));
                return;
            }
            Ok(None) => {
                res.status_code(StatusCode::BAD_REQUEST);
                res.render(format!("no bidirectional connection requested"));
                return;
            }
            Err(err) => {
                res.status_code(StatusCode::INTERNAL_SERVER_ERROR);
                res.render(format!("error with bidirectional connection: {err:?}"));
                return;
            }
        };

        let (outgoing, incoming) = bi.split();
        let res = tokio::spawn(async move {
            if let Err(error) = connect_proxy_impl(mumble_server_address, incoming, outgoing).await
            {
                error!("error connecting proxy {error:?}")
            }
        })
        .await;
        if let Err(err) = res {
            error!("crash in connected proxy {err:?}");
        }
    }
}

#[instrument(skip(incoming, outgoing))]
async fn connect_proxy_impl(
    mumble_server_address: SocketAddr,
    incoming: impl AsyncRead + Send + Sync + 'static,
    outgoing: impl AsyncWrite + Send + Sync + 'static,
) -> Result<()> {
    info!("connecting to Mumble server...");

    let config = ClientConfig::builder()
        .dangerous()
        .with_custom_certificate_verifier(Arc::new(NoCertificateVerification))
        .with_no_client_auth();

    let connector = TlsConnector::from(Arc::new(config));

    let server_tcp = TcpStream::connect(mumble_server_address).await?;
    let server_stream = connector
        .connect("example.com".try_into()?, server_tcp)
        .await?;
    let (read_server, write_server) = tokio::io::split(server_stream);

    info!("connected to Mumble server");

    // Handle transmitting data between the WebTransport client and Mumble TCP Server
    // When one direction completes/fails, the other is dropped and its streams are closed
    tokio::select! {
        res = pass_bytes_loop(incoming, write_server)
            .instrument(info_span!("Handler", "Client to server")) => res?,
        res = pass_bytes_loop(read_server, outgoing)
            .instrument(info_span!("Handler", "Server to client")) => res?,
    };
    Ok(())
}

#[derive(Debug)]
struct NoCertificateVerification;

impl ServerCertVerifier for NoCertificateVerification {
    fn verify_server_cert(
        &self,
        _end_entity: &CertificateDer<'_>,
        _intermediates: &[CertificateDer<'_>],
        _server_name: &ServerName<'_>,
        _ocsp: &[u8],
        _now: UnixTime,
    ) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
        Ok(rustls::client::danger::ServerCertVerified::assertion())
    }

    fn verify_tls12_signature(
        &self,
        _message: &[u8],
        _cert: &CertificateDer<'_>,
        _dss: &DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, rustls::Error> {
        Ok(HandshakeSignatureValid::assertion())
    }

    fn verify_tls13_signature(
        &self,
        _message: &[u8],
        _cert: &CertificateDer<'_>,
        _dss: &DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, rustls::Error> {
        Ok(HandshakeSignatureValid::assertion())
    }

    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
        vec![
            rustls::SignatureScheme::RSA_PKCS1_SHA1,
            rustls::SignatureScheme::ECDSA_SHA1_Legacy,
            rustls::SignatureScheme::RSA_PKCS1_SHA256,
            rustls::SignatureScheme::ECDSA_NISTP256_SHA256,
            rustls::SignatureScheme::RSA_PKCS1_SHA384,
            rustls::SignatureScheme::ECDSA_NISTP384_SHA384,
            rustls::SignatureScheme::RSA_PKCS1_SHA512,
            rustls::SignatureScheme::ECDSA_NISTP521_SHA512,
            rustls::SignatureScheme::RSA_PSS_SHA256,
            rustls::SignatureScheme::RSA_PSS_SHA384,
            rustls::SignatureScheme::RSA_PSS_SHA512,
            rustls::SignatureScheme::ED25519,
            rustls::SignatureScheme::ED448,
        ]
    }
}

async fn pass_bytes_loop(
    client_stream: impl AsyncRead + Sync + Send + 'static,
    server_stream: impl AsyncWrite + Send + Sync + 'static,
) -> Result<()> {
    let mut buffer = vec![0; 65536].into_boxed_slice();
    pin!(client_stream);
    pin!(server_stream);
    loop {
        let bytes_read = client_stream.read(&mut buffer).await?;
        if bytes_read == 0 {
            break Ok(());
        }

        server_stream.write_all(&buffer[..bytes_read]).await?;
        server_stream.flush().await?;
    }
}

fn init_logging() {
    let env_filter = EnvFilter::builder()
        .with_default_directive(LevelFilter::DEBUG.into())
        .from_env_lossy();

    tracing_subscriber::fmt()
        .with_target(true)
        .with_level(true)
        .with_env_filter(env_filter)
        .init();
}