add support for a peer hosts file

README.md

@@ -36,7 +36,7 @@ If none is given, it will listen on `127.0.0.1:6397`.
 #### Client/Peer
 `mgen-client [config.yaml]...`
 
-`mgen-peer [config.yaml]...`
+`mgen-peer [hosts file] [config.yaml]...`
 
 The client and peer configuration files are detailed below.
 
@@ -120,19 +120,25 @@ Distributions not in that crate can also be supported, but would require impleme
 ### Peer configuration
 
 Running in peer-to-peer mode is very similar to running a client.
-The only differences are that recipients in a group must be listed, users and recipients consist of a name and address each, and there is no server.
+The main difference is that the first argument is the path of a "hosts file".
+This file should have on each line an address:port and one or more user names, separated by whitespace (note that this is slightly different syntax to a real hosts file, as there is a port included).
+When a peer connects to another peer, it uses the hosts file to look up the address to connect to.
+See the minimal [example hosts file](/shadow/peer/shadow.data.template/hosts/hostsfile).
+For consistency, this file should be shared by all hosts.
+
+Aside from that, the only differences in the configuration are that recipients in a group must be listed, there is no server, and there is an optional `listen` configuration field to specify which interface the user should listen for connections on (if not provided, the interface given in the hosts file is used).
 Here is an example peer conversation configuration (again, all values are for demonstration purposes only):
 
 ```YAML
 # peer-conversation.yaml
 
-user: { name: "Alice", address: "127.0.0.1:6397" }
+user: "Alice"
 socks: "127.0.0.1:9050"
+listen: "127.0.0.1:6397"
 
 conversations:
   - group: "group1"
-    recipients:
-      - { name: "Bob", address: "insert.ip.or.onion:6397" }
+    recipients: "Bob"
     bootstrap: 5.0
     retry: 5.0
     distributions:
@@ -147,7 +153,8 @@ conversations:
 
 Additional examples can be found in the [peer shadow test configurations](/shadow/peer/shadow.data.template/hosts).
 
-In the likely case that these peers are connecting via onion addresses, you must configure each torrc file to match with each peer configuration (in the above example, Alice's HiddenService lines in the torrc must have a `HiddenServicePort` line that forwards to `127.0.0.1:6397`, and Bob's torrc must have a `HiddenServicePort` line that listens on `6397`).
+In the likely case that these peers are connecting via onion addresses, you must configure each torrc file to match with each peer configuration and hosts file
+(in the above example, Alice's HiddenService lines in the torrc must have a `HiddenServicePort` line that forwards to `127.0.0.1:6397`, and the hosts file must have a line with `[string].onion:[port] Alice`, where "string" and "port" correspond to Alice's onion address and external port).
 Multiple users can share an onion address by using different ports (different circuits will be used), though doing so will of course not simulate, e.g., additional load in Tor's distributed hash table.
 
 ### Weighted

shadow/peer/shadow.data.template/hosts/hostsfile

@@ -0,0 +1,4 @@
+100.0.0.1:6397 Alice
+100.0.0.2:6397 Bob
+100.0.0.2:6398 Carol
+100.0.0.3:6397 Dave

shadow/peer/shadow.data.template/hosts/peer1/alice.yaml

@@ -1,9 +1,7 @@
-user: { name: "Alice", address: "100.0.0.1:6397" }
+user: "Alice"
 conversations:
   - group: "group1"
-    recipients:
-      - { name: "Bob", address: "100.0.0.2:6397" }
-      - { name: "Carol", address: "100.0.0.2:6398"}
+    recipients: ["Bob", "Carol"]
     bootstrap: 5.0
     retry: 5.0
     distributions: &dists
@@ -15,9 +13,7 @@ conversations:
       a_s: { distribution: "Normal", mean: 10.0, std_dev: 5.0 }
       a_r: { distribution: "Normal", mean: 10.0, std_dev: 5.0 }
   - group: "group2"
-    recipients:
-      - { name: "Bob", address: "100.0.0.2:6397" }
-      - { name: "Dave", address: "100.0.0.3:6397" }
+    recipients: ["Bob", "Dave"]
     bootstrap: 5.0
     retry: 5.0
     distributions: *dists

shadow/peer/shadow.data.template/hosts/peer2/bob.yaml

@@ -1,9 +1,7 @@
-user: { name: "Bob", address: "100.0.0.2:6397" }
+user: "Bob"
 conversations:
   - group: "group1"
-    recipients:
-      - { name: "Alice", address: "100.0.0.1:6397" }
-      - { name: "Carol", address: "100.0.0.2:6398" }
+    recipients: ["Alice", "Carol"]
     bootstrap: 5.0
     retry: 5.0
     distributions: &dists
@@ -15,9 +13,7 @@ conversations:
       a_s: { distribution: "Normal", mean: 10.0, std_dev: 5.0 }
       a_r: { distribution: "Normal", mean: 10.0, std_dev: 5.0 }
   - group: "group2"
-    recipients:
-      - { name: "Alice", address: "100.0.0.1:6397" }
-      - { name: "Dave", address: "100.0.0.3:6397" }
+    recipients: ["Alice", "Dave"]
     bootstrap: 5.0
     retry: 5.0
     distributions: *dists

shadow/peer/shadow.data.template/hosts/peer2/carol.yaml

@@ -1,9 +1,7 @@
-user: { name: "Carol", address: "100.0.0.2:6398" }
+user: "Carol"
 conversations:
   - group: "group1"
-    recipients:
-      - { name: "Alice", address: "100.0.0.1:6397" }
-      - { name: "Bob", address: "100.0.0.2:6397" }
+    recipients: ["Alice", "Bob"]
     bootstrap: 5.0
     retry: 5.0
     distributions:

shadow/peer/shadow.data.template/hosts/peer3/dave.yaml

@@ -1,9 +1,7 @@
-user: { name: "Dave", address: "100.0.0.3:6397" }
+user: "Dave"
 conversations:
   - group: "group2"
-    recipients:
-      - { name: "Alice", address: "100.0.0.1:6397" }
-      - { name: "Bob", address: "100.0.0.2:6397" }
+    recipients: ["Alice", "Bob"]
     bootstrap: 5.0
     retry: 5.0
     distributions:

shadow/peer/shadow.yaml

@@ -13,7 +13,7 @@ hosts:
     ip_addr: 100.0.0.1
     processes: &proc
     - path: mgen-peer
-      args: "*.yaml"
+      args: "../hostsfile *.yaml"
       start_time: 5s
   peer2:
     network_node_id: 0

src/bin/mgen-peer.rs

@@ -284,19 +284,23 @@ async fn writer<'a>(
     }
 }
 
-/// This user or a recipient.
-/// If this user, address is a local address to listen on.
-/// If a recipient, address is a remote address to send to.
-#[derive(Debug, Deserialize)]
-struct Peer {
-    name: String,
-    address: String,
+fn parse_hosts_file(file_contents: &str) -> HashMap<&str, &str> {
+    let mut ret = HashMap::new();
+    for line in file_contents.lines() {
+        let mut words = line.split_ascii_whitespace();
+        if let Some(addr) = words.next() {
+            for name in words {
+                ret.insert(name, addr);
+            }
+        }
+    }
+    ret
 }
 
 #[derive(Debug, Deserialize)]
 struct ConversationConfig {
     group: String,
-    recipients: Vec<Peer>,
+    recipients: Vec<String>,
     bootstrap: f64,
     retry: f64,
     distributions: ConfigDistributions,
@@ -304,13 +308,15 @@ struct ConversationConfig {
 
 #[derive(Debug, Deserialize)]
 struct Config {
-    user: Peer,
+    user: String,
     socks: Option<String>,
+    listen: Option<String>,
     conversations: Vec<ConversationConfig>,
 }
 
 fn process_config(
     config: Config,
+    hosts_map: &HashMap<&str, &str>,
     handles: &mut Vec<JoinHandle<Result<(), FatalError>>>,
 ) -> Result<(), Box<dyn std::error::Error>> {
     struct ForIoThreads {
@@ -332,15 +338,18 @@ fn process_config(
             );
 
         for recipient in conversation.recipients.iter() {
-            let state_to_writer = if !recipient_map.contains_key(&recipient.name) {
+            let state_to_writer = if !recipient_map.contains_key(recipient) {
                 let (state_to_writer, writer_from_state) = mpsc::unbounded_channel();
                 let mut reader_to_states = HashMap::new();
                 reader_to_states.insert(conversation.group.clone(), reader_to_state.clone());
+                let address = hosts_map
+                    .get(recipient.as_str())
+                    .unwrap_or_else(|| panic!("recipient not in hosts file: {}", recipient));
                 let str_params = SocksParams {
                     socks: config.socks.clone(),
-                    target: recipient.address.clone(),
-                    user: config.user.name.clone(),
-                    recipient: recipient.name.clone(),
+                    target: address.to_string(),
+                    user: config.user.clone(),
+                    recipient: recipient.clone(),
                 };
                 let for_io = ForIoThreads {
                     state_to_writer: state_to_writer.clone(),
@@ -349,10 +358,10 @@ fn process_config(
                     str_params,
                     retry: conversation.retry,
                 };
-                recipient_map.insert(recipient.name.clone(), for_io);
+                recipient_map.insert(recipient.clone(), for_io);
                 state_to_writer
             } else {
-                let for_io = recipient_map.get_mut(&recipient.name).unwrap();
+                let for_io = recipient_map.get_mut(recipient).unwrap();
                 if !for_io.reader_to_states.contains_key(&conversation.group) {
                     for_io
                         .reader_to_states
@@ -361,7 +370,7 @@ fn process_config(
                 for_io.state_to_writer.clone()
             };
             conversation_recipient_map.insert(
-                recipient.name.clone(),
+                recipient.clone(),
                 StateToWriter {
                     channel: state_to_writer,
                 },
@@ -371,7 +380,7 @@ fn process_config(
         let distributions: Distributions = conversation.distributions.try_into()?;
 
         tokio::spawn(manage_conversation(
-            config.user.name.clone(),
+            config.user.clone(),
             conversation.group,
             distributions,
             conversation.bootstrap,
@@ -406,8 +415,16 @@ fn process_config(
         handles.push(handle);
     }
 
+    let address = if let Some(address) = config.listen {
+        address
+    } else {
+        hosts_map
+            .get(config.user.as_str())
+            .unwrap_or_else(|| panic!("user not found in hosts file: {}", config.user))
+            .to_string()
+    };
     let handle: JoinHandle<Result<(), FatalError>> =
-        tokio::spawn(listener(config.user.address, name_to_io_threads));
+        tokio::spawn(listener(address, name_to_io_threads));
     handles.push(handle);
     Ok(())
 }
@@ -416,13 +433,16 @@ fn process_config(
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let mut args = std::env::args();
     let _ = args.next();
+    let hosts_file = std::fs::read_to_string(args.next().unwrap())?;
+    let hosts_map = parse_hosts_file(&hosts_file);
+    println!("{:?}", hosts_map);
 
     let mut handles = vec![];
 
     for config_file in args.flat_map(|a| glob::glob(a.as_str()).unwrap()) {
         let yaml_s = std::fs::read_to_string(config_file?)?;
         let config: Config = serde_yaml::from_str(&yaml_s)?;
-        process_config(config, &mut handles)?;
+        process_config(config, &hosts_map, &mut handles)?;
     }
 
     let handles: futures::stream::FuturesUnordered<_> = handles.into_iter().collect();