rcs_backend 0.2.1

otafablab rust code server backend
Documentation 
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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

// Much of the boilerplate for the proxy is from hyper's examples
// https://github.com/hyperium/hyper/blob/master/examples/http_proxy.rs
// https://github.com/hyperium/hyper/blob/master/examples/upgrades.rs

use std::io;

use crate::config::Config;
use crate::docker::start_rcs;
use crate::utils::{decode_auth_cookie, HttpClient};
use http::header::HOST;
use http::{request, response, Uri};
use hyper::header::HeaderValue;
use hyper::upgrade::Upgraded;
use hyper::{Body, Request, Response, StatusCode};
use tokio::io::DuplexStream;

/// Proxy incoming request to another server at `authority` and return
/// the response back
///
/// # Panics
///
/// Proxying may panic if:
///
/// - [`decode_access_token`] fails, for example if the cookie is missing or invalid
///
/// - [`start_rcs`] panics
///
/// # Errors
///
/// A [`hyper::Error`] is returned in the case that connecting to
/// downstream rust-code-server hosts fails
pub async fn proxy(
    client: HttpClient,
    mut request: Request<Body>,
    config: &'static Config,
) -> Result<Response<Body>, hyper::Error> {
    println!(" <-- {} {}", request.method(), request.uri().path());

    // TODO parameterize authority to proxy
    let token = decode_auth_cookie(request.headers(), config).unwrap();
    // TODO start_rcs outside the proxy?
    let authority = start_rcs(&token.email, config).await;
    println!("proxying to {}", &authority);

    *request.uri_mut() = Uri::builder()
        .scheme("http")
        .authority(authority.clone())
        .path_and_query(
            request
                .uri()
                .path_and_query()
                .expect("path and query to exist")
                .as_str(),
        )
        .build()
        .unwrap();
    (*request.headers_mut()).insert(HOST, HeaderValue::from_str(authority.host()).unwrap());

    // handle upgrade requests like websocket
    if hyper_tungstenite::is_upgrade_request(&request) {
        // TODO
        // proxy_websocket(req, authority)

        let (parts, body) = request.into_parts();
        let mut request_copy = request::Builder::new()
            .method(parts.method.clone())
            .version(parts.version)
            .uri(parts.uri.clone())
            .body(Body::empty())
            .unwrap();

        *request_copy.headers_mut() = parts.headers.clone();
        let request = Request::from_parts(parts, body);

        // proxy the same request to downstream
        let response = client.request(request_copy).await?;

        let (parts, body) = response.into_parts();
        let mut response_copy = response::Builder::new()
            .status(parts.status)
            .version(parts.version)
            .body(Body::empty())
            .unwrap();
        *response_copy.headers_mut() = parts.headers.clone();
        let response = Response::from_parts(parts, body);

        assert!(
            response.status() == StatusCode::SWITCHING_PROTOCOLS,
            "downstream didn't upgrade: {}",
            response.status()
        );

        let (left, right) = tokio::io::duplex(64);

        tokio::spawn(async move {
            // backend (client) <=> downstream (server) upgrade
            match hyper::upgrade::on(response).await {
                Ok(upgraded) => {
                    // tunnel between upgraded connection from downstream and duplex
                    if let Err(e) = tunnel(upgraded, left).await {
                        eprintln!("downstream tunnel io error: {}", e);
                    };
                }
                Err(e) => eprintln!("upgrade error: {}", e),
            }
        });

        tokio::spawn(async move {
            // upstream (client) <=> backend (server) upgrade
            match hyper::upgrade::on(request).await {
                Ok(upgraded) => {
                    // tunnel between upgraded connection from upstream and duplex
                    if let Err(e) = tunnel(upgraded, right).await {
                        eprintln!("upstream tunnel io error: {}", e);
                    };
                }
                Err(e) => eprintln!("upgrade error: {}", e),
            }
        });

        // return the upgrade response from downstream
        Ok(response_copy)
    } else {
        // handle regular http requests
        client.request(request).await
    }
}

/// Create a TCP connection to host:port, build a tunnel between the
/// connection and the upgraded connection
async fn tunnel(mut upgraded: Upgraded, mut duplex: DuplexStream) -> io::Result<()> {
    println!("tunnel open");
    // Proxying data
    let (from_client, from_server) =
        tokio::io::copy_bidirectional(&mut upgraded, &mut duplex).await?;

    // Print message when done
    println!(
        "tunnel closing: wrote {} bytes and received {} bytes",
        from_client, from_server
    );

    Ok(())
}