vecna
/
lox
mirror of https://gitlab.torproject.org/vecna/lox.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Implements server functionality, next get thread working

This commit is contained in:
onyinyang 2023-03-14 23:23:40 -04:00
parent 34f1d0399f
commit 3b0c8ab5b6
No known key found for this signature in database
GPG Key ID: 156A6435430C2036
1 changed files with 127 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
crates/lox-distributor/src/main.rs
View File

@ -1,9 +1,10 @@
use async_channel::{Sender, Receiver}; use async_channel::{Receiver, Sender};
use futures::future; use futures::future;
use hyper::{ use hyper::{
body, body,
body::Bytes, body::Bytes,
header::HeaderValue, header::HeaderValue,
http::response,
server::conn::AddrStream, server::conn::AddrStream,
service::{make_service_fn, service_fn}, service::{make_service_fn, service_fn},
Body, Method, Request, Response, Server, StatusCode, Body, Method, Request, Response, Server, StatusCode,
@ -32,7 +33,7 @@ use std::{
use serde_json; use serde_json;
use serde_with::serde_as; use serde_with::serde_as;
use tokio::{spawn, time::sleep, sync::mpsc}; use tokio::{spawn, sync::{mpsc, oneshot}, time::sleep};
#[serde_as] #[serde_as]
#[derive(Serialize, Deserialize)] #[derive(Serialize, Deserialize)]
@ -208,9 +209,9 @@ impl LoxServerContext {
} }
} }
async fn handle( async fn handle(
context: LoxServerContext, cloned_context: LoxServerContext,
// addr: SocketAddr,
req: Request<Body>, req: Request<Body>,
) -> Result<Response<Body>, Infallible> { ) -> Result<Response<Body>, Infallible> {
println!("Request: {:?}", req); println!("Request: {:?}", req);
@ -223,41 +224,41 @@ async fn handle(
.body(Body::from("Allow POST")) .body(Body::from("Allow POST"))
.unwrap()), .unwrap()),
_ => match (req.method(), req.uri().path()) { _ => match (req.method(), req.uri().path()) {
(&Method::GET, "/invite") => Ok::<_, Infallible>(generate_invite(context)), (&Method::GET, "/invite") => Ok::<_, Infallible>(generate_invite(cloned_context)),
(&Method::GET, "/reachability") => Ok::<_, Infallible>(send_reachability_cred(context)), (&Method::GET, "/reachability") => Ok::<_, Infallible>(send_reachability_cred(cloned_context)),
(&Method::GET, "/pubkeys") => Ok::<_, Infallible>(send_keys(context)), (&Method::GET, "/pubkeys") => Ok::<_, Infallible>(send_keys(cloned_context)),
(&Method::POST, "/openreq") => Ok::<_, Infallible>({ (&Method::POST, "/openreq") => Ok::<_, Infallible>({
let bytes = body::to_bytes(req.into_body()).await.unwrap(); let bytes = body::to_bytes(req.into_body()).await.unwrap();
verify_and_send_open_cred(bytes, context) verify_and_send_open_cred(bytes, cloned_context)
}), }),
(&Method::POST, "/trustpromo") => Ok::<_, Infallible>({ (&Method::POST, "/trustpromo") => Ok::<_, Infallible>({
let bytes = body::to_bytes(req.into_body()).await.unwrap(); let bytes = body::to_bytes(req.into_body()).await.unwrap();
verify_and_send_trust_promo(bytes, context) verify_and_send_trust_promo(bytes, cloned_context)
}), }),
(&Method::POST, "/trustmig") => Ok::<_, Infallible>({ (&Method::POST, "/trustmig") => Ok::<_, Infallible>({
let bytes = body::to_bytes(req.into_body()).await.unwrap(); let bytes = body::to_bytes(req.into_body()).await.unwrap();
verify_and_send_trust_migration(bytes, context) verify_and_send_trust_migration(bytes, cloned_context)
}), }),
(&Method::POST, "/levelup") => Ok::<_, Infallible>({ (&Method::POST, "/levelup") => Ok::<_, Infallible>({
let bytes = body::to_bytes(req.into_body()).await.unwrap(); let bytes = body::to_bytes(req.into_body()).await.unwrap();
verify_and_send_level_up(bytes, context) verify_and_send_level_up(bytes, cloned_context)
}), }),
(&Method::POST, "/issueinvite") => Ok::<_, Infallible>({ (&Method::POST, "/issueinvite") => Ok::<_, Infallible>({
let bytes = body::to_bytes(req.into_body()).await.unwrap(); let bytes = body::to_bytes(req.into_body()).await.unwrap();
verify_and_send_issue_invite(bytes, context) verify_and_send_issue_invite(bytes, cloned_context)
}), }),
(&Method::POST, "/redeem") => Ok::<_, Infallible>({ (&Method::POST, "/redeem") => Ok::<_, Infallible>({
let bytes = body::to_bytes(req.into_body()).await.unwrap(); let bytes = body::to_bytes(req.into_body()).await.unwrap();
verify_and_send_redeem_invite(bytes, context) verify_and_send_redeem_invite(bytes, cloned_context)
}), }),
(&Method::POST, "/checkblockage") => Ok::<_, Infallible>({ (&Method::POST, "/checkblockage") => Ok::<_, Infallible>({
let bytes = body::to_bytes(req.into_body()).await.unwrap(); let bytes = body::to_bytes(req.into_body()).await.unwrap();
// TEST ONLY: Block all existing bridges and add new ones for migration // TEST ONLY: Block all existing bridges and add new ones for migration
verify_and_send_check_blockage(bytes, context) verify_and_send_check_blockage(bytes, cloned_context)
}), }),
(&Method::POST, "/blockagemigration") => Ok::<_, Infallible>({ (&Method::POST, "/blockagemigration") => Ok::<_, Infallible>({
let bytes = body::to_bytes(req.into_body()).await.unwrap(); let bytes = body::to_bytes(req.into_body()).await.unwrap();
verify_and_send_blockage_migration(bytes, context) verify_and_send_blockage_migration(bytes, cloned_context)
}), }),
_ => { _ => {
// Return 404 not found response. // Return 404 not found response.
@ -378,27 +379,27 @@ async fn rdsys_sender(rstream: ResourceStream, tx: Sender<ResourceDiff>) {
} }
} }
async fn parse_bridges(rdsys_tx: Sender<Command>, rx: Receiver<ResourceDiff>) { async fn parse_bridges(rdsys_tx: mpsc::Sender<Command>, rx: Receiver<ResourceDiff>) {
loop { loop {
let resourcediff = rx.recv().await.unwrap(); let resourcediff = rx.recv().await.unwrap();
let cmd = Command::Rdsys { resourcediff: resourcediff, }; let cmd = Command::Rdsys {
resourcediff: resourcediff,
};
rdsys_tx.send(cmd).await.unwrap(); rdsys_tx.send(cmd).await.unwrap();
} }
} }
#[derive(Debug)] #[derive(Debug)]
enum Command { enum Command {
Rdsys { Rdsys { resourcediff: ResourceDiff },
resourcediff: ResourceDiff,
},
Request { Request {
request: Request<Body>, req: Request<Body>,
} sender: oneshot::Sender<Result<Response<Body>, Infallible>>,
},
} }
// Run with cargo run -- config.json // Run with cargo run -- config.json
#[tokio::main(worker_threads = 2)] #[tokio::main]
async fn main() { async fn main() {
let args: Vec<String> = env::args().collect(); let args: Vec<String> = env::args().collect();
let file = File::open(&args[1]).expect("Should have been able to read config.json file"); let file = File::open(&args[1]).expect("Should have been able to read config.json file");
@ -407,7 +408,7 @@ async fn main() {
let rtype: ResourceInfo = serde_json::from_reader(reader).unwrap(); let rtype: ResourceInfo = serde_json::from_reader(reader).unwrap();
let (rdsys_tx, mut context_rx) = mpsc::channel(32); let (rdsys_tx, mut context_rx) = mpsc::channel(32);
let request_tx = rdsys_tx.clone(); let mut request_tx = rdsys_tx.clone();
let context_manager = spawn(async move { let context_manager = spawn(async move {
// pass in distribution of open invite vs. hot spare buckets? // pass in distribution of open invite vs. hot spare buckets?
@ -448,7 +449,8 @@ let request_tx = rdsys_tx.clone();
resource.flags, resource.flags,
resource.params, resource.params,
); );
let mut info_bytes: [u8; BRIDGE_BYTES - 18] = [0; BRIDGE_BYTES - 18]; let mut info_bytes: [u8; BRIDGE_BYTES - 18] =
[0; BRIDGE_BYTES - 18];
info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes()); info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes());
let bridgeline = BridgeLine { let bridgeline = BridgeLine {
@ -494,7 +496,8 @@ let request_tx = rdsys_tx.clone();
resource.flags, resource.flags,
resource.params, resource.params,
); );
let mut info_bytes: [u8; BRIDGE_BYTES - 18] = [0; BRIDGE_BYTES - 18]; let mut info_bytes: [u8; BRIDGE_BYTES - 18] =
[0; BRIDGE_BYTES - 18];
info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes()); info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes());
let bridgeline = BridgeLine { let bridgeline = BridgeLine {
@ -510,12 +513,12 @@ let request_tx = rdsys_tx.clone();
} }
context.encrypt_table(); context.encrypt_table();
} }
Request {request} => { Request { req, sender } => {
let response = handle(context.clone(), req).await;
sender.send(response);
} }
} }
} }
}); });
//Sender is resource stream and receiver is bridgedb function (add_openinv_bridges) //Sender is resource stream and receiver is bridgedb function (add_openinv_bridges)
@ -524,25 +527,38 @@ let request_tx = rdsys_tx.clone();
let rstream = start_stream(rtype.endpoint, rtype.name, rtype.token, rtype.types) let rstream = start_stream(rtype.endpoint, rtype.name, rtype.token, rtype.types)
.await .await
.unwrap(); .unwrap();
let rdsys_sender_handle = spawn(async {rdsys_sender(rstream, tx).await }); let rdsys_stream_handler = spawn(async { rdsys_sender(rstream, tx).await });
let bridge_parser = spawn( async {parse_bridges(rdsys_tx, rx).await}); let rdsys_resource_receiver = spawn(async { parse_bridges(rdsys_tx, rx).await });
future::join_all([rdsys_sender_handle, bridge_parser]).await;
/*
let new_service = make_service_fn(move |_conn: &AddrStream| { let make_service = make_service_fn(|_conn: &AddrStream| {
let service = service_fn(move |req| handle(context, req)); let request_tx = request_tx.clone();
let service = service_fn( move |req| {
let request_tx = request_tx.clone();
let (response_tx, response_rx) = oneshot::channel();
let cmd = Command::Request {
req: req,
sender: response_tx,
};
async move {
request_tx.send(cmd).await.unwrap();
response_rx.await.unwrap()
}
});
async move { Ok::<_, Infallible>(service) } async move { Ok::<_, Infallible>(service) }
}); });
let addr = SocketAddr::from(([127, 0, 0, 1], 8001)); let addr = SocketAddr::from(([127, 0, 0, 1], 8001));
// let server = Server::bind(&addr).serve(new_service); let server = Server::bind(&addr).serve(make_service);
// let graceful = server.with_graceful_shutdown(shutdown_signal()); let graceful = server.with_graceful_shutdown(shutdown_signal());
println!("Listening on {}", addr); println!("Listening on {}", addr);
if let Err(e) = graceful.await { if let Err(e) = graceful.await {
eprintln!("server error: {}", e); eprintln!("server error: {}", e);
} }
*/
future::join_all([rdsys_stream_handler, rdsys_resource_receiver]).await;
} }