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

Add distributor logic for bridge updates from rdsys, handling for unsuccessful removals/updates

This commit is contained in:
onyinyang 2023-03-20 12:42:40 -04:00
parent 85b4d63358
commit 47f645a4f9
No known key found for this signature in database
GPG Key ID: 156A6435430C2036
1 changed files with 174 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -19,6 +19,8 @@ use lox::{BridgeAuth, BridgeDb, OPENINV_LENGTH};
use rand::RngCore; use rand::RngCore;
use rdsys_backend::{proto::ResourceDiff, start_stream}; use rdsys_backend::{proto::ResourceDiff, start_stream};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use serde_json;
use serde_with::serde_as;
use std::{ use std::{
convert::Infallible, convert::Infallible,
env, env,
@ -28,12 +30,10 @@ use std::{
sync::{Arc, Mutex}, sync::{Arc, Mutex},
time::Duration, time::Duration,
}; };
use serde_json;
use serde_with::serde_as;
use tokio::{ use tokio::{
spawn, signal, spawn,
sync::{broadcast, mpsc, oneshot}, sync::{broadcast, mpsc, oneshot},
time::sleep, signal, time::sleep,
}; };
#[serde_as] #[serde_as]
@ -113,10 +113,16 @@ impl LoxServerContext {
ba_obj.add_openinv_bridges(bucket, &mut db_obj); ba_obj.add_openinv_bridges(bucket, &mut db_obj);
} }
fn add_unreachable(&self, bridgeline: BridgeLine) { fn add_unreachable(&self, bridgeline: BridgeLine) -> bool {
let mut ba_obj = self.ba.lock().unwrap(); let mut ba_obj = self.ba.lock().unwrap();
let mut db_obj = self.db.lock().unwrap(); let mut db_obj = self.db.lock().unwrap();
ba_obj.bridge_unreachable(&bridgeline, &mut db_obj); ba_obj.bridge_unreachable(&bridgeline, &mut db_obj)
}
fn update_bridge(&self, bridgeline: BridgeLine) -> bool {
let mut ba_obj = self.ba.lock().unwrap();
let mut db_obj = self.db.lock().unwrap();
ba_obj.bridge_update(&bridgeline, &mut db_obj)
} }
fn advance_days_TEST(&self, num: u16) { fn advance_days_TEST(&self, num: u16) {
@ -274,7 +280,6 @@ async fn handle(
} }
} }
// Generate and return an open invitation token // Generate and return an open invitation token
fn generate_invite(context: LoxServerContext) -> Response<Body> { fn generate_invite(context: LoxServerContext) -> Response<Body> {
let invite = context.gen_invite(); let invite = context.gen_invite();
@ -378,8 +383,12 @@ async fn shutdown_signal() {
println!("Shut down Lox Server"); println!("Shut down Lox Server");
} }
async fn rdsys_stream(rtype: ResourceInfo, tx: mpsc::Sender<ResourceDiff>, mut kill: broadcast::Receiver<()>) { async fn rdsys_stream(
tokio:: select! { rtype: ResourceInfo,
tx: mpsc::Sender<ResourceDiff>,
mut kill: broadcast::Receiver<()>,
) {
tokio::select! {
start_rdsys_stream = rdsys_sender(rtype, tx) => start_rdsys_stream , start_rdsys_stream = rdsys_sender(rtype, tx) => start_rdsys_stream ,
_ = kill.recv() => {println!("Shut down rdsys stream"); return}, _ = kill.recv() => {println!("Shut down rdsys stream"); return},
} }
@ -389,10 +398,10 @@ async fn rdsys_stream(rtype: ResourceInfo, tx: mpsc::Sender<ResourceDiff>, mut
// in the config.json file. // in the config.json file.
// TODO: ensure this stream gracefully shutdowns on the ctrl_c command. // TODO: ensure this stream gracefully shutdowns on the ctrl_c command.
async fn rdsys_sender(rtype: ResourceInfo, tx: mpsc::Sender<ResourceDiff>) { async fn rdsys_sender(rtype: ResourceInfo, tx: mpsc::Sender<ResourceDiff>) {
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
.expect("rdsys stream initialization failed. Start rdsys or check config.json"); .expect("rdsys stream initialization failed. Start rdsys or check config.json");
sleep(Duration::from_millis(1)).await; sleep(Duration::from_millis(1)).await;
for diff in rstream { for diff in rstream {
println!("Received diff: {:?}", diff); //send this through a channel println!("Received diff: {:?}", diff); //send this through a channel
tx.send(diff).await.unwrap(); tx.send(diff).await.unwrap();
@ -400,8 +409,12 @@ async fn rdsys_sender(rtype: ResourceInfo, tx: mpsc::Sender<ResourceDiff>) {
} }
} }
async fn rdsys_bridge_parser(rdsys_tx: mpsc::Sender<Command>, rx: mpsc::Receiver<ResourceDiff>, mut kill: broadcast::Receiver<()>) { async fn rdsys_bridge_parser(
tokio:: select! { rdsys_tx: mpsc::Sender<Command>,
rx: mpsc::Receiver<ResourceDiff>,
mut kill: broadcast::Receiver<()>,
) {
tokio::select! {
start_bridge_parser = parse_bridges(rdsys_tx, rx) => start_bridge_parser , start_bridge_parser = parse_bridges(rdsys_tx, rx) => start_bridge_parser ,
_ = kill.recv() => {println!("Shut down bridge_parser"); return}, _ = kill.recv() => {println!("Shut down bridge_parser"); return},
} }
@ -420,8 +433,11 @@ async fn parse_bridges(rdsys_tx: mpsc::Sender<Command>, mut rx: mpsc::Receiver<R
} }
} }
async fn create_context_manager(context_rx: mpsc::Receiver<Command>, mut kill: broadcast::Receiver<()>) { async fn create_context_manager(
tokio:: select! { context_rx: mpsc::Receiver<Command>,
mut kill: broadcast::Receiver<()>,
) {
tokio::select! {
create_context = context_manager(context_rx) => create_context, create_context = context_manager(context_rx) => create_context,
_ = kill.recv() => {println!("Shut down context_manager"); return}, _ = kill.recv() => {println!("Shut down context_manager"); return},
} }
@ -431,34 +447,33 @@ async fn create_context_manager(context_rx: mpsc::Receiver<Command>, mut kill: b
// that the DB can be updated from the rdsys stream and client requests // that the DB can be updated from the rdsys stream and client requests
// can be responded to with an updated BridgeDB state // can be responded to with an updated BridgeDB state
async fn context_manager(mut context_rx: mpsc::Receiver<Command>) { async fn context_manager(mut context_rx: mpsc::Receiver<Command>) {
let bridgedb = BridgeDb::new();
let lox_auth = BridgeAuth::new(bridgedb.pubkey);
let bridgedb = BridgeDb::new(); let context = LoxServerContext {
let lox_auth = BridgeAuth::new(bridgedb.pubkey); db: Arc::new(Mutex::new(bridgedb)),
ba: Arc::new(Mutex::new(lox_auth)),
};
let context = LoxServerContext { while let Some(cmd) = context_rx.recv().await {
db: Arc::new(Mutex::new(bridgedb)), use Command::*;
ba: Arc::new(Mutex::new(lox_auth)),
};
while let Some(cmd) = context_rx.recv().await { match cmd {
use Command::*; Rdsys { resourcediff } => {
for new_resource in resourcediff.new {
match cmd { for pt in new_resource {
Rdsys { resourcediff } => { println!("A NEW RESOURCE: {:?}", pt);
for new_resource in resourcediff.new { let mut bucket = [
for pt in new_resource { BridgeLine::default(),
println!("A NEW RESOURCE: {:?}", pt); BridgeLine::default(),
let mut bucket = [ BridgeLine::default(),
BridgeLine::default(), ];
BridgeLine::default(), let mut count = 0;
BridgeLine::default(), for resource in pt.1 {
]; let mut ip_bytes: [u8; 16] = [0; 16];
let mut count = 0; ip_bytes[..resource.address.len()]
for resource in pt.1 { .copy_from_slice(resource.address.as_bytes());
let mut ip_bytes: [u8; 16] = [0; 16]; let infostr: String = format!(
ip_bytes[..resource.address.len()]
.copy_from_slice(resource.address.as_bytes());
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,
@ -469,43 +484,77 @@ async fn context_manager(mut context_rx: mpsc::Receiver<Command>) {
resource.flags, resource.flags,
resource.params, resource.params,
); );
let mut info_bytes: [u8; BRIDGE_BYTES - 18] = let mut info_bytes: [u8; BRIDGE_BYTES - 18] = [0; 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 {
addr: ip_bytes, addr: ip_bytes,
port: resource.port, port: resource.port,
info: info_bytes, info: info_bytes,
}; };
println!("Now it's a bridgeline: {:?}", bridgeline); println!("Now it's a bridgeline: {:?}", bridgeline);
if count < 2 { if count < 2 {
bucket[count] = bridgeline; bucket[count] = bridgeline;
count += 1; count += 1;
} else { } else {
context.add_openinv_bucket(bucket); context.add_openinv_bucket(bucket);
count = 0; count = 0;
bucket = [ bucket = [
BridgeLine::default(), 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); for pt in changed_resource {
} println!("A NEW CHANGED RESOURCE: {:?}", pt);
for gone_resource in resourcediff.gone { for resource in pt.1 {
for pt in gone_resource { let mut ip_bytes: [u8; 16] = [0; 16];
println!("A NEW GONE RESOURCE: {:?}", pt); ip_bytes[..resource.address.len()]
for resource in pt.1 { .copy_from_slice(resource.address.as_bytes());
let mut ip_bytes: [u8; 16] = [0; 16]; let infostr: String = format!(
ip_bytes[..resource.address.len()] "type={} blocked_in={:?} protocol={} fingerprint={} or_addresses={:?} distribution={} flags={:?} params={:?}",
.copy_from_slice(resource.address.as_bytes()); resource.r#type,
let infostr: String = format!( resource.blocked_in,
resource.protocol,
resource.fingerprint,
resource.or_addresses,
resource.distribution,
resource.flags,
resource.params,
);
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 {
addr: ip_bytes,
port: resource.port,
info: info_bytes,
};
println!("BridgeLine to be changed: {:?}", bridgeline);
let res = context.update_bridge(bridgeline);
if res {
println!("BridgeLine successfully updated: {:?}", bridgeline);
} else {
println!("'Changed' BridgeLine NOT UPDATED!! : {:?}", bridgeline);
//TODO probably do something else here
}
}
}
}
for gone_resource in resourcediff.gone {
for pt in gone_resource {
println!("A NEW GONE RESOURCE: {:?}", pt);
for resource in pt.1 {
let mut ip_bytes: [u8; 16] = [0; 16];
ip_bytes[..resource.address.len()]
.copy_from_slice(resource.address.as_bytes());
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,
@ -516,40 +565,46 @@ async fn context_manager(mut context_rx: mpsc::Receiver<Command>) {
resource.flags, resource.flags,
resource.params, resource.params,
); );
let mut info_bytes: [u8; BRIDGE_BYTES - 18] = let mut info_bytes: [u8; BRIDGE_BYTES - 18] = [0; 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 {
addr: ip_bytes, addr: ip_bytes,
port: resource.port, port: resource.port,
info: info_bytes, info: info_bytes,
}; };
println!("Now it's a bridgeline: {:?}", bridgeline); println!("BridgeLine to be removed: {:?}", bridgeline);
context.add_unreachable(bridgeline); let res = context.add_unreachable(bridgeline);
} if res {
} println!(
} "BridgeLine successfully marked unreachable: {:?}",
context.encrypt_table(); bridgeline
sleep(Duration::from_millis(1)).await; );
} } else {
Request { req, sender } => { println!("'Gone' BridgeLine NOT REMOVED!! : {:?}", bridgeline);
let response = handle(context.clone(), req).await; //TODO probably do something else here
if let Err(e) = sender.send(response) { }
eprintln!("Server Response Error: {:?}", e); }
}; }
sleep(Duration::from_millis(1)).await; }
} context.encrypt_table();
Shutdown { shutdown_sig} => { sleep(Duration::from_millis(1)).await;
}
Request { req, sender } => {
let response = handle(context.clone(), req).await;
if let Err(e) = sender.send(response) {
eprintln!("Server Response Error: {:?}", e);
};
sleep(Duration::from_millis(1)).await;
}
Shutdown { shutdown_sig } => {
println!("Sending Shutdown Signal, all threads should shutdown."); println!("Sending Shutdown Signal, all threads should shutdown.");
drop(shutdown_sig); drop(shutdown_sig);
println!("Shutdown Sent."); println!("Shutdown Sent.");
} }
} }
} }
} }
// Each of the commands that the Context Manager handles // Each of the commands that the Context Manager handles
@ -564,7 +619,7 @@ enum Command {
}, },
Shutdown { Shutdown {
shutdown_sig: broadcast::Sender<()>, shutdown_sig: broadcast::Sender<()>,
} },
} }
#[tokio::main] #[tokio::main]
@ -573,22 +628,21 @@ async fn main() {
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");
let reader = BufReader::new(file); let reader = BufReader::new(file);
// 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).expect("Reading ResourceInfo from JSON failed."); let rtype: ResourceInfo =
serde_json::from_reader(reader).expect("Reading ResourceInfo from JSON failed.");
let (rdsys_tx, context_rx) = mpsc::channel(32); let (rdsys_tx, context_rx) = mpsc::channel(32);
let request_tx = rdsys_tx.clone(); let request_tx = rdsys_tx.clone();
let shutdown_cmd_tx = rdsys_tx.clone(); let shutdown_cmd_tx = rdsys_tx.clone();
// create the shutdown broadcast channel and clone for every thread // create the shutdown broadcast channel and clone for every thread
let (shutdown_tx, mut shutdown_rx) = broadcast::channel(16); let (shutdown_tx, mut shutdown_rx) = broadcast::channel(16);
let kill_stream = shutdown_tx.subscribe(); let kill_stream = shutdown_tx.subscribe();
let kill_parser = shutdown_tx.subscribe(); let kill_parser = shutdown_tx.subscribe();
let kill_context= shutdown_tx.subscribe(); let kill_context = shutdown_tx.subscribe();
// Listen for ctrl_c, send signal to broadcast shutdown to all threads by dropping shutdown_tx // Listen for ctrl_c, send signal to broadcast shutdown to all threads by dropping shutdown_tx
let shutdown_handler = spawn(async move { let shutdown_handler = spawn(async move {
tokio::select! { tokio::select! {
_ = signal::ctrl_c() => { _ = signal::ctrl_c() => {
let cmd = Command::Shutdown { let cmd = Command::Shutdown {
@ -600,15 +654,16 @@ async fn main() {
_ = shutdown_rx.recv().await; _ = shutdown_rx.recv().await;
} }
} }
}); });
let context_manager =
let context_manager = spawn(async move { create_context_manager(context_rx, kill_context).await }); spawn(async move { create_context_manager(context_rx, kill_context).await });
let (tx, rx) = mpsc::channel(32); let (tx, rx) = mpsc::channel(32);
let rdsys_stream_handler = spawn(async { rdsys_stream(rtype, tx, kill_stream).await }); let rdsys_stream_handler = spawn(async { rdsys_stream(rtype, tx, kill_stream).await });
let rdsys_resource_receiver = spawn(async { rdsys_bridge_parser(rdsys_tx, rx, kill_parser).await }); let rdsys_resource_receiver =
spawn(async { rdsys_bridge_parser(rdsys_tx, rx, kill_parser).await });
let make_service = make_service_fn(move |_conn: &AddrStream| { let make_service = make_service_fn(move |_conn: &AddrStream| {
let request_tx = request_tx.clone(); let request_tx = request_tx.clone();
@ -634,13 +689,11 @@ async fn main() {
if let Err(e) = graceful.await { if let Err(e) = graceful.await {
eprintln!("server error: {}", e); eprintln!("server error: {}", e);
} }
future::join_all([ future::join_all([
rdsys_stream_handler, rdsys_stream_handler,
rdsys_resource_receiver, rdsys_resource_receiver,
context_manager, context_manager,
shutdown_handler, shutdown_handler,
]) ])
.await; .await;
} }