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
92
93
94
95
96
97
98
99

use std::io::Error as IoError;

use log::debug;
use log::info;
use log::error;
use futures_util::stream::StreamExt;
use futures_util::io::AsyncReadExt;
use futures_lite::io::copy;
use fluvio_future::net::TcpListener;
use fluvio_future::net::TcpStream;
use fluvio_future::tls::TlsAcceptor;
use fluvio_future::tls::DefaultServerTlsStream;
use fluvio_future::task::spawn;


/// start TLS proxy at addr to target
pub async fn start(addr: &str, acceptor: TlsAcceptor, target: String) -> Result<(),IoError> {

    let listener = TcpListener::bind(addr).await?;
    info!("proxy started at: {}",addr);
    let mut incoming = listener.incoming();
    while let Some(stream) = incoming.next().await {
        
        debug!("server: got connection from client");
        if let Ok(tcp_stream) = stream {

            debug!(
                "new connection from {}",
                tcp_stream
                    .peer_addr()
                    .map(|addr| addr.to_string())
                    .unwrap_or("".to_owned())
            );
            
            let handshake = acceptor.accept(tcp_stream);

            match handshake.await {
                Ok(inner_stream) => {
                    debug!("handshake success: starting");
                    process_stream(inner_stream,target.clone()).await;
                },
                Err(err) => error!("error handshaking: {}",err)
            }
        }
    }

    info!("server terminated");
    Ok(())
}


async fn process_stream(stream: DefaultServerTlsStream, target: String) {

    // connect to other end
    if let Err(err) = proxy(stream,target).await {
        error!("error processing tls: {}",err);
    }
    
}

async fn proxy(tls_stream: DefaultServerTlsStream, target: String) -> Result<(),IoError> {

    debug!("trying to connect to target at: {}", target);
    let tcp_stream = TcpStream::connect(&target).await?;

    debug!("connect to target: {}",target);
    let (mut target_stream,mut target_sink) = tcp_stream.split();
    let (mut from_tls_stream,mut from_tls_sink) = tls_stream.split();
    
    let source_to_target_ft = async move {

        match copy(&mut from_tls_stream,&mut target_sink).await {
            Ok(_) => {
                debug!("done copying from source to target");
            },
            Err(err) => {
                error!("error copying: {}",err);
            }
        }
    };

    let target_to_source = async move {
        match copy(&mut target_stream,&mut from_tls_sink).await {
            Ok(_) => {
                debug!("done copying from source to target");
            },
            Err(err) => {
                error!("error copying: {}",err);
            }
        }
    };

    spawn(source_to_target_ft);
    spawn(target_to_source);

    Ok(())
}