Deuxfleurs/diplonat
23
4
Fork 1
Code Issues 8 Pull requests Releases Wiki Activity

Modular Diplonat (with the option to disable useless modules) #8

Closed
adrien wants to merge 12 commits from adrien/diplonat:feature/modular-config into main
pull from: adrien/diplonat:feature/modular-config
merge into: Deuxfleurs:main
Conversation 4 Commits 12 Files changed 11 +424 -358
13 changed files with 424 additions and 358 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit 2e797d2f62 - Show all commits

2
.rustfmt.toml Normal file
View file

@ -0,0 +1,2 @@
hard_tabs = false
tab_spaces = 2

8
src/config/mod.rs
View file

@ -3,8 +3,12 @@ mod options;
mod options_test;
mod runtime;
pub use options::{ConfigOpts, ConfigOptsConsul, ConfigOptsAcme, ConfigOptsFirewall, ConfigOptsIgd};
pub use runtime::{RuntimeConfig, RuntimeConfigAcme, RuntimeConfigConsul, RuntimeConfigFirewall, RuntimeConfigIgd};
pub use options::{
ConfigOpts, ConfigOptsAcme, ConfigOptsConsul, ConfigOptsFirewall, ConfigOptsIgd,
};
pub use runtime::{
RuntimeConfig, RuntimeConfigAcme, RuntimeConfigConsul, RuntimeConfigFirewall, RuntimeConfigIgd,
};
pub const EXPIRATION_TIME: u16 = 300;
pub const REFRESH_TIME: u16 = 60;

8
src/config/options.rs
View file

@ -15,7 +15,6 @@ use crate::config::RuntimeConfig;
// Only in runtime.rs would these options find their proper location in each
// module's struct.
/// Consul configuration options
#[derive(Clone, Default, Deserialize)]
pub struct ConfigOptsConsul {
@ -88,10 +87,13 @@ impl ConfigOpts {
// Currently only used in tests
#[allow(dead_code)]
pub fn from_iter<Iter: Clone>(iter: Iter) -> Result<RuntimeConfig>
where Iter: IntoIterator<Item = (String, String)> {
where
Iter: IntoIterator<Item = (String, String)>,
{
let consul: ConfigOptsConsul = envy::prefixed("DIPLONAT_CONSUL_").from_iter(iter.clone())?;
let acme: ConfigOptsAcme = envy::prefixed("DIPLONAT_ACME_").from_iter(iter.clone())?;
let firewall: ConfigOptsFirewall = envy::prefixed("DIPLONAT_FIREWALL_").from_iter(iter.clone())?;
let firewall: ConfigOptsFirewall =
envy::prefixed("DIPLONAT_FIREWALL_").from_iter(iter.clone())?;
let igd: ConfigOptsIgd = envy::prefixed("DIPLONAT_IGD_").from_iter(iter.clone())?;
RuntimeConfig::new(Self {

78
src/config/options_test.rs
View file

@ -11,19 +11,34 @@ use crate::config::*;
fn minimal_valid_options() -> HashMap<String, String> {
let mut opts = HashMap::new();
opts.insert("DIPLONAT_CONSUL_NODE_NAME".to_string(), "consul_node".to_string());
opts.insert(
"DIPLONAT_CONSUL_NODE_NAME".to_string(),
"consul_node".to_string(),
);
opts
}
fn all_valid_options() -> HashMap<String, String> {
let mut opts = minimal_valid_options();
opts.insert("DIPLONAT_CONSUL_URL".to_string(), "http://127.0.0.1:9999".to_string());
opts.insert(
"DIPLONAT_CONSUL_URL".to_string(),
"http://127.0.0.1:9999".to_string(),
);
opts.insert("DIPLONAT_ACME_ENABLE".to_string(), "true".to_string());
opts.insert("DIPLONAT_ACME_EMAIL".to_string(), "bozo@bozo.net".to_string());
opts.insert(
"DIPLONAT_ACME_EMAIL".to_string(),
"bozo@bozo.net".to_string(),
);
opts.insert("DIPLONAT_FIREWALL_ENABLE".to_string(), "true".to_string());
opts.insert("DIPLONAT_FIREWALL_REFRESH_TIME".to_string(), "20".to_string());
opts.insert(
"DIPLONAT_FIREWALL_REFRESH_TIME".to_string(),
"20".to_string(),
);
opts.insert("DIPLONAT_IGD_ENABLE".to_string(), "true".to_string());
opts.insert("DIPLONAT_IGD_PRIVATE_IP".to_string(), "172.123.43.555".to_string());
opts.insert(
"DIPLONAT_IGD_PRIVATE_IP".to_string(),
"172.123.43.555".to_string(),
);
opts.insert("DIPLONAT_IGD_EXPIRATION_TIME".to_string(), "60".to_string());
opts.insert("DIPLONAT_IGD_REFRESH_TIME".to_string(), "10".to_string());
opts
@ -53,10 +68,7 @@ fn ok_minimal_valid_options() {
&rt_config.consul.node_name,
opts.get(&"DIPLONAT_CONSUL_NODE_NAME".to_string()).unwrap()
);
assert_eq!(
rt_config.consul.url,
CONSUL_URL.to_string()
);
assert_eq!(rt_config.consul.url, CONSUL_URL.to_string());
assert!(rt_config.acme.is_none());
assert!(rt_config.firewall.is_none());
assert!(rt_config.igd.is_none());
@ -94,17 +106,29 @@ fn ok_all_valid_options() {
let rt_config = ConfigOpts::from_iter(opts.clone()).unwrap();
let firewall_refresh_time = Duration::from_secs(
opts.get(&"DIPLONAT_FIREWALL_REFRESH_TIME".to_string()).unwrap()
.parse::<u64>().unwrap()
.into());
opts
.get(&"DIPLONAT_FIREWALL_REFRESH_TIME".to_string())
.unwrap()
.parse::<u64>()
.unwrap()
.into(),
);
let igd_expiration_time = Duration::from_secs(
opts.get(&"DIPLONAT_IGD_EXPIRATION_TIME".to_string()).unwrap()
.parse::<u64>().unwrap()
.into());
opts
.get(&"DIPLONAT_IGD_EXPIRATION_TIME".to_string())
.unwrap()
.parse::<u64>()
.unwrap()
.into(),
);
let igd_refresh_time = Duration::from_secs(
opts.get(&"DIPLONAT_IGD_REFRESH_TIME".to_string()).unwrap()
.parse::<u64>().unwrap()
.into());
opts
.get(&"DIPLONAT_IGD_REFRESH_TIME".to_string())
.unwrap()
.parse::<u64>()
.unwrap()
.into(),
);
assert_eq!(
&rt_config.consul.node_name,
@ -119,14 +143,12 @@ fn ok_all_valid_options() {
let acme = rt_config.acme.unwrap();
assert_eq!(
&acme.email,
opts.get(&"DIPLONAT_ACME_EMAIL".to_string()).unwrap());
opts.get(&"DIPLONAT_ACME_EMAIL".to_string()).unwrap()
);
assert!(rt_config.firewall.is_some());
let firewall = rt_config.firewall.unwrap();
assert_eq!(
firewall.refresh_time,
firewall_refresh_time
);
assert_eq!(firewall.refresh_time, firewall_refresh_time);
assert!(rt_config.igd.is_some());
let igd = rt_config.igd.unwrap();
@ -134,12 +156,6 @@ fn ok_all_valid_options() {
&igd.private_ip,
opts.get(&"DIPLONAT_IGD_PRIVATE_IP".to_string()).unwrap()
);
assert_eq!(
igd.expiration_time,
igd_expiration_time
);
assert_eq!(
igd.refresh_time,
igd_refresh_time
);
assert_eq!(igd.expiration_time, igd_expiration_time);
assert_eq!(igd.refresh_time, igd_refresh_time);
}

46
src/config/runtime.rs
View file

@ -1,8 +1,10 @@
use std::time::Duration;
use anyhow::{Result, anyhow};
use anyhow::{anyhow, Result};
use crate::config::{ConfigOpts, ConfigOptsConsul, ConfigOptsAcme, ConfigOptsFirewall, ConfigOptsIgd};
use crate::config::{
ConfigOpts, ConfigOptsAcme, ConfigOptsConsul, ConfigOptsFirewall, ConfigOptsIgd,
};
// This code is inspired by the Trunk crate (https://github.com/thedodd/trunk)
@ -60,14 +62,12 @@ impl RuntimeConfig {
impl RuntimeConfigConsul {
pub(super) fn new(opts: ConfigOptsConsul) -> Result<Self> {
let node_name = opts.node_name.expect(
"'DIPLONAT_CONSUL_NODE_NAME' is required");
let node_name = opts
.node_name
.expect("'DIPLONAT_CONSUL_NODE_NAME' is required");
quentin commented 2021-09-20 15:51:11 +00:00
Outdated
Review
Copy link

cf my following comment on DIPLONAT_ACME_EMAIL

cf my following comment on `DIPLONAT_ACME_EMAIL`
let url = opts.url.unwrap_or(super::CONSUL_URL.to_string());
Ok(Self {
node_name,
url,
})
Ok(Self { node_name, url })
}
}
@ -77,12 +77,11 @@ impl RuntimeConfigAcme {
return Ok(None);
}
let email = opts.email.expect(
"'DIPLONAT_ACME_EMAIL' is required if ACME is enabled");
let email = opts
.email
.expect("'DIPLONAT_ACME_EMAIL' is required if ACME is enabled");
quentin commented 2021-09-20 15:49:07 +00:00
Outdated
Review
Copy link

I know this is not linked to this specific pull request but when your function returns a Result, there is no reason to panic with expect.

Instead, replace it with:

let email = opts
  .email
  .context("'DIPLONAT_ACME_EMAIL' is required if ACME is enabled")?;

Please do it for the whole file :)

I know this is not linked to this specific pull request but when your function returns a Result, there is no reason to panic with expect. Instead, replace it with: ```rust let email = opts .email .context("'DIPLONAT_ACME_EMAIL' is required if ACME is enabled")?; ``` Please do it for the whole file :)
adrien commented 2021-09-22 11:13:56 +00:00
Outdated
Review
Copy link

.context(...) is not a method of Option.

There's sense in doing so: when a parameter is invalid (like here), the method should panic or return an Err(..).

I'm still modifying the calls to return errors instead of panicking straight away. Using match for better clarity ;)

`.context(...)` is not a method of `Option`. There's sense in doing so: when a parameter is invalid (like here), the method should panic or return an Err(..). I'm still modifying the calls to return errors instead of panicking straight away. Using `match` for better clarity ;)
quentin commented 2021-09-23 05:27:22 +00:00
Outdated
Review
Copy link

You can convert an Option to an Error with ok_or: https://doc.rust-lang.org/std/option/enum.Option.html#method.ok_or

But we need an Error to map to the None. It is a bit challenging as it will be our first error in Diplonat, previously we were relaying only on errors generated by our dependencies.

For now, let's keep it simple, our error library provides us a simple macro to create an error from a string. Just a simple example:

return Err(anyhow!("Missing attribute: {}", missing))

So you can write:

let email = opts
  .email
  .ok_or(anyhow!("'DIPLONAT_ACME_EMAIL' is required if ACME is enabled"))?;

For your information, context allows to enrich an existing error with some context. Indeed, here it is not appropriate as we have no error at all. But in the case where your library throw a cryptic error, you can enrich this error with some context in your program and your users will thank you many times later.

You made the following point:

There's sense in doing so: when a parameter is invalid (like here), the method should panic or return an Err(..).

First, you're right, when a parameter is invalid, we must handle this as an error. It was what I was suggesting to you in my first comment but I thought the return type of opts.email would be a Result and not an Option.

Second, we are writing Rust and we are in a library: in this situation, it is not acceptable to panic. As there is no clean way to recover from a panic, we terminate the program from a library without giving the program a chance to recover from the error. Even if we are not a "standalone library" and are writing the main program, we might change our mind the future. We should panic! only when there is no option. To illustrate my point, we could think that later, we would allow our main program to reload its configuration and retry the initialization if one actor fails.

Third, I am not sure we discussed it but ? is syntaxic sugar for:

let err = anyhow!("'DIPLONAT_ACME_EMAIL' is required if ACME is enabled")

match opts.email.ok_or(err) {
  Ok(email) => {
    // our logic
  }
  Err(e) => return Err(e)
}

So in fact, what I suggest is that we return from the function with an Error if this parameter has not be provided.

You can convert an Option to an Error with `ok_or`: https://doc.rust-lang.org/std/option/enum.Option.html#method.ok_or But we need an Error to map to the None. It is a bit challenging as it will be our first error in Diplonat, previously we were relaying only on errors generated by our dependencies. For now, let's keep it simple, our error library provides us a simple macro to create an error from a string. Just a simple example: ```rust return Err(anyhow!("Missing attribute: {}", missing)) ``` So you can write: ```rust let email = opts .email .ok_or(anyhow!("'DIPLONAT_ACME_EMAIL' is required if ACME is enabled"))?; ``` For your information, `context` allows to enrich an existing error with some context. Indeed, here it is not appropriate as we have no error at all. But in the case where your library throw a cryptic error, you can enrich this error with some context in your program and your users will thank you many times later. You made the following point: > There's sense in doing so: when a parameter is invalid (like here), the method should panic or return an Err(..). First, you're right, when a parameter is invalid, we must handle this as an error. It was what I was suggesting to you in my first comment but I thought the return type of `opts.email` would be a `Result` and not an `Option`. Second, we are writing Rust and we are in a library: in this situation, it is not acceptable to panic. As there is no clean way to recover from a panic, we terminate the program from a library without giving the program a chance to recover from the error. Even if we are not a "standalone library" and are writing the main program, we might change our mind the future. We should `panic!` only when there is no option. To illustrate my point, we could think that later, we would allow our main program to reload its configuration and retry the initialization if one actor fails. Third, I am not sure we discussed it but `?` is syntaxic sugar for: ```rust let err = anyhow!("'DIPLONAT_ACME_EMAIL' is required if ACME is enabled") match opts.email.ok_or(err) { Ok(email) => { // our logic } Err(e) => return Err(e) } ``` So in fact, what I suggest is that we return from the function with an Error if this parameter has not be provided.
adrien commented 2021-09-23 08:01:08 +00:00
Outdated
Review
Copy link

Here's my new RuntimeConfigAcme implementation in full (in commit f5ac36e). Do you find it better?

impl RuntimeConfigAcme {
  pub fn new(opts: ConfigOptsAcme) -> Result<Option<Self>> {
    if !opts.enable {
      return Ok(None)
    };

    let email = match opts.email {
      Some(email) => email,
      _ => {
        return Err(anyhow!(
          "'DIPLONAT_ACME_EMAIL' is required if ACME is enabled"
        ))
      }
    };

    Ok(Some(Self { email }))
  }
}

We do return an error instead of panicking. But I may have overdone it with using match, but I didn't find any nice shorthand in Option methods (I checked ok_or). At least the intent is clear!

PS: Quite puzzled by the fact that Rust allows putting a return inside a match arm, which should be an expression returning a String (in this case). Welp, gotta get used to it!

Here's my new RuntimeConfigAcme implementation in full (in commit `f5ac36e`). Do you find it better? ```rust impl RuntimeConfigAcme { pub fn new(opts: ConfigOptsAcme) -> Result<Option<Self>> { if !opts.enable { return Ok(None) }; let email = match opts.email { Some(email) => email, _ => { return Err(anyhow!( "'DIPLONAT_ACME_EMAIL' is required if ACME is enabled" )) } }; Ok(Some(Self { email })) } } ``` We do return an error instead of panicking. But I may have overdone it with using match, but I didn't find any nice shorthand in `Option` methods (I checked `ok_or`). At least the intent is clear! PS: Quite puzzled by the fact that Rust allows putting a `return` inside a `match` arm, which should be an expression returning a String (in this case). Welp, gotta get used to it!
quentin commented 2021-09-23 12:26:17 +00:00
Outdated
Review
Copy link

Sorry, I was not clear enough ><

About handling error, I want to avoid expect and unwrap because they make our program panic, ie. the program is killed.

But unwrap_or is totally fine as it does not make the program panic.
And when the case is simple enough we can use one of these keywords: unwrap_or on Result or ok_or on Options instead of a match, it is totally ok.

So, I don't think you should replace your unwrap_or with a match in f5ac36e.

I have written a small example on how to use ok_or to convert an Option into a Result and checked that it compiles here: https://play.rust-lang.org/?version=stable&mode=debug&edition=2018&gist=5bc6046485bab0cd4ae694b94d2ba1db

See the code without leaving gitea 
use std::{
  fmt,
  error::Error
};

#[derive(Debug)]
struct RequiredParameterError;
impl Error for RequiredParameterError {}
impl fmt::Display for RequiredParameterError {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "A required parameter is not defined")
    }
}

fn foo(param: Option<String>) -> Result<String, RequiredParameterError> {
  let email = param.ok_or(RequiredParameterError{})?;
  return Ok("maito:".to_string() + &email)
}

fn main() {
  let user_param1 = Some("hello@example.org".to_string());
  let user_param2 = None;

  assert!(foo(user_param1).is_ok());
  assert!(foo(user_param2).is_err());
}

So, you may ask why I was advocating for match?
Because in fw_actors.rs you were using again unwrap() that can cause a panic in the program and we can't replace it by these simple keywords that are ok_or, unwrap_or, etc.

To sum up, I think we can define the following rules:

  1. Never write unwrap() or expect() (yes, never is too brutal, but it should be an exception)
  2. Replace them, when possible, by an appropriate small function if it exists
  3. Otherwise, if your case is too specific, use a match

I hope that this time I will be more clear.
I know that when we start coding together, it takes some times to align our minds but once it will be done, it will be easier :)

Sorry, I was not clear enough >< About handling error, I want to avoid `expect` and `unwrap` because they make our program panic, ie. the program is killed. But `unwrap_or` is totally fine as it does not make the program panic. And when the case is simple enough we can use one of these keywords: `unwrap_or` on `Result` or `ok_or` on Options instead of a `match`, it is totally ok. So, I don't think you should replace your `unwrap_or` with a `match` in `f5ac36e`. I have written a small example on how to use `ok_or` to convert an `Option` into a `Result` and checked that it compiles here: https://play.rust-lang.org/?version=stable&mode=debug&edition=2018&gist=5bc6046485bab0cd4ae694b94d2ba1db <details> <summary>See the code without leaving gitea</summary> ```rust use std::{ fmt, error::Error }; #[derive(Debug)] struct RequiredParameterError; impl Error for RequiredParameterError {} impl fmt::Display for RequiredParameterError { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { write!(f, "A required parameter is not defined") } } fn foo(param: Option<String>) -> Result<String, RequiredParameterError> { let email = param.ok_or(RequiredParameterError{})?; return Ok("maito:".to_string() + &email) } fn main() { let user_param1 = Some("hello@example.org".to_string()); let user_param2 = None; assert!(foo(user_param1).is_ok()); assert!(foo(user_param2).is_err()); } ``` </details> <br/> So, you may ask why I was advocating for `match`? Because in `fw_actors.rs` you were using again `unwrap()` that can cause a panic in the program and we can't replace it by these simple keywords that are `ok_or`, `unwrap_or`, etc. To sum up, I think we can define the following rules: 1. Never write `unwrap()` or `expect()` (yes, never is too brutal, but it should be an exception) 2. Replace them, when possible, by an appropriate small function if it exists 3. Otherwise, if your case is too specific, use a match I hope that this time I will be more clear. I know that when we start coding together, it takes some times to align our minds but once it will be done, it will be easier :)
Ok(Some(Self {
email,
}))
Ok(Some(Self { email }))
}
}
@ -92,12 +91,9 @@ impl RuntimeConfigFirewall {
return Ok(None);
}
let refresh_time = Duration::from_secs(
opts.refresh_time.unwrap_or(super::REFRESH_TIME).into());
let refresh_time = Duration::from_secs(opts.refresh_time.unwrap_or(super::REFRESH_TIME).into());
Ok(Some(Self {
refresh_time,
}))
Ok(Some(Self { refresh_time }))
}
}
@ -107,12 +103,16 @@ impl RuntimeConfigIgd {
return Ok(None);
}
let private_ip = opts.private_ip.expect(
"'DIPLONAT_IGD_PRIVATE_IP' is required if IGD is enabled");
let private_ip = opts
.private_ip
.expect("'DIPLONAT_IGD_PRIVATE_IP' is required if IGD is enabled");
let expiration_time = Duration::from_secs(
opts.expiration_time.unwrap_or(super::EXPIRATION_TIME).into());
let refresh_time = Duration::from_secs(
opts.refresh_time.unwrap_or(super::REFRESH_TIME).into());
opts
.expiration_time
.unwrap_or(super::EXPIRATION_TIME)
.into(),
);
let refresh_time = Duration::from_secs(opts.refresh_time.unwrap_or(super::REFRESH_TIME).into());
if refresh_time.as_secs() * 2 > expiration_time.as_secs() {
return Err(anyhow!(

18
src/consul.rs
View file

@ -1,22 +1,22 @@
use std::collections::HashMap;
use anyhow::{Result, anyhow};
use serde::{Serialize, Deserialize};
use anyhow::{anyhow, Result};
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize, Debug)]
pub struct ServiceEntry {
pub Tags: Vec<String>
pub Tags: Vec<String>,
}
#[derive(Serialize, Deserialize, Debug)]
pub struct CatalogNode {
pub Services: HashMap<String, ServiceEntry>
pub Services: HashMap<String, ServiceEntry>,
}
pub struct Consul {
client: reqwest::Client,
url: String,
idx: Option<u64>
idx: Option<u64>,
}
impl Consul {
@ -24,7 +24,7 @@ impl Consul {
return Self {
client: reqwest::Client::new(),
url: url.to_string(),
idx: None
idx: None,
};
}
@ -35,16 +35,16 @@ impl Consul {
pub async fn watch_node(&mut self, host: &str) -> Result<CatalogNode> {
let url = match self.idx {
Some(i) => format!("{}/v1/catalog/node/{}?index={}", self.url, host, i),
None => format!("{}/v1/catalog/node/{}", self.url, host)
None => format!("{}/v1/catalog/node/{}", self.url, host),
};
let http = self.client.get(&url).send().await?;
self.idx = match http.headers().get("X-Consul-Index") {
Some(v) => Some(v.to_str()?.parse::<u64>()?),
None => return Err(anyhow!("X-Consul-Index header not found"))
None => return Err(anyhow!("X-Consul-Index header not found")),
};
let resp: CatalogNode = http.json().await?;
return Ok(resp)
return Ok(resp);
}
}

27
src/consul_actor.rs
View file

@ -4,8 +4,8 @@ use std::time::Duration;
use anyhow::Result;
use log::*;
use serde::{Serialize, Deserialize};
use serde_lexpr::{from_str,error};
use serde::{Deserialize, Serialize};
use serde_lexpr::{error, from_str};
use tokio::sync::watch;
use tokio::time::delay_for;
@ -16,12 +16,12 @@ use crate::messages;
#[derive(Serialize, Deserialize, Debug)]
pub enum DiplonatParameter {
tcp_port(HashSet<u16>),
udp_port(HashSet<u16>)
udp_port(HashSet<u16>),
}
#[derive(Serialize, Deserialize, Debug)]
pub enum DiplonatConsul {
diplonat(Vec<DiplonatParameter>)
diplonat(Vec<DiplonatParameter>),
}
pub struct ConsulActor {
@ -31,13 +31,16 @@ pub struct ConsulActor {
node: String,
retries: u32,
tx_open_ports: watch::Sender<messages::PublicExposedPorts>
tx_open_ports: watch::Sender<messages::PublicExposedPorts>,
}
fn retry_to_time(retries: u32, max_time: Duration) -> Duration {
// 1.2^x seems to be a good value to exponentially increase time at a good pace
// eg. 1.2^32 = 341 seconds ~= 5 minutes - ie. after 32 retries we wait 5 minutes
return Duration::from_secs(cmp::min(max_time.as_secs(), 1.2f64.powf(retries as f64) as u64))
return Duration::from_secs(cmp::min(
max_time.as_secs(),
1.2f64.powf(retries as f64) as u64,
));
}
fn to_parameters(catalog: &consul::CatalogNode) -> Vec<DiplonatConsul> {
@ -59,7 +62,7 @@ fn to_parameters(catalog: &consul::CatalogNode) -> Vec<DiplonatConsul> {
fn to_open_ports(params: &Vec<DiplonatConsul>) -> messages::PublicExposedPorts {
let mut op = messages::PublicExposedPorts {
tcp_ports: HashSet::new(),
udp_ports: HashSet::new()
udp_ports: HashSet::new(),
};
for conf in params {
@ -77,9 +80,9 @@ fn to_open_ports(params: &Vec<DiplonatConsul>) -> messages::PublicExposedPorts {
impl ConsulActor {
pub fn new(config: RuntimeConfigConsul) -> Self {
let (tx, rx) = watch::channel(messages::PublicExposedPorts{
let (tx, rx) = watch::channel(messages::PublicExposedPorts {
tcp_ports: HashSet::new(),
udp_ports: HashSet::new()
udp_ports: HashSet::new(),
});
return Self {
@ -99,7 +102,11 @@ impl ConsulActor {
self.consul.watch_node_reset();
self.retries = cmp::min(std::u32::MAX - 1, self.retries) + 1;
let will_retry_in = retry_to_time(self.retries, Duration::from_secs(600));
error!("Failed to query consul. Will retry in {}s. {}", will_retry_in.as_secs(), e);
error!(
"Failed to query consul. Will retry in {}s. {}",
will_retry_in.as_secs(),
e
);
delay_for(will_retry_in).await;
continue;
}

19
src/diplonat.rs
View file

@ -1,4 +1,4 @@
use anyhow::{Result, anyhow};
use anyhow::{anyhow, Result};
use tokio::try_join;
use crate::config::ConfigOpts;
@ -20,19 +20,14 @@ impl Diplonat {
quentin commented 2021-09-20 16:06:44 +00:00
Review
Copy link

Now that our actors are modular, I think that this function fails to convey the real meaning of what we want to do.

IMHO, I think we want to:

  1. Instantiate actors chosen by the user
  2. Run them

With the current code, I see many problems.

((1)) I have a feeling that we are duplicating logic between the "runtime config" and our actor initialization. Currently, the constructor of the actor is only:

  • some boilerplate
  • a duplicated check found in runtime

To fix this problem, I see 2 options:

  • Delete your runtime configuration file and let actors initialize themselves from their constructor
  • Remove actors' constructor and instantiate them directly from this function only if they are required

((2)) The current try_join! fails to convey the information that we run only the initialized actors. I think that, independently of the choice you made for 1, you should create a vector of initialized actors that you will populate and then run. You will replace try_join! by try_join_all

Tip: start by solving problem number 2, it will help you solve problem number 1

Now that our actors are modular, I think that this function fails to convey the real meaning of what we want to do. IMHO, I think we want to: 1. Instantiate actors chosen by the user 2. Run them --- With the current code, I see many problems. ((1)) I have a feeling that we are duplicating logic between the "runtime config" and our actor initialization. Currently, the constructor of the actor is only: - some boilerplate - a duplicated check found in runtime To fix this problem, I see 2 options: - Delete your runtime configuration file and let actors initialize themselves from their constructor - Remove actors' constructor and instantiate them directly from this function only if they are required ((2)) The current `try_join!` fails to convey the information that we run only the initialized actors. I think that, independently of the choice you made for 1, you should create a vector of initialized actors that you will populate and then run. You will replace `try_join!` by [`try_join_all`](https://docs.rs/futures/0.3.17/futures/future/fn.try_join_all.html) *Tip: start by solving problem number 2, it will help you solve problem number 1*
let consul_actor = ConsulActor::new(config.consul);
let firewall_actor = FirewallActor::new(
config.firewall,
&consul_actor.rx_open_ports
).await?;
let firewall_actor = FirewallActor::new(config.firewall, &consul_actor.rx_open_ports).await?;
let igd_actor = IgdActor::new(
config.igd,
&consul_actor.rx_open_ports
).await?;
let igd_actor = IgdActor::new(config.igd, &consul_actor.rx_open_ports).await?;
if firewall_actor.is_none() && igd_actor.is_none() {
return Err(anyhow!(
"At least enable *one* module, otherwise it's boring!"));
"At least enable *one* module, otherwise it's boring!"
));
}
let ctx = Self {
@ -53,13 +48,13 @@ impl Diplonat {
async {
match firewall {
Some(x) => x.listen().await,
None => Ok(())
None => Ok(()),
}
},
async {
match igd {
Some(x) => x.listen().await,
None => Ok(())
None => Ok(()),
}
},
)?;

52
src/fw.rs
View file

@ -1,6 +1,6 @@
use std::collections::HashSet;
use anyhow::{Result,Context};
use anyhow::{Context, Result};
use iptables;
use log::*;
use regex::Regex;
@ -8,23 +8,38 @@ use regex::Regex;
use crate::messages;
pub fn setup(ipt: &iptables::IPTables) -> Result<()> {
// ensure we start from a clean state without any rule already set
cleanup(ipt)?;
ipt.new_chain("filter", "DIPLONAT").context("Failed to create new chain")?;
ipt.insert_unique("filter", "INPUT", "-j DIPLONAT", 1).context("Failed to insert jump rule")?;
ipt
.new_chain("filter", "DIPLONAT")
.context("Failed to create new chain")?;
ipt
.insert_unique("filter", "INPUT", "-j DIPLONAT", 1)
.context("Failed to insert jump rule")?;
Ok(())
}
pub fn open_ports(ipt: &iptables::IPTables, ports: messages::PublicExposedPorts) -> Result<()> {
for p in ports.tcp_ports {
ipt.append("filter", "DIPLONAT", &format!("-p tcp --dport {} -j ACCEPT", p)).context("Failed to insert port rule")?;
ipt
.append(
"filter",
"DIPLONAT",
&format!("-p tcp --dport {} -j ACCEPT", p),
)
.context("Failed to insert port rule")?;
}
for p in ports.udp_ports {
ipt.append("filter", "DIPLONAT", &format!("-p udp --dport {} -j ACCEPT", p)).context("Failed to insert port rule")?;
ipt
.append(
"filter",
"DIPLONAT",
&format!("-p udp --dport {} -j ACCEPT", p),
)
.context("Failed to insert port rule")?;
}
Ok(())
@ -33,18 +48,17 @@ pub fn open_ports(ipt: &iptables::IPTables, ports: messages::PublicExposedPorts)
pub fn get_opened_ports(ipt: &iptables::IPTables) -> Result<messages::PublicExposedPorts> {
let mut ports = messages::PublicExposedPorts {
tcp_ports: HashSet::new(),
udp_ports: HashSet::new()
udp_ports: HashSet::new(),
};
let list = ipt.list("filter", "DIPLONAT")?;
let re = Regex::new(r"\-A.*? \-p (\w+).*\-\-dport (\d+).*?\-j ACCEPT").context("Regex matching open ports encountered an unexpected rule")?;
let re = Regex::new(r"\-A.*? \-p (\w+).*\-\-dport (\d+).*?\-j ACCEPT")
.context("Regex matching open ports encountered an unexpected rule")?;
for i in list {
let caps = re.captures(&i);
match caps {
Some(c) => {
if let (Some(raw_proto), Some(raw_port)) = (c.get(1), c.get(2)) {
let proto = String::from(raw_proto.as_str());
let number = String::from(raw_port.as_str()).parse::<u16>()?;
@ -53,12 +67,10 @@ pub fn get_opened_ports(ipt: &iptables::IPTables) -> Result<messages::PublicExpo
} else {
ports.udp_ports.insert(number);
}
} else {
error!("Unexpected rule found in DIPLONAT chain")
}
},
}
_ => {}
}
}
@ -67,17 +79,21 @@ pub fn get_opened_ports(ipt: &iptables::IPTables) -> Result<messages::PublicExpo
}
pub fn cleanup(ipt: &iptables::IPTables) -> Result<()> {
if ipt.chain_exists("filter", "DIPLONAT")? {
ipt.flush_chain("filter", "DIPLONAT").context("Failed to flush the DIPLONAT chain")?;
ipt
.flush_chain("filter", "DIPLONAT")
.context("Failed to flush the DIPLONAT chain")?;
if ipt.exists("filter", "INPUT", "-j DIPLONAT")? {
ipt.delete("filter", "INPUT", "-j DIPLONAT").context("Failed to delete jump rule")?;
ipt
.delete("filter", "INPUT", "-j DIPLONAT")
.context("Failed to delete jump rule")?;
}
ipt.delete_chain("filter", "DIPLONAT").context("Failed to delete chain")?;
ipt
.delete_chain("filter", "DIPLONAT")
.context("Failed to delete chain")?;
}
Ok(())
}

34
src/fw_actor.rs
View file

@ -6,16 +6,13 @@ use log::*;
use tokio::{
select,
sync::watch,
time::{
Duration,
self,
}};
time::{self, Duration},
};
use crate::config::RuntimeConfigFirewall;
use crate::fw;
use crate::messages;
pub struct FirewallActor {
pub ipt: iptables::IPTables,
@ -26,7 +23,10 @@ pub struct FirewallActor {
}
impl FirewallActor {
pub async fn new(config: Option<RuntimeConfigFirewall>, rxp: &watch::Receiver<messages::PublicExposedPorts>) -> Result<Option<Self>> {
pub async fn new(
config: Option<RuntimeConfigFirewall>,
rxp: &watch::Receiver<messages::PublicExposedPorts>,
) -> Result<Option<Self>> {
if config.is_none() {
return Ok(None);
quentin commented 2021-09-20 16:12:14 +00:00
Outdated
Review
Copy link

No. Never use unwrap.

Your function should be written with a match:

impl FirewallActor {
  pub async fn new(
    _refresh: Duration,
    config: Option<RuntimeConfigFirewall>,
    rxp: &watch::Receiver<messages::PublicExposedPorts>,
  ) -> Result<Option<Self>> {
  match config {
    None => Ok(None)
    Some(c) => {
      let ctx = Self { ... }
      fw::setup(&ctx.ipt)?;
      return Ok(Some(ctx))
    }
  }
}

As a general rule, in many places in your code, you could better communicate your intent with a match

No. Never use unwrap. Your function should be written with a match: ```rust impl FirewallActor { pub async fn new( _refresh: Duration, config: Option<RuntimeConfigFirewall>, rxp: &watch::Receiver<messages::PublicExposedPorts>, ) -> Result<Option<Self>> { match config { None => Ok(None) Some(c) => { let ctx = Self { ... } fw::setup(&ctx.ipt)?; return Ok(Some(ctx)) } } } ``` As a general rule, in many places in your code, you could better communicate your intent with a `match`
}
@ -55,7 +55,9 @@ impl FirewallActor {
};
// 2. Update last ports if needed
if let Some(p) = new_ports { self.last_ports = p; }
if let Some(p) = new_ports {
self.last_ports = p;
}
// 3. Update firewall rules
match self.do_fw_update().await {
@ -68,18 +70,26 @@ impl FirewallActor {
pub async fn do_fw_update(&self) -> Result<()> {
let curr_opened_ports = fw::get_opened_ports(&self.ipt)?;
let diff_tcp = self.last_ports.tcp_ports.difference(&curr_opened_ports.tcp_ports).copied().collect::<HashSet<u16>>();
let diff_udp = self.last_ports.udp_ports.difference(&curr_opened_ports.udp_ports).copied().collect::<HashSet<u16>>();
let diff_tcp = self
.last_ports
.tcp_ports
.difference(&curr_opened_ports.tcp_ports)
.copied()
.collect::<HashSet<u16>>();
let diff_udp = self
.last_ports
.udp_ports
.difference(&curr_opened_ports.udp_ports)
.copied()
.collect::<HashSet<u16>>();
let ports_to_open = messages::PublicExposedPorts {
tcp_ports: diff_tcp,
udp_ports: diff_udp
udp_ports: diff_udp,
};
fw::open_ports(&self.ipt, ports_to_open)?;
return Ok(());
}
}

34
src/igd_actor.rs
View file

@ -1,16 +1,14 @@
use std::net::SocketAddrV4;
use anyhow::{Result, Context};
use anyhow::{Context, Result};
use igd::aio::*;
use igd::PortMappingProtocol;
use log::*;
use tokio::{
select,
sync::watch,
time::{
Duration,
self,
}};
time::{self, Duration},
};
use crate::config::RuntimeConfigIgd;
use crate::messages;
@ -26,7 +24,10 @@ pub struct IgdActor {
}
impl IgdActor {
pub async fn new(config: Option<RuntimeConfigIgd>, rxp: &watch::Receiver<messages::PublicExposedPorts>) -> Result<Option<Self>> {
pub async fn new(
config: Option<RuntimeConfigIgd>,
rxp: &watch::Receiver<messages::PublicExposedPorts>,
quentin commented 2021-09-20 16:12:30 +00:00
Outdated
Review
Copy link

Same as my previous comment

Same as my previous comment
) -> Result<Option<Self>> {
if config.is_none() {
return Ok(None);
}
@ -60,7 +61,9 @@ impl IgdActor {
};
// 2. Update last ports if needed
if let Some(p) = new_ports { self.last_ports = p; }
if let Some(p) = new_ports {
self.last_ports = p;
}
// 3. Flush IGD requests
match self.do_igd().await {
@ -73,14 +76,25 @@ impl IgdActor {
pub async fn do_igd(&self) -> Result<()> {
let actions = [
(PortMappingProtocol::TCP, &self.last_ports.tcp_ports),
(PortMappingProtocol::UDP, &self.last_ports.udp_ports)
(PortMappingProtocol::UDP, &self.last_ports.udp_ports),
];
for (proto, list) in actions.iter() {
for port in *list {
let service_str = format!("{}:{}", self.private_ip, port);
let service = service_str.parse::<SocketAddrV4>().context("Invalid socket address")?;
self.gateway.add_port(*proto, *port, service, self.expire.as_secs() as u32, "diplonat").await?;
let service = service_str
.parse::<SocketAddrV4>()
.context("Invalid socket address")?;
self
.gateway
.add_port(
*proto,
*port,
service,
self.expire.as_secs() as u32,
"diplonat",
)
.await?;
debug!("IGD request successful for {:#?} {}", proto, service);
}
}

2
src/main.rs
View file

@ -7,8 +7,8 @@ mod fw_actor;
mod igd_actor;
mod messages;
use log::*;
use diplonat::Diplonat;
use log::*;
#[tokio::main]
async fn main() {

6
src/messages.rs
View file

@ -3,14 +3,14 @@ use std::collections::HashSet;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PublicExposedPorts {
pub tcp_ports: HashSet<u16>,
pub udp_ports: HashSet<u16>
pub udp_ports: HashSet<u16>,
}
impl PublicExposedPorts {
pub fn new() -> Self {
return Self {
tcp_ports: HashSet::new(),
udp_ports: HashSet::new()
}
udp_ports: HashSet::new(),
};
}
}