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 24e0635e30
commit ad6c8278d2
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

238
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");
@ -406,38 +407,38 @@ async fn main() {
// Read the JSON contents of the file as a ResourceInfo // Read the JSON contents of the file as a ResourceInfo
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?
let bridgedb = BridgeDb::new(); let bridgedb = BridgeDb::new();
let lox_auth = BridgeAuth::new(bridgedb.pubkey); let lox_auth = BridgeAuth::new(bridgedb.pubkey);
let context = LoxServerContext { let context = LoxServerContext {
db: Arc::new(Mutex::new(bridgedb)), db: Arc::new(Mutex::new(bridgedb)),
ba: Arc::new(Mutex::new(lox_auth)), ba: Arc::new(Mutex::new(lox_auth)),
}; };
while let Some(cmd) = context_rx.recv().await { while let Some(cmd) = context_rx.recv().await {
use Command::*; use Command::*;
match cmd { match cmd {
Rdsys {resourcediff} => { Rdsys { resourcediff } => {
for new_resource in resourcediff.new { for new_resource in resourcediff.new {
for pt in new_resource { for pt in new_resource {
println!("A NEW RESOURCE: {:?}", pt); println!("A NEW RESOURCE: {:?}", pt);
let mut bucket = [ let mut bucket = [
BridgeLine::default(), BridgeLine::default(),
BridgeLine::default(), BridgeLine::default(),
BridgeLine::default(), BridgeLine::default(),
]; ];
let mut count = 0; let mut count = 0;
for resource in pt.1 { for resource in pt.1 {
let mut ip_bytes: [u8; 16] = [0; 16]; let mut ip_bytes: [u8; 16] = [0; 16];
ip_bytes[..resource.address.len()] ip_bytes[..resource.address.len()]
.copy_from_slice(resource.address.as_bytes()); .copy_from_slice(resource.address.as_bytes());
let infostr: String = format!( let infostr: String = format!(
"type={} blocked_in={:?} protocol={} fingerprint={} or_addresses={:?} distribution={} flags={:?} params={:?}", "type={} blocked_in={:?} protocol={} fingerprint={} or_addresses={:?} distribution={} flags={:?} params={:?}",
resource.r#type, resource.r#type,
resource.blocked_in, resource.blocked_in,
@ -448,42 +449,43 @@ 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());
let bridgeline = BridgeLine { info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes());
addr: ip_bytes, let bridgeline = BridgeLine {
port: resource.port, addr: ip_bytes,
info: info_bytes, port: resource.port,
}; info: info_bytes,
};
println!("Now it's a bridgeline: {:?}", bridgeline);
if count < 2 { println!("Now it's a bridgeline: {:?}", bridgeline);
bucket[count] = bridgeline; if count < 2 {
count += 1; bucket[count] = bridgeline;
} else { count += 1;
context.add_openinv_bucket(bucket); } else {
count = 0; context.add_openinv_bucket(bucket);
bucket = [ count = 0;
BridgeLine::default(), bucket = [
BridgeLine::default(), BridgeLine::default(),
BridgeLine::default(), BridgeLine::default(),
]; BridgeLine::default(),
];
}
} }
} }
} }
} for changed_resource in resourcediff.changed {
for changed_resource in resourcediff.changed { println!("A NEW CHANGED RESOURCE: {:?}", changed_resource);
println!("A NEW CHANGED RESOURCE: {:?}", changed_resource); }
} for gone_resource in resourcediff.gone {
for gone_resource in resourcediff.gone { for pt in gone_resource {
for pt in gone_resource { println!("A NEW GONE RESOURCE: {:?}", pt);
println!("A NEW GONE RESOURCE: {:?}", pt); for resource in pt.1 {
for resource in pt.1 { let mut ip_bytes: [u8; 16] = [0; 16];
let mut ip_bytes: [u8; 16] = [0; 16]; ip_bytes[..resource.address.len()]
ip_bytes[..resource.address.len()] .copy_from_slice(resource.address.as_bytes());
.copy_from_slice(resource.address.as_bytes()); let infostr: String = format!(
let infostr: String = format!(
"type={} blocked_in={:?} protocol={} fingerprint={} or_addresses={:?} distribution={} flags={:?} params={:?}", "type={} blocked_in={:?} protocol={} fingerprint={} or_addresses={:?} distribution={} flags={:?} params={:?}",
resource.r#type, resource.r#type,
resource.blocked_in, resource.blocked_in,
@ -494,28 +496,29 @@ 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());
let bridgeline = BridgeLine { info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes());
addr: ip_bytes, let bridgeline = BridgeLine {
port: resource.port, addr: ip_bytes,
info: info_bytes, port: resource.port,
}; info: info_bytes,
};
println!("Now it's a bridgeline: {:?}", bridgeline);
context.add_unreachable(bridgeline); println!("Now it's a bridgeline: {:?}", bridgeline);
context.add_unreachable(bridgeline);
}
} }
} }
context.encrypt_table();
}
Request { req, sender } => {
let response = handle(context.clone(), req).await;
sender.send(response);
} }
context.encrypt_table();
} }
Request {request} => {
} }
}
}
}); });
//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;
} }