apprtc 0.1.0

AppRTC P2P/SFU Signaling Server in Rust
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

use anyhow::{Context, Result, bail};
use axum::serve::Listener;
use rustls::ServerConfig;
use rustls::pki_types::CertificateDer;
use std::io::BufReader;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc;
use tokio_rustls::TlsAcceptor;
use tokio_rustls::server::TlsStream;

pub fn config(certificate: &str, private_key: &str) -> Result<Arc<ServerConfig>> {
    let (certificate, private_key) = if certificate.is_empty() && private_key.is_empty() {
        (
            include_bytes!("../cert/cert.pem").to_vec(),
            include_bytes!("../cert/key.pem").to_vec(),
        )
    } else if !certificate.is_empty() && !private_key.is_empty() {
        (
            std::fs::read(certificate)
                .with_context(|| format!("failed to read certificate {certificate}"))?,
            std::fs::read(private_key)
                .with_context(|| format!("failed to read private key {private_key}"))?,
        )
    } else {
        bail!("--certificate and --private-key must be supplied together");
    };
    let certificates: Vec<CertificateDer<'static>> =
        rustls_pemfile::certs(&mut BufReader::new(&certificate[..])).collect::<Result<_, _>>()?;
    let private_key = rustls_pemfile::private_key(&mut BufReader::new(&private_key[..]))?
        .ok_or_else(|| anyhow::anyhow!("no private key found in PEM input"))?;
    let _ = rustls::crypto::ring::default_provider().install_default();
    Ok(Arc::new(
        ServerConfig::builder()
            .with_no_client_auth()
            .with_single_cert(certificates, private_key)?,
    ))
}

pub struct TlsListener {
    address: SocketAddr,
    streams: mpsc::Receiver<(TlsStream<TcpStream>, SocketAddr)>,
}

impl TlsListener {
    pub fn new(listener: TcpListener, config: Arc<ServerConfig>) -> Self {
        let address = listener.local_addr().expect("TLS listener local address");
        let acceptor = TlsAcceptor::from(config);
        let (sender, streams) = mpsc::channel(1024);
        tokio::spawn(async move {
            loop {
                let (stream, peer) = match listener.accept().await {
                    Ok(value) => value,
                    Err(error) => {
                        log::error!("TCP accept failed: {error}");
                        continue;
                    }
                };
                let acceptor = acceptor.clone();
                let sender = sender.clone();
                tokio::spawn(async move {
                    match tokio::time::timeout(Duration::from_secs(10), acceptor.accept(stream))
                        .await
                    {
                        Ok(Ok(stream)) => {
                            let _ = sender.send((stream, peer)).await;
                        }
                        Ok(Err(error)) => log::warn!("TLS handshake from {peer} failed: {error}"),
                        Err(_) => log::warn!("TLS handshake from {peer} timed out"),
                    }
                });
            }
        });
        Self { address, streams }
    }
}

impl Listener for TlsListener {
    type Io = TlsStream<TcpStream>;
    type Addr = SocketAddr;

    async fn accept(&mut self) -> (Self::Io, Self::Addr) {
        self.streams.recv().await.expect("TLS accept loop stopped")
    }

    fn local_addr(&self) -> std::io::Result<Self::Addr> {
        Ok(self.address)
    }
}