Add simulation binary code, move server code to main

Cargo.toml

@@ -46,3 +46,7 @@ faketime = "0.2"
 
 [features]
 simulation = ["faketime", "lox_cli"]
+
+[[bin]]
+name = "simulation"
+required-features = ["simulation"]

config.json

@@ -10,7 +10,10 @@
     "confidence": 0.95,
     "max_threshold": 5,
     "scaling_factor": 0.25,
+    "min_historical_days": 30,
+    "max_historical_days": 30,
     "port": 8003,
     "require_bridge_token": false,
+    "updater_port": 8123,
     "updater_schedule": "* * 22 * * * *"
 }

src/bin/server.rs

@@ -1,270 +0,0 @@
-use troll_patrol::{request_handler::handle, *};
-
-use clap::Parser;
-use futures::future;
-use hyper::{
-    server::conn::AddrStream,
-    service::{make_service_fn, service_fn},
-    Body, Request, Response, Server,
-};
-use serde::Deserialize;
-use sled::Db;
-use std::{
-    collections::BTreeMap, convert::Infallible, fs::File, io::BufReader, net::SocketAddr,
-    path::PathBuf, time::Duration,
-};
-use tokio::{
-    signal, spawn,
-    sync::{broadcast, mpsc, oneshot},
-    time::sleep,
-};
-use tokio_cron::{Job, Scheduler};
-
-async fn shutdown_signal() {
-    tokio::signal::ctrl_c()
-        .await
-        .expect("failed to listen for ctrl+c signal");
-    println!("Shut down Troll Patrol Server");
-}
-
-#[derive(Parser, Debug)]
-#[command(author, version, about, long_about = None)]
-struct Args {
-    /// Name/path of the configuration file
-    #[arg(short, long, default_value = "config.json")]
-    config: PathBuf,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct Config {
-    pub db: DbConfig,
-    // map of distributor name to IP:port to contact it
-    pub distributors: BTreeMap<BridgeDistributor, String>,
-    extra_infos_base_url: String,
-
-    // confidence required to consider a bridge blocked
-    confidence: f64,
-
-    // block open-entry bridges if they get more negative reports than this
-    max_threshold: u32,
-
-    // block open-entry bridges if they get more negative reports than
-    // scaling_factor * bridge_ips
-    scaling_factor: f64,
-
-    // minimum number of historical days for statistical analysis
-    min_historical_days: u32,
-
-    // maximum number of historical days to consider in historical analysis
-    max_historical_days: u32,
-
-    //require_bridge_token: bool,
-    port: u16,
-    updater_schedule: String,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct DbConfig {
-    // The path for the server database, default is "server_db"
-    pub db_path: String,
-}
-
-impl Default for DbConfig {
-    fn default() -> DbConfig {
-        DbConfig {
-            db_path: "server_db".to_owned(),
-        }
-    }
-}
-
-async fn update_daily_info(
-    db: &Db,
-    distributors: &BTreeMap<BridgeDistributor, String>,
-    extra_infos_base_url: &str,
-    confidence: f64,
-    max_threshold: u32,
-    scaling_factor: f64,
-    min_historical_days: u32,
-    max_historical_days: u32,
-) {
-    update_extra_infos(&db, &extra_infos_base_url)
-        .await
-        .unwrap();
-    update_negative_reports(&db, &distributors).await;
-    update_positive_reports(&db, &distributors).await;
-    let new_blockages = guess_blockages(
-        &db,
-        &analysis::NormalAnalyzer::new(max_threshold, scaling_factor),
-        confidence,
-        min_historical_days,
-        max_historical_days,
-    );
-    report_blockages(&distributors, new_blockages).await;
-
-    // Generate tomorrow's key if we don't already have it
-    new_negative_report_key(&db, get_date() + 1);
-}
-
-async fn run_updater(updater_tx: mpsc::Sender<Command>) {
-    updater_tx.send(Command::Update {}).await.unwrap();
-}
-
-async fn create_context_manager(
-    db_config: DbConfig,
-    distributors: BTreeMap<BridgeDistributor, String>,
-    extra_infos_base_url: &str,
-    confidence: f64,
-    max_threshold: u32,
-    scaling_factor: f64,
-    min_historical_days: u32,
-    max_historical_days: u32,
-    context_rx: mpsc::Receiver<Command>,
-    mut kill: broadcast::Receiver<()>,
-) {
-    tokio::select! {
-        create_context = context_manager(db_config, distributors, extra_infos_base_url, confidence, max_threshold, scaling_factor, min_historical_days, max_historical_days, context_rx) => create_context,
-        _ = kill.recv() => {println!("Shut down manager");},
-    }
-}
-
-async fn context_manager(
-    db_config: DbConfig,
-    distributors: BTreeMap<BridgeDistributor, String>,
-    extra_infos_base_url: &str,
-    confidence: f64,
-    max_threshold: u32,
-    scaling_factor: f64,
-    min_historical_days: u32,
-    max_historical_days: u32,
-    mut context_rx: mpsc::Receiver<Command>,
-) {
-    let db: Db = sled::open(&db_config.db_path).unwrap();
-
-    while let Some(cmd) = context_rx.recv().await {
-        use Command::*;
-        match cmd {
-            Request { req, sender } => {
-                let response = handle(&db, 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.");
-                drop(shutdown_sig);
-                println!("Shutdown Sent.");
-            }
-            Update {} => {
-                update_daily_info(
-                    &db,
-                    &distributors,
-                    &extra_infos_base_url,
-                    confidence,
-                    max_threshold,
-                    scaling_factor,
-                    min_historical_days,
-                    max_historical_days,
-                )
-                .await;
-            }
-        }
-    }
-}
-
-// Each of the commands that can be handled
-#[derive(Debug)]
-enum Command {
-    Request {
-        req: Request<Body>,
-        sender: oneshot::Sender<Result<Response<Body>, Infallible>>,
-    },
-    Shutdown {
-        shutdown_sig: broadcast::Sender<()>,
-    },
-    Update {},
-}
-
-#[tokio::main]
-async fn main() {
-    let args: Args = Args::parse();
-
-    let config: Config = serde_json::from_reader(BufReader::new(
-        File::open(&args.config).expect("Could not read config file"),
-    ))
-    .expect("Reading config file from JSON failed");
-
-    let (request_tx, request_rx) = mpsc::channel(32);
-
-    let updater_tx = request_tx.clone();
-    let shutdown_cmd_tx = request_tx.clone();
-
-    // create the shutdown broadcast channel
-    let (shutdown_tx, mut shutdown_rx) = broadcast::channel(16);
-    let kill = shutdown_tx.subscribe();
-
-    // Listen for ctrl_c, send signal to broadcast shutdown to all threads by dropping shutdown_tx
-    let shutdown_handler = spawn(async move {
-        tokio::select! {
-            _ = signal::ctrl_c() => {
-                let cmd = Command::Shutdown {
-                    shutdown_sig: shutdown_tx,
-                };
-                shutdown_cmd_tx.send(cmd).await.unwrap();
-                sleep(Duration::from_secs(1)).await;
-
-                _ = shutdown_rx.recv().await;
-            }
-        }
-    });
-
-    let updater = spawn(async move {
-        // Run updater once per day
-        let mut sched = Scheduler::utc();
-        sched.add(Job::new(config.updater_schedule, move || {
-            run_updater(updater_tx.clone())
-        }));
-    });
-
-    let context_manager = spawn(async move {
-        create_context_manager(
-            config.db,
-            config.distributors,
-            &config.extra_infos_base_url,
-            config.confidence,
-            config.max_threshold,
-            config.scaling_factor,
-            config.min_historical_days,
-            config.max_historical_days,
-            request_rx,
-            kill,
-        )
-        .await
-    });
-
-    let make_service = make_service_fn(move |_conn: &AddrStream| {
-        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,
-                sender: response_tx,
-            };
-            async move {
-                request_tx.send(cmd).await.unwrap();
-                response_rx.await.unwrap()
-            }
-        });
-        async move { Ok::<_, Infallible>(service) }
-    });
-
-    let addr = SocketAddr::from(([0, 0, 0, 0], config.port));
-    let server = Server::bind(&addr).serve(make_service);
-    let graceful = server.with_graceful_shutdown(shutdown_signal());
-    println!("Listening on {}", addr);
-    if let Err(e) = graceful.await {
-        eprintln!("server error: {}", e);
-    }
-    future::join_all([context_manager, updater, shutdown_handler]).await;
-}

src/bin/simulation.rs

@@ -0,0 +1,178 @@
+// Before running this, run:
+// 1. rdsys
+// 2. lox-distributor
+// 3. troll-patrol with the feature "simulation"
+
+use troll_patrol::{
+    extra_info::ExtraInfo,
+    increment_simulated_date,
+    simulation::{bridge::Bridge, censor::Censor, extra_infos_server, state::State, user::User},
+};
+
+use clap::Parser;
+use lox_cli::{networking::*, *};
+use rand::Rng;
+use serde::Deserialize;
+use std::{
+    collections::{HashMap, HashSet},
+    fs::File,
+    io::BufReader,
+    path::PathBuf,
+};
+use tokio::spawn;
+
+#[derive(Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+struct Args {
+    /// Name/path of the configuration file
+    #[arg(short, long, default_value = "simulation_config.json")]
+    config: PathBuf,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct Config {
+    pub la_port: u16,
+    pub la_test_port: u16,
+    pub tp_port: u16,
+    pub tp_test_port: u16,
+    pub min_new_users_per_day: u32,
+    pub max_new_users_per_day: u32,
+    // How many days to simulate
+    pub num_days: u32,
+    pub prob_connection_fails: f64,
+    pub prob_friend_in_same_country: f64,
+    pub prob_user_invites_friend: f64,
+    pub prob_user_is_censor: f64,
+    pub prob_user_submits_reports: f64,
+    pub probs_user_in_country: Vec<(String, f64)>,
+    pub sharing: bool,
+}
+
+#[tokio::main]
+pub async fn main() {
+    let args: Args = Args::parse();
+
+    let config: Config = serde_json::from_reader(BufReader::new(
+        File::open(&args.config).expect("Could not read config file"),
+    ))
+    .expect("Reading config file from JSON failed");
+
+    let la_net = HyperNet {
+        hostname: format!("http://localhost:{}", config.la_port),
+    };
+    let la_net_test = HyperNet {
+        hostname: format!("http://localhost:{}", config.la_test_port),
+    };
+    let tp_net = HyperNet {
+        hostname: format!("http://localhost:{}", config.tp_port),
+    };
+    let tp_net_test = HyperNet {
+        hostname: format!("http://localhost:{}", config.tp_test_port),
+    };
+    let extra_infos_net = HyperNet {
+        hostname: "http://localhost:8003".to_string(),
+    };
+
+    let la_pubkeys = get_lox_auth_keys(&la_net).await;
+
+    let state = State {
+        la_net,
+        tp_net,
+        la_pubkeys,
+        prob_connection_fails: config.prob_connection_fails,
+        prob_friend_in_same_country: config.prob_friend_in_same_country,
+        prob_user_invites_friend: config.prob_user_invites_friend,
+        prob_user_is_censor: config.prob_user_is_censor,
+        prob_user_submits_reports: config.prob_user_submits_reports,
+        probs_user_in_country: config.probs_user_in_country,
+        sharing: config.sharing,
+    };
+
+    let mut rng = rand::thread_rng();
+
+    // Set up censors
+    let mut censors = HashMap::<String, Censor>::new();
+
+    // Set up bridges (no bridges yet)
+    let mut bridges = HashMap::<[u8; 20], Bridge>::new();
+
+    // Set up users (no users yet)
+    let mut users = Vec::<User>::new();
+
+    // Set up extra-infos server
+    spawn(async move {
+        extra_infos_server::server().await;
+    });
+
+    // Main loop
+    for _ in 0..config.num_days {
+        // USER TASKS
+
+        // Add some new users
+        let num_new_users: u32 =
+            rng.gen_range(config.min_new_users_per_day..=config.max_new_users_per_day);
+        for _ in 0..num_new_users {
+            users.push(User::new(&state).await);
+        }
+
+        let mut new_users = Vec::<User>::new();
+
+        // Users do daily actions
+        for user in &mut users {
+            // TODO: Refactor out connections from return
+            let (mut invited_friends, _connections) =
+                user.daily_tasks(&state, &mut bridges, &mut censors).await;
+
+            // If this user invited any friends, add them to the list of users
+            new_users.append(&mut invited_friends);
+        }
+
+        // Add new users
+        users.append(&mut new_users);
+
+        // CENSOR TASKS
+        for (_, censor) in censors.iter_mut() {
+            censor.end_of_day_tasks(&state, &mut bridges).await;
+        }
+
+        // BRIDGE TASKS
+        let mut new_extra_infos = HashSet::<ExtraInfo>::new();
+        for (_, bridge) in bridges.iter_mut() {
+            // Bridge reports its connections for the day
+            new_extra_infos.insert(bridge.gen_extra_info());
+
+            // Bridge resets for tomorrow
+            bridge.reset_for_tomorrow();
+        }
+
+        // Publish all the bridges' extra-infos for today
+        extra_infos_net
+            .request(
+                "/add".to_string(),
+                serde_json::to_string(&new_extra_infos).unwrap().into(),
+            )
+            .await;
+
+        // TROLL PATROL TASKS
+        let new_blockages_resp = tp_net_test.request("/update".to_string(), vec![]).await;
+        let new_blockages: HashMap<String, HashSet<String>> =
+            serde_json::from_slice(&new_blockages_resp).unwrap();
+
+        // TODO: Track stats about new blockages
+
+        // LOX AUTHORITY TASKS
+
+        // Advance LA's time to tomorrow
+        la_net_test
+            .request(
+                "/advancedays".to_string(),
+                serde_json::to_string(&(1 as u16)).unwrap().into(),
+            )
+            .await;
+
+        // SIMULATION TASKS
+
+        // Advance simulated time to tomorrow
+        increment_simulated_date();
+    }
+}

src/lib.rs

@@ -65,6 +65,7 @@ pub fn get_date() -> u32 {
     return get_simulated_date();
 
     // If we're not running a simulation, return today's date
+    #[allow(unreachable_code)]
     get_real_date()
 }
 

src/main.rs

@@ -1,3 +1,333 @@
-fn main() {
-    println!("Hello, world!");
+use troll_patrol::{request_handler::*, *};
+
+use clap::Parser;
+use futures::future;
+use futures::join;
+use hyper::{
+    server::conn::AddrStream,
+    service::{make_service_fn, service_fn},
+    Body, Request, Response, Server,
+};
+use serde::Deserialize;
+use sled::Db;
+use std::{
+    collections::{BTreeMap, HashMap, HashSet},
+    convert::Infallible,
+    fs::File,
+    io::BufReader,
+    net::SocketAddr,
+    path::PathBuf,
+    time::Duration,
+};
+use tokio::{
+    signal, spawn,
+    sync::{broadcast, mpsc, oneshot},
+    time::sleep,
+};
+#[cfg(not(features = "simulation"))]
+use tokio_cron::{Job, Scheduler};
+
+async fn shutdown_signal() {
+    tokio::signal::ctrl_c()
+        .await
+        .expect("failed to listen for ctrl+c signal");
+    println!("Shut down Troll Patrol Server");
+}
+
+#[derive(Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+struct Args {
+    /// Name/path of the configuration file
+    #[arg(short, long, default_value = "config.json")]
+    config: PathBuf,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct Config {
+    pub db: DbConfig,
+    // map of distributor name to IP:port to contact it
+    pub distributors: BTreeMap<BridgeDistributor, String>,
+    extra_infos_base_url: String,
+
+    // confidence required to consider a bridge blocked
+    confidence: f64,
+
+    // block open-entry bridges if they get more negative reports than this
+    max_threshold: u32,
+
+    // block open-entry bridges if they get more negative reports than
+    // scaling_factor * bridge_ips
+    scaling_factor: f64,
+
+    // minimum number of historical days for statistical analysis
+    min_historical_days: u32,
+
+    // maximum number of historical days to consider in historical analysis
+    max_historical_days: u32,
+
+    //require_bridge_token: bool,
+    port: u16,
+    updater_port: u16,
+    updater_schedule: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct DbConfig {
+    // The path for the server database, default is "server_db"
+    pub db_path: String,
+}
+
+impl Default for DbConfig {
+    fn default() -> DbConfig {
+        DbConfig {
+            db_path: "server_db".to_owned(),
+        }
+    }
+}
+
+async fn update_daily_info(
+    db: &Db,
+    distributors: &BTreeMap<BridgeDistributor, String>,
+    extra_infos_base_url: &str,
+    confidence: f64,
+    max_threshold: u32,
+    scaling_factor: f64,
+    min_historical_days: u32,
+    max_historical_days: u32,
+) -> HashMap<[u8; 20], HashSet<String>> {
+    update_extra_infos(&db, &extra_infos_base_url)
+        .await
+        .unwrap();
+    update_negative_reports(&db, &distributors).await;
+    update_positive_reports(&db, &distributors).await;
+    let new_blockages = guess_blockages(
+        &db,
+        &analysis::NormalAnalyzer::new(max_threshold, scaling_factor),
+        confidence,
+        min_historical_days,
+        max_historical_days,
+    );
+    report_blockages(&distributors, new_blockages.clone()).await;
+
+    // Generate tomorrow's key if we don't already have it
+    new_negative_report_key(&db, get_date() + 1);
+
+    // Return new detected blockages
+    new_blockages
+}
+
+/*
+async fn run_updater(updater_tx: mpsc::Sender<Command>) {
+    updater_tx.send(Command::Update {
+
+    }).await.unwrap();
+}
+*/
+
+async fn create_context_manager(
+    db_config: DbConfig,
+    distributors: BTreeMap<BridgeDistributor, String>,
+    extra_infos_base_url: &str,
+    confidence: f64,
+    max_threshold: u32,
+    scaling_factor: f64,
+    min_historical_days: u32,
+    max_historical_days: u32,
+    context_rx: mpsc::Receiver<Command>,
+    mut kill: broadcast::Receiver<()>,
+) {
+    tokio::select! {
+        create_context = context_manager(db_config, distributors, extra_infos_base_url, confidence, max_threshold, scaling_factor, min_historical_days, max_historical_days, context_rx) => create_context,
+        _ = kill.recv() => {println!("Shut down manager");},
+    }
+}
+
+async fn context_manager(
+    db_config: DbConfig,
+    distributors: BTreeMap<BridgeDistributor, String>,
+    extra_infos_base_url: &str,
+    confidence: f64,
+    max_threshold: u32,
+    scaling_factor: f64,
+    min_historical_days: u32,
+    max_historical_days: u32,
+    mut context_rx: mpsc::Receiver<Command>,
+) {
+    let db: Db = sled::open(&db_config.db_path).unwrap();
+
+    while let Some(cmd) = context_rx.recv().await {
+        use Command::*;
+        match cmd {
+            Request { req, sender } => {
+                let response = handle(&db, 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.");
+                drop(shutdown_sig);
+                println!("Shutdown Sent.");
+            }
+            Update { _req, sender } => {
+                let blockages = update_daily_info(
+                    &db,
+                    &distributors,
+                    &extra_infos_base_url,
+                    confidence,
+                    max_threshold,
+                    scaling_factor,
+                    min_historical_days,
+                    max_historical_days,
+                )
+                .await;
+                let response = if cfg!(feature = "simulation") {
+                    // Convert map keys from [u8; 20] to 40-character hex strings
+                    let mut blockages_str = HashMap::<String, HashSet<String>>::new();
+                    for (fingerprint, countries) in blockages {
+                        let fpr_string = array_bytes::bytes2hex("", fingerprint);
+                        blockages_str.insert(fpr_string, countries);
+                    }
+                    Ok(prepare_header(
+                        serde_json::to_string(&blockages_str).unwrap(),
+                    ))
+                } else {
+                    Ok(prepare_header("OK".to_string()))
+                };
+                if let Err(e) = sender.send(response) {
+                    eprintln!("Update Response Error: {:?}", e);
+                };
+                sleep(Duration::from_millis(1)).await;
+            }
+        }
+    }
+}
+
+// Each of the commands that can be handled
+#[derive(Debug)]
+enum Command {
+    Request {
+        req: Request<Body>,
+        sender: oneshot::Sender<Result<Response<Body>, Infallible>>,
+    },
+    Shutdown {
+        shutdown_sig: broadcast::Sender<()>,
+    },
+    Update {
+        _req: Request<Body>,
+        sender: oneshot::Sender<Result<Response<Body>, Infallible>>,
+    },
+}
+
+#[tokio::main]
+async fn main() {
+    let args: Args = Args::parse();
+
+    let config: Config = serde_json::from_reader(BufReader::new(
+        File::open(&args.config).expect("Could not read config file"),
+    ))
+    .expect("Reading config file from JSON failed");
+
+    let (request_tx, request_rx) = mpsc::channel(32);
+
+    let updater_tx = request_tx.clone();
+    let shutdown_cmd_tx = request_tx.clone();
+
+    // create the shutdown broadcast channel
+    let (shutdown_tx, mut shutdown_rx) = broadcast::channel(16);
+    let kill = shutdown_tx.subscribe();
+
+    // Listen for ctrl_c, send signal to broadcast shutdown to all threads by dropping shutdown_tx
+    let shutdown_handler = spawn(async move {
+        tokio::select! {
+            _ = signal::ctrl_c() => {
+                let cmd = Command::Shutdown {
+                    shutdown_sig: shutdown_tx,
+                };
+                shutdown_cmd_tx.send(cmd).await.unwrap();
+                sleep(Duration::from_secs(1)).await;
+
+                _ = shutdown_rx.recv().await;
+            }
+        }
+    });
+
+    // TODO: Reintroduce this
+    /*
+        #[cfg(not(feature = "simulation"))]
+        let updater = spawn(async move {
+            // Run updater once per day
+            let mut sched = Scheduler::utc();
+            sched.add(Job::new(config.updater_schedule, move || {
+                run_updater(updater_tx.clone())
+            }));
+        });
+    */
+    let context_manager = spawn(async move {
+        create_context_manager(
+            config.db,
+            config.distributors,
+            &config.extra_infos_base_url,
+            config.confidence,
+            config.max_threshold,
+            config.scaling_factor,
+            config.min_historical_days,
+            config.max_historical_days,
+            request_rx,
+            kill,
+        )
+        .await
+    });
+
+    let make_service = make_service_fn(move |_conn: &AddrStream| {
+        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,
+                sender: response_tx,
+            };
+            async move {
+                request_tx.send(cmd).await.unwrap();
+                response_rx.await.unwrap()
+            }
+        });
+        async move { Ok::<_, Infallible>(service) }
+    });
+
+    let updater_make_service = make_service_fn(move |_conn: &AddrStream| {
+        let request_tx = updater_tx.clone();
+        let service = service_fn(move |_req| {
+            let request_tx = request_tx.clone();
+            let (response_tx, response_rx) = oneshot::channel();
+            let cmd = Command::Update {
+                _req,
+                sender: response_tx,
+            };
+            async move {
+                request_tx.send(cmd).await.unwrap();
+                response_rx.await.unwrap()
+            }
+        });
+        async move { Ok::<_, Infallible>(service) }
+    });
+
+    let addr = SocketAddr::from(([0, 0, 0, 0], config.port));
+    let server = Server::bind(&addr).serve(make_service);
+    let graceful = server.with_graceful_shutdown(shutdown_signal());
+    let updater_addr = SocketAddr::from(([127, 0, 0, 1], config.updater_port));
+    let updater_server = Server::bind(&updater_addr).serve(updater_make_service);
+    let updater_graceful = updater_server.with_graceful_shutdown(shutdown_signal());
+    println!("Listening on {}", addr);
+    println!("Updater listening on {}", updater_addr);
+    let (a, b) = join!(graceful, updater_graceful);
+    if a.is_err() {
+        eprintln!("server error: {}", a.unwrap_err());
+    }
+    if b.is_err() {
+        eprintln!("server error: {}", b.unwrap_err());
+    }
+    future::join_all([context_manager, shutdown_handler]).await;
 }

src/request_handler.rs

@@ -76,7 +76,7 @@ pub async fn handle(db: &Db, req: Request<Body>) -> Result<Response<Body>, Infal
 }
 
 // Prepare HTTP Response for successful Server Request
-fn prepare_header(response: String) -> Response<Body> {
+pub fn prepare_header(response: String) -> Response<Body> {
     let mut resp = Response::new(Body::from(response));
     resp.headers_mut()
         .insert("Access-Control-Allow-Origin", HeaderValue::from_static("*"));

src/simulation/bridge.rs

@@ -1,10 +1,12 @@
+use crate::{extra_info::ExtraInfo, get_date};
 use lox_library::bridge_table::BridgeLine;
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 
+// The Bridge struct only tracks data for today
 pub struct Bridge {
     pub fingerprint: [u8; 20],
     real_connections: HashMap<String, u32>,
-    total_connections: HashMap<String, u32>,
+    total_connections: BTreeMap<String, u32>,
 }
 
 impl Bridge {
@@ -12,7 +14,7 @@ impl Bridge {
         Self {
             fingerprint: *fingerprint,
             real_connections: HashMap::<String, u32>::new(),
-            total_connections: HashMap::<String, u32>::new(),
+            total_connections: BTreeMap::<String, u32>::new(),
         }
     }
 
@@ -46,4 +48,33 @@ impl Bridge {
                 .unwrap();
         }
     }
+
+    // Let the censor simulate a bunch of connections at once
+    pub fn censor_flood(&mut self, country: &str, num_connections: u32) {
+        if self.total_connections.contains_key(country) {
+            let prev = self.total_connections.get(country).unwrap();
+            self.total_connections
+                .insert(country.to_string(), prev + num_connections)
+                .unwrap();
+        } else {
+            self.total_connections
+                .insert(country.to_string(), num_connections)
+                .unwrap();
+        }
+    }
+
+    // Generate an extra-info report for today
+    pub fn gen_extra_info(&self) -> ExtraInfo {
+        ExtraInfo {
+            nickname: String::default(),
+            fingerprint: self.fingerprint,
+            date: get_date(),
+            bridge_ips: self.total_connections.clone(),
+        }
+    }
+
+    pub fn reset_for_tomorrow(&mut self) {
+        self.real_connections = HashMap::<String, u32>::new();
+        self.total_connections = BTreeMap::<String, u32>::new();
+    }
 }

src/simulation/censor.rs

@@ -1,5 +1,10 @@
-use crate::{get_date, simulation::bridge::Bridge};
+use crate::{
+    get_date,
+    simulation::{bridge::Bridge, state::State},
+    PositiveReport,
+};
 
+use lox_cli::{get_lox_pub, networking::Networking};
 use lox_library::{cred::Lox, scalar_u32};
 use rand::Rng;
 use std::collections::{HashMap, HashSet};
@@ -53,28 +58,99 @@ impl Censor {
         }
     }
 
-    pub fn knows_bridge(&self, bridge: &Bridge) -> bool {
-        self.known_bridges.contains(&bridge.fingerprint)
+    pub fn knows_bridge(&self, fingerprint: &[u8; 20]) -> bool {
+        self.known_bridges.contains(fingerprint)
     }
 
     pub fn learn_bridge(&mut self, fingerprint: &[u8; 20]) {
         self.known_bridges.insert(*fingerprint);
     }
 
-    pub fn has_lox_cred(&self, bridge: &Bridge) -> bool {
-        self.lox_credentials.contains_key(&bridge.fingerprint)
+    pub fn has_lox_cred(&self, fingerprint: &[u8; 20]) -> bool {
+        self.lox_credentials.contains_key(fingerprint)
     }
 
-    pub fn give_lox_cred(&mut self, bridge: &Bridge, cred: Lox) {
+    pub fn give_lox_cred(&mut self, fingerprint: &[u8; 20], cred: &Lox) {
         // We only need one level 3+ credential per bridge. (This will
         // change if we restrict positive reports to one per bridge per
         // credential.)
-        if !self.lox_credentials.contains_key(&bridge.fingerprint)
-            && scalar_u32(&cred.trust_level).unwrap() >= 3
-        {
-            self.lox_credentials.insert(bridge.fingerprint, cred);
+        if !self.has_lox_cred(fingerprint) && scalar_u32(&cred.trust_level).unwrap() >= 3 {
+            // We want to clone the credential, but that's not allowed,
+            // so we're going to serialize it and then deserialize it.
+            let cloned_cred = bincode::deserialize(&bincode::serialize(&cred).unwrap()).unwrap();
+            self.lox_credentials.insert(*fingerprint, cloned_cred);
         }
     }
+
+    // Make a bunch of connections and submit positive reports if possible
+    async fn flood(&self, state: &State, bridges: &mut HashMap<[u8; 20], Bridge>) {
+        // Only do this if Flooding censor
+        if self.hides == Hides::Flooding {
+            for fingerprint in &self.known_bridges {
+                // Only do this if we're blocking the bridge
+                if self.speed == Speed::Fast
+                    || self.speed == Speed::Lox && self.has_lox_cred(fingerprint)
+                    || self.speed == Speed::Random && self.delay_date <= get_date()
+                {
+                    let bridge = bridges.get_mut(fingerprint).unwrap();
+                    let mut rng = rand::thread_rng();
+                    let num_connections = rng.gen_range(1000..30000);
+
+                    // Make a bunch of connections to the bridge
+                    bridge.censor_flood(&self.country, num_connections);
+
+                    // If we have a lv3+ credential, submit a bunch of
+                    // positive reports
+                    if self.has_lox_cred(fingerprint) {
+                        let lox_pub = get_lox_pub(&state.la_pubkeys);
+                        for _ in 0..num_connections {
+                            let pr = PositiveReport::from_lox_credential(
+                                bridge.fingerprint,
+                                None,
+                                &self.lox_credentials.get(&bridge.fingerprint).unwrap(),
+                                lox_pub,
+                                self.country.clone(),
+                            )
+                            .unwrap();
+                            state
+                                .tp_net
+                                .request("/positivereport".to_string(), pr.to_json().into_bytes())
+                                .await;
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    // TODO: How do we want to do this? We don't want to stop blocking
+    // bridges the day after we start.
+    fn recompute_delay(&mut self) {
+        // Only do this if Random censor
+        if self.speed == Speed::Random && self.delay_date <= get_date() {
+            // Compute new delay date
+            self.delay_date = {
+                let mut rng = rand::thread_rng();
+                let num: u32 = rng.gen_range(1..365);
+                get_date() + num
+            }
+        }
+    }
+
+    pub async fn end_of_day_tasks(
+        &mut self,
+        state: &State,
+        bridges: &mut HashMap<[u8; 20], Bridge>,
+    ) {
+        if self.hides == Hides::Flooding
+            && !(self.speed == Speed::Random && self.delay_date <= get_date())
+        {
+            self.flood(state, bridges).await;
+        }
+
+        // TODO: recompute_delay sometimes
+        //self.recompute_delay();
+    }
 }
 
 #[derive(PartialEq)]

src/simulation/state.rs

@@ -3,9 +3,8 @@ use lox_library::IssuerPubKey;
 
 pub struct State {
     pub la_pubkeys: Vec<IssuerPubKey>,
-    pub net: HyperNet,
-    pub net_test: HyperNet,
-    pub net_tp: HyperNet,
+    pub la_net: HyperNet,
+    pub tp_net: HyperNet,
     // Probability that a connection randomly fails, even though censor
     // does not block the bridge
     pub prob_connection_fails: f64,
@@ -17,4 +16,6 @@ pub struct State {
     pub prob_user_is_censor: f64,
     pub prob_user_submits_reports: f64,
     pub probs_user_in_country: Vec<(String, f64)>,
+    // Do the censors talk to each other?
+    pub sharing: bool,
 }

src/simulation/user.rs

@@ -49,8 +49,8 @@ pub struct User {
 impl User {
     pub async fn new(state: &State) -> Self {
         let cred = get_lox_credential(
-            &state.net,
-            &get_open_invitation(&state.net).await,
+            &state.la_net,
+            &get_open_invitation(&state.la_net).await,
             get_lox_pub(&state.la_pubkeys),
         )
         .await
@@ -95,9 +95,9 @@ impl User {
 
     // TODO: This should probably return an actual error type
     pub async fn invite(&mut self, state: &State) -> Result<Self, String> {
-        let etable = get_reachability_credential(&state.net).await;
+        let etable = get_reachability_credential(&state.la_net).await;
         let (new_cred, invite) = issue_invite(
-            &state.net,
+            &state.la_net,
             &self.primary_cred,
             &etable,
             get_lox_pub(&state.la_pubkeys),
@@ -107,7 +107,7 @@ impl User {
         .await;
         self.primary_cred = new_cred;
         let friend_cred = redeem_invite(
-            &state.net,
+            &state.la_net,
             &invite,
             get_lox_pub(&state.la_pubkeys),
             get_invitation_pub(&state.la_pubkeys),
@@ -162,10 +162,10 @@ impl User {
 
     // Attempt to "connect" to the bridge, returns true if successful
     pub fn connect(&self, state: &State, bridge: &mut Bridge, censor: &Censor) -> bool {
-        if censor.knows_bridge(bridge) {
+        if censor.knows_bridge(&bridge.fingerprint) {
             if censor.speed == Fast
                 || censor.speed == Random && censor.delay_date <= get_date()
-                || censor.speed == Lox && censor.has_lox_cred(bridge)
+                || censor.speed == Lox && censor.has_lox_cred(&bridge.fingerprint)
             {
                 if censor.totality == Full
                     || censor.totality == Partial && event_happens(censor.partial_blocking_percent)
@@ -196,10 +196,9 @@ impl User {
 
     pub async fn send_negative_reports(state: &State, reports: Vec<NegativeReport>) {
         let date = get_date();
-        //let pubkey = state.tp_pubkeys.get(&date).unwrap();
         let pubkey = serde_json::from_slice::<Option<PublicKey>>(
             &state
-                .net_tp
+                .tp_net
                 .request(
                     "/nrkey".to_string(),
                     serde_json::to_string(&date).unwrap().into(),
@@ -210,7 +209,7 @@ impl User {
         .unwrap();
         for report in reports {
             state
-                .net_tp
+                .tp_net
                 .request(
                     "/negativereport".to_string(),
                     bincode::serialize(&report.encrypt(&pubkey)).unwrap(),
@@ -222,7 +221,7 @@ impl User {
     pub async fn send_positive_reports(state: &State, reports: Vec<PositiveReport>) {
         for report in reports {
             state
-                .net_tp
+                .tp_net
                 .request("/positivereport".to_string(), report.to_json().into_bytes())
                 .await;
         }
@@ -239,30 +238,43 @@ impl User {
         bridges: &mut HashMap<[u8; 20], Bridge>,
         censors: &mut HashMap<String, Censor>,
     ) -> (Vec<User>, Vec<[u8; 20]>) {
-        let censor = censors.get(&self.country).unwrap();
-
         // Probabilistically decide if the user should use bridges today
         if event_happens(self.prob_use_bridges) {
-            // Download bucket to see if bridge is still reachable
-            // (We assume that this step can be done even if the user can't actually
-            // talk to the LA.)
-            let (bucket, reachcred) = get_bucket(&state.net, &self.primary_cred).await;
+            // Download bucket to see if bridge is still reachable. (We
+            // assume that this step can be done even if the user can't
+            // actually talk to the LA.)
+            let (bucket, reachcred) = get_bucket(&state.la_net, &self.primary_cred).await;
             let level = scalar_u32(&self.primary_cred.trust_level).unwrap();
 
             // Make sure each bridge in bucket is in the global bridges set
-            for bridge in bucket {
-                if !bridges.contains_key(&bridge.fingerprint) {
-                    bridges
-                        .insert(bridge.fingerprint, Bridge::from_bridge_line(&bridge))
-                        .unwrap();
+            for bridgeline in bucket {
+                if !bridges.contains_key(&bridgeline.fingerprint) {
+                    let bridge = Bridge::from_bridge_line(&bridgeline);
+                    bridges.insert(bridgeline.fingerprint, bridge).unwrap();
+                }
+                // Also, if this user cooperates with censors, make sure
+                // each applicable censor knows about their bridges.
+                if self.censor {
+                    if state.sharing {
+                        for c in censors.values_mut() {
+                            if !c.knows_bridge(&bridgeline.fingerprint) {
+                                c.learn_bridge(&bridgeline.fingerprint);
+                            }
+                        }
+                    } else {
+                        let censor = censors.get_mut(&self.country).unwrap();
+                        if !censor.knows_bridge(&bridgeline.fingerprint) {
+                            censor.learn_bridge(&bridgeline.fingerprint);
+                        }
+                    }
                 }
             }
 
             // Can we level up the main credential?
             let can_level_up = reachcred.is_some()
                 && (level == 0
-                    && eligible_for_trust_promotion(&state.net, &self.primary_cred).await
-                    || level > 0 && eligible_for_level_up(&state.net, &self.primary_cred).await);
+                    && eligible_for_trust_promotion(&state.la_net, &self.primary_cred).await
+                    || level > 0 && eligible_for_level_up(&state.la_net, &self.primary_cred).await);
 
             // Can we migrate the main credential?
             let can_migrate = reachcred.is_none() && level >= MIN_TRUST_LEVEL;
@@ -278,7 +290,7 @@ impl User {
                     if self.connect(
                         &state,
                         bridges.get_mut(&bucket[i].fingerprint).unwrap(),
-                        &censor,
+                        &censors.get(&self.country).unwrap(),
                     ) {
                         succeeded.push(bucket[i]);
                     } else {
@@ -292,8 +304,8 @@ impl User {
                 } else {
                     // Get new credential
                     let cred = get_lox_credential(
-                        &state.net,
-                        &get_open_invitation(&state.net).await,
+                        &state.la_net,
+                        &get_open_invitation(&state.la_net).await,
                         get_lox_pub(&state.la_pubkeys),
                     )
                     .await
@@ -307,7 +319,8 @@ impl User {
             };
             if second_cred.is_some() {
                 let second_cred = second_cred.as_ref().unwrap();
-                let (second_bucket, second_reachcred) = get_bucket(&state.net, &second_cred).await;
+                let (second_bucket, second_reachcred) =
+                    get_bucket(&state.la_net, &second_cred).await;
                 if !bridges.contains_key(&second_bucket[0].fingerprint) {
                     bridges
                         .insert(
@@ -319,11 +332,11 @@ impl User {
                 if self.connect(
                     &state,
                     bridges.get_mut(&second_bucket[0].fingerprint).unwrap(),
-                    &censor,
+                    &censors.get(&self.country).unwrap(),
                 ) {
                     succeeded.push(second_bucket[0]);
                     if second_reachcred.is_some()
-                        && eligible_for_trust_promotion(&state.net, &second_cred).await
+                        && eligible_for_trust_promotion(&state.la_net, &second_cred).await
                     {
                         second_level_up = true;
                     }
@@ -358,11 +371,12 @@ impl User {
                 }
             }
 
-            // We might restrict these steps to succeeded.len() > 0, but we do
-            // assume the user can contact the LA somehow, so let's just allow it.
+            // We might restrict these steps to succeeded.len() > 0, but
+            // we do assume the user can contact the LA somehow, so
+            // let's just allow it.
             if can_level_up {
                 let cred = level_up(
-                    &state.net,
+                    &state.la_net,
                     &self.primary_cred,
                     &reachcred.unwrap(),
                     get_lox_pub(&state.la_pubkeys),
@@ -371,14 +385,25 @@ impl User {
                 .await;
                 self.primary_cred = cred;
                 self.secondary_cred = None;
+
+                if self.censor {
+                    // Make sure censor has access to each bridge and
+                    // each credential
+                    let censor = censors.get_mut(&self.country).unwrap();
+                    let (bucket, reachcred) = get_bucket(&state.la_net, &self.primary_cred).await;
+                    for bl in bucket {
+                        censor.learn_bridge(&bl.fingerprint);
+                        censor.give_lox_cred(&bl.fingerprint, &self.primary_cred);
+                    }
+                }
             }
             // We favor starting over at level 1 to migrating
             else if second_level_up {
                 let second_cred = second_cred.as_ref().unwrap();
                 let cred = trust_migration(
-                    &state.net,
+                    &state.la_net,
                     &second_cred,
-                    &trust_promotion(&state.net, &second_cred, get_lox_pub(&state.la_pubkeys))
+                    &trust_promotion(&state.la_net, &second_cred, get_lox_pub(&state.la_pubkeys))
                         .await,
                     get_lox_pub(&state.la_pubkeys),
                     get_migration_pub(&state.la_pubkeys),
@@ -388,10 +413,10 @@ impl User {
                 self.secondary_cred = None;
             } else if can_migrate {
                 let cred = blockage_migration(
-                    &state.net,
+                    &state.la_net,
                     &self.primary_cred,
                     &check_blockage(
-                        &state.net,
+                        &state.la_net,
                         &self.primary_cred,
                         get_lox_pub(&state.la_pubkeys),
                     )
@@ -424,8 +449,9 @@ impl User {
                 if event_happens(state.prob_user_invites_friend) {
                     match self.invite(&state).await {
                         Ok(friend) => {
-                            // You really shouldn't push your friends, especially
-                            // new ones whose boundaries you might not know well.
+                            // You really shouldn't push your friends,
+                            // especially new ones whose boundaries you
+                            // might not know well.
                             new_friends.push(friend);
                         }
                         Err(e) => {
@@ -435,8 +461,8 @@ impl User {
                 }
             }
 
-            // List of fingerprints we contacted. This should not actually be more
-            // than one.
+            // List of fingerprints we contacted. This should not
+            // actually be more than one.
             let mut connections = Vec::<[u8; 20]>::new();
             for bridge in succeeded {
                 connections.push(bridge.get_hashed_fingerprint());