reactor
/
ptth
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
ptth/crates/ptth_quic/src/executable_relay_server.rs

490 lines
11 KiB
Rust
Raw Normal View History

:recycle: refactor: break out most of the relay server into a module I want an end-to-end test that runs all 5 nodes in 1 process, like PTTH itself has.
2022-10-09 16:43:50 +00:00
use hyper::{
Body,
Request,
Response,
Server,
service::{
make_service_fn,
service_fn,
},
StatusCode,
};
use structopt::StructOpt;
use tokio::{
net::UdpSocket,
};
use crate::prelude::*;
use protocol::PeerId;
#[derive (Debug, StructOpt)]
pub struct Opt {
#[structopt (long)]
listen_addr: Option <String>,
:heavy_minus_sign: remove old unused TCP server from PTTH_QUIC relay
2022-10-31 17:55:29 +00:00
#[structopt (long)]
:star: add udp_over_tcp server to PTTH_QUIC relay server maybe it'll work
2022-10-31 18:11:31 +00:00
tcp_listen_port: Option <u16>,
:recycle: refactor: break out most of the relay server into a module I want an end-to-end test that runs all 5 nodes in 1 process, like PTTH itself has.
2022-10-09 16:43:50 +00:00
}
pub async fn main (opt: Opt) -> anyhow::Result <()>
{
let listen_addr = opt.listen_addr.unwrap_or_else (|| String::from ("0.0.0.0:30380")).parse ()?;
:construction: wip: PTTH_QUIC to quinn 0.9.x
2022-12-16 19:49:34 +00:00
let (endpoint, server_cert) = make_server_endpoint (listen_addr)?;
:recycle: refactor: break out most of the relay server into a module I want an end-to-end test that runs all 5 nodes in 1 process, like PTTH itself has.
2022-10-09 16:43:50 +00:00
println! ("Base64 cert: {}", base64::encode (&server_cert));
tokio::fs::create_dir_all ("ptth_quic_output").await?;
tokio::fs::write ("ptth_quic_output/quic_server.crt", &server_cert).await?;
let relay_state = Arc::new (RelayState::default ());
let make_svc = {
let relay_state = Arc::clone (&relay_state);
make_service_fn (move |_conn| {
let relay_state = Arc::clone (&relay_state);
async move {
Ok::<_, String> (service_fn (move |req| {
let relay_state = Arc::clone (&relay_state);
handle_http (req, relay_state)
}))
}
})
};
let http_addr = SocketAddr::from (([0, 0, 0, 0], 4004));
let http_server = Server::bind (&http_addr);
let task_quic_server = {
let relay_state = Arc::clone (&relay_state);
tokio::spawn (async move {
:construction: wip: PTTH_QUIC to quinn 0.9.x
2022-12-16 19:49:34 +00:00
while let Some (conn) = endpoint.accept ().await {
:recycle: refactor: break out most of the relay server into a module I want an end-to-end test that runs all 5 nodes in 1 process, like PTTH itself has.
2022-10-09 16:43:50 +00:00
let relay_state = Arc::clone (&relay_state);
// Each new peer QUIC connection gets its own task
tokio::spawn (async move {
let active = relay_state.stats.quic.connect ();
debug! ("QUIC connections: {}", active);
match handle_quic_connection (Arc::clone (&relay_state), conn).await {
Ok (_) => (),
Err (e) => warn! ("handle_quic_connection `{:?}`", e),
}
let active = relay_state.stats.quic.disconnect ();
debug! ("QUIC connections: {}", active);
});
}
Ok::<_, anyhow::Error> (())
})
};
let task_direc_server = {
let relay_state = Arc::clone (&relay_state);
tokio::spawn (async move {
let sock = UdpSocket::bind("0.0.0.0:30379").await?;
let mut buf = [0; 2048];
loop {
let (len, addr) = sock.recv_from (&mut buf).await?;
debug! ("{:?} bytes received from {:?}", len, addr);
let packet = Vec::from_iter ((&buf [0..len]).into_iter ().map (|x| *x));
{
let mut direc_cookies = relay_state.direc_cookies.lock ().await;
if let Some (direc_state) = direc_cookies.remove (&packet) {
debug! ("Got PTTH_DIREC cookie for {}", direc_state.p2_id);
direc_state.p2_addr.send (addr).ok ();
}
else {
debug! ("UDP packet didn't match any PTTH_DIREC cookie");
}
}
}
Ok::<_, anyhow::Error> (())
})
};
let task_http_server = tokio::spawn (async move {
http_server.serve (make_svc).await?;
Ok::<_, anyhow::Error> (())
});
debug! ("Serving HTTP on {:?}", http_addr);
:star: use udp_over_tcp for both PTTH_QUIC end server and relay server - `cargo run -p ptth_quic --bin ptth_quic_relay_server -- --tcp-listen-port 4440` - `cargo run -p ptth_quic --bin ptth_quic_end_server -- --use-udp-over-tcp true --relay-addr 127.0.0.1:4440 --server-id my_server` - `cargo run -p ptth_quic_client_gui -- --client-id me` - `nc -l -p 5900` - (Open my_server, 5900 in the client GUI) - `nc 127.0.0.1 50369` (or whatever port the GUI picked)
2022-10-31 18:50:42 +00:00
if let Some (tcp_listen_port) = opt.tcp_listen_port {
:star: add udp_over_tcp server to PTTH_QUIC relay server maybe it'll work
2022-10-31 18:11:31 +00:00
tokio::spawn (async move {
let cfg = udp_over_tcp::server::Config {
tcp_port: tcp_listen_port,
udp_port: listen_addr.port (),
};
:star: use udp_over_tcp for both PTTH_QUIC end server and relay server - `cargo run -p ptth_quic --bin ptth_quic_relay_server -- --tcp-listen-port 4440` - `cargo run -p ptth_quic --bin ptth_quic_end_server -- --use-udp-over-tcp true --relay-addr 127.0.0.1:4440 --server-id my_server` - `cargo run -p ptth_quic_client_gui -- --client-id me` - `nc -l -p 5900` - (Open my_server, 5900 in the client GUI) - `nc 127.0.0.1 50369` (or whatever port the GUI picked)
2022-10-31 18:50:42 +00:00
if let Err (e) = udp_over_tcp::server::main (cfg).await {
eprintln! ("udp_over_tcp::server::main exited with err {:?}", e);
:star: add udp_over_tcp server to PTTH_QUIC relay server maybe it'll work
2022-10-31 18:11:31 +00:00
}
:star: use udp_over_tcp for both PTTH_QUIC end server and relay server - `cargo run -p ptth_quic --bin ptth_quic_relay_server -- --tcp-listen-port 4440` - `cargo run -p ptth_quic --bin ptth_quic_end_server -- --use-udp-over-tcp true --relay-addr 127.0.0.1:4440 --server-id my_server` - `cargo run -p ptth_quic_client_gui -- --client-id me` - `nc -l -p 5900` - (Open my_server, 5900 in the client GUI) - `nc 127.0.0.1 50369` (or whatever port the GUI picked)
2022-10-31 18:50:42 +00:00
});
}
:star: add udp_over_tcp server to PTTH_QUIC relay server maybe it'll work
2022-10-31 18:11:31 +00:00
tokio::select! {
_val = task_quic_server => {
eprintln! ("QUIC relay server exited, exiting");
},
_val = task_http_server => {
eprintln! ("HTTP server exited, exiting");
},
_val = task_direc_server => {
eprintln! ("PTTH_DIREC server exited, exiting");
},
}
:recycle: refactor: break out most of the relay server into a module I want an end-to-end test that runs all 5 nodes in 1 process, like PTTH itself has.
2022-10-09 16:43:50 +00:00
Ok (())
}
async fn handle_http (_req: Request <Body>, relay_state: Arc <RelayState>)
-> anyhow::Result <Response <Body>>
{
let debug_string;
{
let p4_server_proxies = relay_state.p4_server_proxies.lock ().await;
debug_string = format! ("{:#?}\n", p4_server_proxies.keys ().collect::<Vec<_>> ());
}
let resp = Response::builder ()
.status (StatusCode::OK)
.header ("content-type", "text/plain")
.body (Body::from (debug_string))?;
Ok (resp)
}
#[derive (Default)]
struct RelayState {
p4_server_proxies: Mutex <HashMap <PeerId, P4State>>,
direc_cookies: Mutex <HashMap <Vec <u8>, DirecState>>,
stats: Stats,
}
struct DirecState {
start_time: Instant,
p2_id: PeerId,
p2_addr: tokio::sync::oneshot::Sender <SocketAddr>,
}
#[derive (Default)]
struct Stats {
quic: ConnectEvents,
}
#[derive (Default)]
struct ConnectEvents {
connects: AtomicU64,
disconnects: AtomicU64,
}
impl ConnectEvents {
fn connect (&self) -> u64 {
let connects = self.connects.fetch_add (1, Ordering::Relaxed) + 1;
let disconnects = self.disconnects.load (Ordering::Relaxed);
connects - disconnects
}
fn disconnect (&self) -> u64 {
let disconnects = self.disconnects.fetch_add (1, Ordering::Relaxed) + 1;
let connects = self.connects.load (Ordering::Relaxed);
connects - disconnects
}
fn _active (&self) -> u64 {
let connects = self.connects.load (Ordering::Relaxed);
let disconnects = self.disconnects.load (Ordering::Relaxed);
connects - disconnects
}
}
struct P4State {
req_channel: mpsc::Sender <RequestP2ToP4>,
}
impl RelayState {
}
struct RequestP2ToP4 {
client_send: quinn::SendStream,
client_recv: quinn::RecvStream,
client_id: String,
}
struct PtthNewConnection {
client_send: quinn::SendStream,
client_recv: quinn::RecvStream,
server_send: quinn::SendStream,
server_recv: quinn::RecvStream,
}
struct PtthConnection {
uplink_task: JoinHandle <anyhow::Result <()>>,
downlink_task: JoinHandle <anyhow::Result <()>>,
}
impl PtthNewConnection {
fn build (self) -> PtthConnection {
let Self {
mut client_send,
mut client_recv,
mut server_send,
mut server_recv,
} = self;
let uplink_task = tokio::spawn (async move {
// Uplink - Client to end server
let mut buf = vec! [0u8; 65_536];
while let Some (bytes_read) = client_recv.read (&mut buf).await? {
if bytes_read == 0 {
break;
}
let buf_slice = &buf [0..bytes_read];
trace! ("Uplink relaying {} bytes", bytes_read);
server_send.write_all (buf_slice).await?;
}
trace! ("Uplink closed");
Ok::<_, anyhow::Error> (())
});
let downlink_task = tokio::spawn (async move {
// Downlink - End server to client
let mut buf = vec! [0u8; 65_536];
while let Some (bytes_read) = server_recv.read (&mut buf).await? {
let buf_slice = &buf [0..bytes_read];
trace! ("Downlink relaying {} bytes", bytes_read);
client_send.write_all (buf_slice).await?;
}
trace! ("Downlink closed");
Ok::<_, anyhow::Error> (())
});
PtthConnection {
uplink_task,
downlink_task,
}
}
}
async fn handle_quic_connection (
relay_state: Arc <RelayState>,
conn: quinn::Connecting,
) -> anyhow::Result <()>
{
:construction: wip: PTTH_QUIC to quinn 0.9.x
2022-12-16 19:49:34 +00:00
let conn = conn.await?;
:recycle: refactor: break out most of the relay server into a module I want an end-to-end test that runs all 5 nodes in 1 process, like PTTH itself has.
2022-10-09 16:43:50 +00:00
// Everyone who connects must identify themselves with the first
// bi stream
// TODO: Timeout
:construction: wip: PTTH_QUIC to quinn 0.9.x
2022-12-16 19:49:34 +00:00
let (mut send, mut recv) = conn.accept_bi ().await?;
:recycle: refactor: break out most of the relay server into a module I want an end-to-end test that runs all 5 nodes in 1 process, like PTTH itself has.
2022-10-09 16:43:50 +00:00
let peer = protocol::p3_accept_peer (&mut recv).await?;
match peer {
protocol::P3Peer::P2ClientProxy (peer) => {
:loud_sound: add random IDs for QUIC connection events
2022-12-16 19:25:21 +00:00
trace! ("H36JTVE5 Accepting connection from P2 client");
:recycle: refactor: break out most of the relay server into a module I want an end-to-end test that runs all 5 nodes in 1 process, like PTTH itself has.
2022-10-09 16:43:50 +00:00
// TODO: Check authorization for P2 peers
protocol::p3_authorize_p2_peer (&mut send).await?;
handle_p2_connection (relay_state, conn, peer).await?;
},
protocol::P3Peer::P4ServerProxy (peer) => {
:loud_sound: add random IDs for QUIC connection events
2022-12-16 19:25:21 +00:00
trace! ("LRHUKB7K Accepting connection from P4 end server");
:recycle: refactor: break out most of the relay server into a module I want an end-to-end test that runs all 5 nodes in 1 process, like PTTH itself has.
2022-10-09 16:43:50 +00:00
// TODO: Check authorization for P4 peers
protocol::p3_authorize_p4_peer (&mut send).await?;
handle_p4_connection (relay_state, conn, peer).await?;
},
}
Ok::<_, anyhow::Error> (())
}
async fn handle_p2_connection (
relay_state: Arc <RelayState>,
:construction: wip: PTTH_QUIC to quinn 0.9.x
2022-12-16 19:49:34 +00:00
conn: quinn::Connection,
:recycle: refactor: break out most of the relay server into a module I want an end-to-end test that runs all 5 nodes in 1 process, like PTTH itself has.
2022-10-09 16:43:50 +00:00
peer: protocol::P2ClientProxy,
) -> anyhow::Result <()>
{
let client_id = peer.id;
:construction: wip: PTTH_QUIC to quinn 0.9.x
2022-12-16 19:49:34 +00:00
while let Ok ((send, mut recv)) = conn.accept_bi ().await {
:recycle: refactor: break out most of the relay server into a module I want an end-to-end test that runs all 5 nodes in 1 process, like PTTH itself has.
2022-10-09 16:43:50 +00:00
let relay_state = Arc::clone (&relay_state);
let client_id = client_id.clone ();
tokio::spawn (async move {
debug! ("Request started for P2");
match protocol::p3_accept_p2_stream (&mut recv).await? {
protocol::P2ToP3Stream::ConnectP2ToP4 {
server_id,
} => {
handle_request_p2_to_p4 (
relay_state,
client_id,
server_id,
send,
recv
).await?
},
protocol::P2ToP3Stream::DirecP2ToP4 {
server_id,
cookie,
} => {
handle_direc_p2_to_p4 (
relay_state,
client_id,
server_id,
cookie,
send,
recv
).await?
},
}
debug! ("Request ended for P2");
Ok::<_, anyhow::Error> (())
});
}
debug! ("P2 {} disconnected", client_id);
Ok (())
}
async fn handle_request_p2_to_p4 (
relay_state: Arc <RelayState>,
client_id: String,
server_id: PeerId,
mut client_send: quinn::SendStream,
client_recv: quinn::RecvStream,
) -> anyhow::Result <()>
{
trace! ("P2 {} wants to connect to P4 {}", client_id, server_id);
// TODO: Check authorization for P2 to connect to P4
protocol::p3_authorize_p2_to_p4_connection (&mut client_send).await?;
{
let p4_server_proxies = relay_state.p4_server_proxies.lock ().await;
match p4_server_proxies.get (&server_id) {
Some (p4_state) => {
p4_state.req_channel.send (RequestP2ToP4 {
client_send,
client_recv,
client_id,
}).await.map_err (|_| anyhow::anyhow! ("Can't send request to P4 server"))?;
},
None => warn! ("That server isn't connected"),
}
}
Ok (())
}
async fn handle_direc_p2_to_p4 (
relay_state: Arc <RelayState>,
client_id: String,
server_id: PeerId,
cookie: Vec <u8>,
mut client_send: quinn::SendStream,
client_recv: quinn::RecvStream,
) -> anyhow::Result <()>
{
debug! ("P2 {} wants a P2P connection to P4 {}", client_id, server_id);
// TODO: Check authorization
protocol::p3_authorize_p2_to_p4_direc (&mut client_send).await?;
let (tx, rx) = tokio::sync::oneshot::channel ();
{
let mut direc_cookies = relay_state.direc_cookies.lock ().await;
direc_cookies.insert (cookie, DirecState {
start_time: Instant::now (),
p2_id: client_id.clone (),
p2_addr: tx,
});
}
debug! ("Waiting to learn P2's WAN address...");
let wan_addr = rx.await?;
debug! ("And that WAN address is {}", wan_addr);
Ok (())
}
async fn handle_p4_connection (
relay_state: Arc <RelayState>,
:construction: wip: PTTH_QUIC to quinn 0.9.x
2022-12-16 19:49:34 +00:00
connection: quinn::Connection,
:recycle: refactor: break out most of the relay server into a module I want an end-to-end test that runs all 5 nodes in 1 process, like PTTH itself has.
2022-10-09 16:43:50 +00:00
peer: protocol::P4ServerProxy,
) -> anyhow::Result <()>
{
let server_id = peer.id;
let (tx, mut rx) = mpsc::channel (2);
let p4_state = P4State {
req_channel: tx,
};
{
let mut p4_server_proxies = relay_state.p4_server_proxies.lock ().await;
p4_server_proxies.insert (server_id.clone (), p4_state);
}
while let Some (req) = rx.recv ().await {
let connection = connection.clone ();
let server_id = server_id.clone ();
tokio::spawn (async move {
let RequestP2ToP4 {
client_send,
client_recv,
client_id,
} = req;
debug! ("P4 {} got a request from P2 {}", server_id, client_id);
let (server_send, server_recv) = protocol::p3_connect_p2_to_p4 (&connection, &client_id).await?;
trace! ("Relaying bytes...");
let ptth_conn = PtthNewConnection {
client_send,
client_recv,
server_send,
server_recv,
}.build ();
ptth_conn.uplink_task.await??;
ptth_conn.downlink_task.await??;
debug! ("Request ended for P4");
Ok::<_, anyhow::Error> (())
});
}
debug! ("P4 {} disconnected", server_id);
Ok (())
}