From 76fe63791baea5577bde3f83243f85ba781f46fe Mon Sep 17 00:00:00 2001 From: adrien Date: Wed, 25 Aug 2021 17:20:31 +0200 Subject: [PATCH 01/10] Rewrote the configuration options to make Diplonat modular: IGD and Firewall modules will only be enabled if correponding `DIPLONAT_*_ENABLE` is set. (Breaking changes!) Also rearranged imports throughout the project, because I couldn't resist, sorry. --- docker-compose.yml | 9 ++-- src/config/mod.rs | 2 +- src/config/options.rs | 88 +++++++++++++++++++++++++------------- src/config/options_test.rs | 87 +++++++++++++++++++++++-------------- src/config/runtime.rs | 75 ++++++++++++++++++-------------- src/consul.rs | 3 +- src/consul_actor.rs | 22 ++++++---- src/diplonat.rs | 55 ++++++++++++++++-------- src/fw.rs | 8 ++-- src/fw_actor.rs | 32 +++++++++----- src/igd_actor.rs | 37 ++++++++++------ 11 files changed, 263 insertions(+), 155 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 0780c86..ab0dd92 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -5,10 +5,13 @@ services: image: darkgallium/amd64_diplonat:v2 network_mode: host # required by UPNP/IGD environment: - DIPLONAT_PRIVATE_IP: 192.168.0.18 - DIPLONAT_REFRESH_TIME: 60 - DIPLONAT_EXPIRATION_TIME: 300 DIPLONAT_CONSUL_NODE_NAME: lheureduthe + DIPLONAT_FIREWALL_ENABLE: true + DIPLONAT_FIREWALL_REFRESH_TIME: 60 + DIPLONAT_IGD_ENABLE: true + DIPLONAT_IGD_PRIVATE_IP: 192.168.0.18 + DIPLONAT_IGD_EXPIRATION_TIME: 300 + DIPLONAT_IGD_REFRESH_TIME: 60 RUST_LOG: debug diff --git a/src/config/mod.rs b/src/config/mod.rs index 14926bd..024e114 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -3,7 +3,7 @@ mod options; mod options_test; mod runtime; -pub use options::{ConfigOpts, ConfigOptsAcme, ConfigOptsBase, ConfigOptsConsul}; +pub use options::{ConfigOpts, ConfigOptsConsul, ConfigOptsAcme, ConfigOptsFirewall, ConfigOptsIgd}; pub use runtime::{RuntimeConfig, RuntimeConfigAcme, RuntimeConfigConsul, RuntimeConfigFirewall, RuntimeConfigIgd}; pub const EXPIRATION_TIME: u16 = 300; diff --git a/src/config/options.rs b/src/config/options.rs index 36da475..b3a63b0 100644 --- a/src/config/options.rs +++ b/src/config/options.rs @@ -8,27 +8,13 @@ use crate::config::RuntimeConfig; // This file parses the options that can be declared in the environment. // runtime.rs applies business logic and builds RuntimeConfig structs. -/// Base configuration options -#[derive(Clone, Default, Deserialize)] -pub struct ConfigOptsBase { - /// This node's private IP address [default: None] - pub private_ip: Option, - /// Expiration time for IGD rules [default: 60] - pub expiration_time: Option, - /// Refresh time for IGD and Firewall rules [default: 300] - pub refresh_time: Option, -} +// - Note for the future - +// There is no *need* to have a 'DIPLONAT_XXX_*' prefix for all config options. +// If some config options are shared by several modules, a ConfigOptsBase could +// contain them, and parse the 'DIPLONAT_*' prefix directly. +// Only in runtime.rs would these options find their proper location in each +// module's struct. -/// ACME configuration options -#[derive(Clone, Default, Deserialize)] -pub struct ConfigOptsAcme { - /// Whether ACME is enabled [default: false] - #[serde(default)] - pub enable: bool, - - /// The default domain holder's e-mail [default: None] - pub email: Option, -} /// Consul configuration options #[derive(Clone, Default, Deserialize)] @@ -39,23 +25,63 @@ pub struct ConfigOptsConsul { pub url: Option, } +/// ACME configuration options +#[derive(Clone, Default, Deserialize)] +pub struct ConfigOptsAcme { + /// Whether the ACME module is enabled [default: false] + #[serde(default)] + pub enable: bool, + + /// The default domain holder's e-mail [default: None] + pub email: Option, +} + +/// Firewall configuration options +#[derive(Clone, Default, Deserialize)] +pub struct ConfigOptsFirewall { + /// Whether the firewall module is enabled [default: false] + #[serde(default)] + pub enable: bool, + + /// Refresh time for firewall rules [default: 300] + pub refresh_time: Option, +} + +/// IGD configuration options +#[derive(Clone, Default, Deserialize)] +pub struct ConfigOptsIgd { + /// Whether the IGD module is enabled [default: false] + #[serde(default)] + pub enable: bool, + + /// This node's private IP address [default: None] + pub private_ip: Option, + /// Expiration time for IGD rules [default: 60] + pub expiration_time: Option, + /// Refresh time for IGD rules [default: 300] + pub refresh_time: Option, +} + /// Model of all potential configuration options pub struct ConfigOpts { - pub base: ConfigOptsBase, - pub acme: ConfigOptsAcme, pub consul: ConfigOptsConsul, + pub acme: ConfigOptsAcme, + pub firewall: ConfigOptsFirewall, + pub igd: ConfigOptsIgd, } impl ConfigOpts { pub fn from_env() -> Result { - let base: ConfigOptsBase = envy::prefixed("DIPLONAT_").from_env()?; let consul: ConfigOptsConsul = envy::prefixed("DIPLONAT_CONSUL_").from_env()?; let acme: ConfigOptsAcme = envy::prefixed("DIPLONAT_ACME_").from_env()?; + let firewall: ConfigOptsFirewall = envy::prefixed("DIPLONAT_FIREWALL_").from_env()?; + let igd: ConfigOptsIgd = envy::prefixed("DIPLONAT_IGD_").from_env()?; RuntimeConfig::new(Self { - base: base, - consul: consul, - acme: acme, + consul, + acme, + firewall, + igd, }) } @@ -63,14 +89,16 @@ impl ConfigOpts { #[allow(dead_code)] pub fn from_iter(iter: Iter) -> Result where Iter: IntoIterator { - let base: ConfigOptsBase = envy::prefixed("DIPLONAT_").from_iter(iter.clone())?; 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 igd: ConfigOptsIgd = envy::prefixed("DIPLONAT_IGD_").from_iter(iter.clone())?; RuntimeConfig::new(Self { - base: base, - consul: consul, - acme: acme, + consul, + acme, + firewall, + igd, }) } } \ No newline at end of file diff --git a/src/config/options_test.rs b/src/config/options_test.rs index a6063fd..98fc625 100644 --- a/src/config/options_test.rs +++ b/src/config/options_test.rs @@ -11,35 +11,44 @@ use crate::config::*; fn minimal_valid_options() -> HashMap { let mut opts = HashMap::new(); - opts.insert("DIPLONAT_PRIVATE_IP".to_string(), "172.123.43.555".to_string()); opts.insert("DIPLONAT_CONSUL_NODE_NAME".to_string(), "consul_node".to_string()); opts } fn all_valid_options() -> HashMap { let mut opts = minimal_valid_options(); - opts.insert("DIPLONAT_EXPIRATION_TIME".to_string(), "30".to_string()); - opts.insert("DIPLONAT_REFRESH_TIME".to_string(), "10".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_FIREWALL_ENABLE".to_string(), "true".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_EXPIRATION_TIME".to_string(), "60".to_string()); + opts.insert("DIPLONAT_IGD_REFRESH_TIME".to_string(), "10".to_string()); opts } +// #[test] +// #[should_panic] +// fn err_empty_env() { +// std::env::remove_var("DIPLONAT_CONSUL_NODE_NAME"); +// ConfigOpts::from_env().unwrap(); +// } + #[test] #[should_panic] fn err_empty_env() { - std::env::remove_var("DIPLONAT_PRIVATE_IP"); std::env::remove_var("DIPLONAT_CONSUL_NODE_NAME"); - ConfigOpts::from_env().unwrap(); + let opts: HashMap = HashMap::new(); + ConfigOpts::from_iter(opts).unwrap(); } #[test] -fn ok_from_iter_minimal_valid_options() { +fn ok_minimal_valid_options() { let opts = minimal_valid_options(); let rt_config = ConfigOpts::from_iter(opts.clone()).unwrap(); - assert!(rt_config.acme.is_none()); assert_eq!( &rt_config.consul.node_name, opts.get(&"DIPLONAT_CONSUL_NODE_NAME".to_string()).unwrap() @@ -48,7 +57,10 @@ fn ok_from_iter_minimal_valid_options() { rt_config.consul.url, CONSUL_URL.to_string() ); - assert_eq!( + assert!(rt_config.acme.is_none()); + assert!(rt_config.firewall.is_none()); + assert!(rt_config.igd.is_none()); + /*assert_eq!( rt_config.firewall.refresh_time, Duration::from_secs(REFRESH_TIME.into()) ); @@ -63,36 +75,37 @@ fn ok_from_iter_minimal_valid_options() { assert_eq!( rt_config.igd.refresh_time, Duration::from_secs(REFRESH_TIME.into()) - ); + );*/ } #[test] #[should_panic] -fn err_from_iter_invalid_refresh_time() { +fn err_invalid_igd_options() { let mut opts = minimal_valid_options(); - opts.insert("DIPLONAT_EXPIRATION_TIME".to_string(), "60".to_string()); - opts.insert("DIPLONAT_REFRESH_TIME".to_string(), "60".to_string()); + opts.insert("DIPLONAT_IGD_ENABLE".to_string(), "true".to_string()); + opts.insert("DIPLONAT_IGD_EXPIRATION_TIME".to_string(), "60".to_string()); + opts.insert("DIPLONAT_IGD_REFRESH_TIME".to_string(), "60".to_string()); ConfigOpts::from_iter(opts).unwrap(); } #[test] -fn ok_from_iter_all_valid_options() { +fn ok_all_valid_options() { let opts = all_valid_options(); let rt_config = ConfigOpts::from_iter(opts.clone()).unwrap(); - let expiration_time = Duration::from_secs( - opts.get(&"DIPLONAT_EXPIRATION_TIME".to_string()).unwrap() + let firewall_refresh_time = Duration::from_secs( + opts.get(&"DIPLONAT_FIREWALL_REFRESH_TIME".to_string()).unwrap() .parse::().unwrap() .into()); - let refresh_time = Duration::from_secs( - opts.get(&"DIPLONAT_REFRESH_TIME".to_string()).unwrap() + let igd_expiration_time = Duration::from_secs( + opts.get(&"DIPLONAT_IGD_EXPIRATION_TIME".to_string()).unwrap() + .parse::().unwrap() + .into()); + let igd_refresh_time = Duration::from_secs( + opts.get(&"DIPLONAT_IGD_REFRESH_TIME".to_string()).unwrap() .parse::().unwrap() .into()); - assert!(rt_config.acme.is_some()); - assert_eq!( - &rt_config.acme.unwrap().email, - opts.get(&"DIPLONAT_ACME_EMAIL".to_string()).unwrap()); assert_eq!( &rt_config.consul.node_name, opts.get(&"DIPLONAT_CONSUL_NODE_NAME".to_string()).unwrap() @@ -101,20 +114,32 @@ fn ok_from_iter_all_valid_options() { &rt_config.consul.url, opts.get(&"DIPLONAT_CONSUL_URL".to_string()).unwrap() ); + + assert!(rt_config.acme.is_some()); + let acme = rt_config.acme.unwrap(); assert_eq!( - rt_config.firewall.refresh_time, - refresh_time + &acme.email, + 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!(rt_config.igd.is_some()); + let igd = rt_config.igd.unwrap(); + assert_eq!( + &igd.private_ip, + opts.get(&"DIPLONAT_IGD_PRIVATE_IP".to_string()).unwrap() ); assert_eq!( - &rt_config.igd.private_ip, - opts.get(&"DIPLONAT_PRIVATE_IP".to_string()).unwrap() + igd.expiration_time, + igd_expiration_time ); assert_eq!( - rt_config.igd.expiration_time, - expiration_time - ); - assert_eq!( - rt_config.igd.refresh_time, - refresh_time + igd.refresh_time, + igd_refresh_time ); } \ No newline at end of file diff --git a/src/config/runtime.rs b/src/config/runtime.rs index 58c86b9..f83a6b5 100644 --- a/src/config/runtime.rs +++ b/src/config/runtime.rs @@ -2,17 +2,14 @@ use std::time::Duration; use anyhow::{Result, anyhow}; -use crate::config::{ConfigOpts, ConfigOptsAcme, ConfigOptsBase, ConfigOptsConsul}; +use crate::config::{ConfigOpts, ConfigOptsConsul, ConfigOptsAcme, ConfigOptsFirewall, ConfigOptsIgd}; // This code is inspired by the Trunk crate (https://github.com/thedodd/trunk) // In this file, we take ConfigOpts and transform them into ready-to-use RuntimeConfig. // We apply default values and business logic. -#[derive(Debug)] -pub struct RuntimeConfigAcme { - pub email: String, -} +// Consul config is mandatory, all the others are optional. #[derive(Debug)] pub struct RuntimeConfigConsul { @@ -20,6 +17,11 @@ pub struct RuntimeConfigConsul { pub url: String, } +#[derive(Debug)] +pub struct RuntimeConfigAcme { + pub email: String, +} + #[derive(Debug)] pub struct RuntimeConfigFirewall { pub refresh_time: Duration, @@ -34,18 +36,18 @@ pub struct RuntimeConfigIgd { #[derive(Debug)] pub struct RuntimeConfig { - pub acme: Option, pub consul: RuntimeConfigConsul, - pub firewall: RuntimeConfigFirewall, - pub igd: RuntimeConfigIgd, + pub acme: Option, + pub firewall: Option, + pub igd: Option, } impl RuntimeConfig { pub fn new(opts: ConfigOpts) -> Result { - let acme = RuntimeConfigAcme::new(opts.acme.clone())?; let consul = RuntimeConfigConsul::new(opts.consul.clone())?; - let firewall = RuntimeConfigFirewall::new(opts.base.clone())?; - let igd = RuntimeConfigIgd::new(opts.base.clone())?; + let acme = RuntimeConfigAcme::new(opts.acme.clone())?; + let firewall = RuntimeConfigFirewall::new(opts.firewall.clone())?; + let igd = RuntimeConfigIgd::new(opts.igd.clone())?; Ok(Self { acme, @@ -56,6 +58,19 @@ impl RuntimeConfig { } } +impl RuntimeConfigConsul { + pub(super) fn new(opts: ConfigOptsConsul) -> Result { + let node_name = opts.node_name.expect( + "'DIPLONAT_CONSUL_NODE_NAME' is required"); + let url = opts.url.unwrap_or(super::CONSUL_URL.to_string()); + + Ok(Self { + node_name, + url, + }) + } +} + impl RuntimeConfigAcme { pub fn new(opts: ConfigOptsAcme) -> Result> { if !opts.enable { @@ -63,8 +78,7 @@ impl RuntimeConfigAcme { } let email = opts.email.expect( - "'DIPLONAT_ACME_EMAIL' environment variable is required \ - if 'DIPLONAT_ACME_ENABLE' == 'true'"); + "'DIPLONAT_ACME_EMAIL' is required if ACME is enabled"); Ok(Some(Self { email, @@ -72,34 +86,29 @@ impl RuntimeConfigAcme { } } -impl RuntimeConfigConsul { - pub(super) fn new(opts: ConfigOptsConsul) -> Result { - let node_name = opts.node_name.expect( - "'DIPLONAT_CONSUL_NODE_NAME' environment variable is required"); - let url = opts.url.unwrap_or(super::CONSUL_URL.to_string()); - - Ok(Self { - node_name, - url, - }) - } -} - impl RuntimeConfigFirewall { - pub(super) fn new(opts: ConfigOptsBase) -> Result { + pub(super) fn new(opts: ConfigOptsFirewall) -> Result> { + if !opts.enable { + return Ok(None); + } + let refresh_time = Duration::from_secs( opts.refresh_time.unwrap_or(super::REFRESH_TIME).into()); - Ok(Self { + Ok(Some(Self { refresh_time, - }) + })) } } impl RuntimeConfigIgd { - pub(super) fn new(opts: ConfigOptsBase) -> Result { + pub(super) fn new(opts: ConfigOptsIgd) -> Result> { + if !opts.enable { + return Ok(None); + } + let private_ip = opts.private_ip.expect( - "'DIPLONAT_PRIVATE_IP' environment variable is required"); + "'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( @@ -112,10 +121,10 @@ impl RuntimeConfigIgd { refresh_time.as_secs())); } - Ok(Self { + Ok(Some(Self { private_ip, expiration_time, refresh_time, - }) + })) } } \ No newline at end of file diff --git a/src/consul.rs b/src/consul.rs index 1bb30aa..9a91782 100644 --- a/src/consul.rs +++ b/src/consul.rs @@ -1,6 +1,7 @@ -use serde::{Serialize, Deserialize}; use std::collections::HashMap; + use anyhow::{Result, anyhow}; +use serde::{Serialize, Deserialize}; #[derive(Serialize, Deserialize, Debug)] pub struct ServiceEntry { diff --git a/src/consul_actor.rs b/src/consul_actor.rs index ba5d704..dcbd79e 100644 --- a/src/consul_actor.rs +++ b/src/consul_actor.rs @@ -1,14 +1,17 @@ use std::cmp; +use std::collections::HashSet; use std::time::Duration; -use log::*; -use tokio::sync::watch; -use tokio::time::delay_for; + use anyhow::Result; +use log::*; use serde::{Serialize, Deserialize}; use serde_lexpr::{from_str,error}; -use crate::messages; +use tokio::sync::watch; +use tokio::time::delay_for; + +use crate::config::RuntimeConfigConsul; use crate::consul; -use std::collections::HashSet; +use crate::messages; #[derive(Serialize, Deserialize, Debug)] pub enum DiplonatParameter { @@ -27,6 +30,7 @@ pub struct ConsulActor { consul: consul::Consul, node: String, retries: u32, + tx_open_ports: watch::Sender } @@ -72,18 +76,18 @@ fn to_open_ports(params: &Vec) -> messages::PublicExposedPorts { } impl ConsulActor { - pub fn new(url: &str, node: &str) -> Self { + pub fn new(config: RuntimeConfigConsul) -> Self { let (tx, rx) = watch::channel(messages::PublicExposedPorts{ tcp_ports: HashSet::new(), udp_ports: HashSet::new() }); return Self { - consul: consul::Consul::new(url), + consul: consul::Consul::new(&config.url), + node: config.node_name, + retries: 0, rx_open_ports: rx, tx_open_ports: tx, - node: node.to_string(), - retries: 0, }; } diff --git a/src/diplonat.rs b/src/diplonat.rs index 7049530..6334e5b 100644 --- a/src/diplonat.rs +++ b/src/diplonat.rs @@ -1,4 +1,4 @@ -use anyhow::Result; +use anyhow::{Result, anyhow}; use tokio::try_join; use crate::config::ConfigOpts; @@ -8,43 +8,60 @@ use crate::igd_actor::IgdActor; pub struct Diplonat { consul: ConsulActor, - firewall: FirewallActor, - igd: IgdActor, + + firewall: Option, + igd: Option, } impl Diplonat { pub async fn new() -> Result { - let rt_cfg = ConfigOpts::from_env()?; - println!("{:#?}", rt_cfg); + let config = ConfigOpts::from_env()?; + println!("{:#?}", config); - let ca = ConsulActor::new(&rt_cfg.consul.url, &rt_cfg.consul.node_name); + let consul_actor = ConsulActor::new(config.consul); - let fw = FirewallActor::new( - rt_cfg.firewall.refresh_time, - &ca.rx_open_ports + let firewall_actor = FirewallActor::new( + config.firewall, + &consul_actor.rx_open_ports ).await?; - let ia = IgdActor::new( - &rt_cfg.igd.private_ip, - rt_cfg.igd.refresh_time, - rt_cfg.igd.expiration_time, - &ca.rx_open_ports + 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!")); + } + let ctx = Self { - consul: ca, - igd: ia, - firewall: fw + consul: consul_actor, + firewall: firewall_actor, + igd: igd_actor, }; return Ok(ctx); } pub async fn listen(&mut self) -> Result<()> { + let firewall = &mut self.firewall; + let igd = &mut self.igd; + try_join!( self.consul.listen(), - self.igd.listen(), - self.firewall.listen() + async { + match firewall { + Some(x) => x.listen().await, + None => Ok(()) + } + }, + async { + match igd { + Some(x) => x.listen().await, + None => Ok(()) + } + }, )?; return Ok(()); diff --git a/src/fw.rs b/src/fw.rs index bc4d119..a71dd1c 100644 --- a/src/fw.rs +++ b/src/fw.rs @@ -1,9 +1,11 @@ -use iptables; -use regex::Regex; use std::collections::HashSet; -use crate::messages; + use anyhow::{Result,Context}; +use iptables; use log::*; +use regex::Regex; + +use crate::messages; pub fn setup(ipt: &iptables::IPTables) -> Result<()> { diff --git a/src/fw_actor.rs b/src/fw_actor.rs index b5e4c7e..29e6473 100644 --- a/src/fw_actor.rs +++ b/src/fw_actor.rs @@ -1,37 +1,47 @@ +use std::collections::HashSet; + use anyhow::Result; +use iptables; +use log::*; use tokio::{ select, sync::watch, time::{ + Duration, self, - Duration }}; -use log::*; -use iptables; -use crate::messages; +use crate::config::RuntimeConfigFirewall; use crate::fw; -use std::collections::HashSet; +use crate::messages; + pub struct FirewallActor { pub ipt: iptables::IPTables, - rx_ports: watch::Receiver, + last_ports: messages::PublicExposedPorts, - refresh: Duration + refresh: Duration, + + rx_ports: watch::Receiver, } impl FirewallActor { - pub async fn new(_refresh: Duration, rxp: &watch::Receiver) -> Result { + pub async fn new(config: Option, rxp: &watch::Receiver) -> Result> { + if config.is_none() { + return Ok(None); + } + let config = config.unwrap(); + let ctx = Self { ipt: iptables::new(false)?, - rx_ports: rxp.clone(), last_ports: messages::PublicExposedPorts::new(), - refresh: _refresh, + refresh: config.refresh_time, + rx_ports: rxp.clone(), }; fw::setup(&ctx.ipt)?; - return Ok(ctx); + return Ok(Some(ctx)); } pub async fn listen(&mut self) -> Result<()> { diff --git a/src/igd_actor.rs b/src/igd_actor.rs index 55d9c5f..19a7e93 100644 --- a/src/igd_actor.rs +++ b/src/igd_actor.rs @@ -1,43 +1,52 @@ +use std::net::SocketAddrV4; + +use anyhow::{Result, Context}; use igd::aio::*; use igd::PortMappingProtocol; -use std::net::SocketAddrV4; use log::*; -use anyhow::{Result, Context}; use tokio::{ select, sync::watch, time::{ + Duration, self, - Duration }}; + +use crate::config::RuntimeConfigIgd; use crate::messages; pub struct IgdActor { - last_ports: messages::PublicExposedPorts, - rx_ports: watch::Receiver, - gateway: Gateway, - refresh: Duration, expire: Duration, - private_ip: String + gateway: Gateway, + last_ports: messages::PublicExposedPorts, + private_ip: String, + refresh: Duration, + + rx_ports: watch::Receiver, } impl IgdActor { - pub async fn new(priv_ip: &str, refresh: Duration, expire: Duration, rxp: &watch::Receiver) -> Result { + pub async fn new(config: Option, rxp: &watch::Receiver) -> Result> { + if config.is_none() { + return Ok(None); + } + let config = config.unwrap(); + let gw = search_gateway(Default::default()) .await .context("Failed to find IGD gateway")?; info!("IGD gateway: {}", gw); let ctx = Self { + expire: config.expiration_time, gateway: gw, + last_ports: messages::PublicExposedPorts::new(), + private_ip: config.private_ip, + refresh: config.refresh_time, rx_ports: rxp.clone(), - private_ip: priv_ip.to_string(), - refresh: refresh, - expire: expire, - last_ports: messages::PublicExposedPorts::new() }; - return Ok(ctx); + return Ok(Some(ctx)); } pub async fn listen(&mut self) -> Result<()> { -- 2.43.0 From 195aec2cfe738f4025ea540d2591f876e1d209b9 Mon Sep 17 00:00:00 2001 From: adrien Date: Thu, 26 Aug 2021 16:16:15 +0200 Subject: [PATCH 02/10] updated README.md --- README.md | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 3be85a3..6f3b7f4 100644 --- a/README.md +++ b/README.md @@ -40,10 +40,13 @@ cargo build consul agent -dev # in a separate terminal # adapt following values to your configuration -export DIPLONAT_PRIVATE_IP="192.168.0.18" -export DIPLONAT_REFRESH_TIME="60" -export DIPLONAT_EXPIRATION_TIME="300" export DIPLONAT_CONSUL_NODE_NAME="lheureduthe" +export DIPLONAT_FIREWALL_ENABLE="true" +export DIPLONAT_FIREWALL_REFRESH_TIME="300" +export DIPLONAT_IGD_ENABLE="true" +export DIPLONAT_IGD_PRIVATE_IP="192.168.0.18" +export DIPLONAT_IGD_REFRESH_TIME="60" +export DIPLONAT_IGD_EXPIRATION_TIME="300" export RUST_LOG=debug cargo run ``` -- 2.43.0 From 2e797d2f62c82f59ad3f3601392192e0ff90e8df Mon Sep 17 00:00:00 2001 From: Quentin Dufour Date: Mon, 20 Sep 2021 11:30:00 +0200 Subject: [PATCH 03/10] Local code formatting --- .rustfmt.toml | 2 + src/config/mod.rs | 10 +- src/config/options.rs | 116 ++++++++++---------- src/config/options_test.rs | 216 ++++++++++++++++++++----------------- src/config/runtime.rs | 148 ++++++++++++------------- src/consul.rs | 18 ++-- src/consul_actor.rs | 29 +++-- src/diplonat.rs | 23 ++-- src/fw.rs | 122 ++++++++++++--------- src/fw_actor.rs | 44 +++++--- src/igd_actor.rs | 46 +++++--- src/main.rs | 2 +- src/messages.rs | 6 +- 13 files changed, 424 insertions(+), 358 deletions(-) create mode 100644 .rustfmt.toml diff --git a/.rustfmt.toml b/.rustfmt.toml new file mode 100644 index 0000000..f0fa07f --- /dev/null +++ b/.rustfmt.toml @@ -0,0 +1,2 @@ +hard_tabs = false +tab_spaces = 2 diff --git a/src/config/mod.rs b/src/config/mod.rs index 024e114..9dddd04 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -3,9 +3,13 @@ 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; -pub const CONSUL_URL: &str = "http://127.0.0.1:8500"; \ No newline at end of file +pub const CONSUL_URL: &str = "http://127.0.0.1:8500"; diff --git a/src/config/options.rs b/src/config/options.rs index b3a63b0..638a2f3 100644 --- a/src/config/options.rs +++ b/src/config/options.rs @@ -12,93 +12,95 @@ use crate::config::RuntimeConfig; // There is no *need* to have a 'DIPLONAT_XXX_*' prefix for all config options. // If some config options are shared by several modules, a ConfigOptsBase could // contain them, and parse the 'DIPLONAT_*' prefix directly. -// Only in runtime.rs would these options find their proper location in each +// 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 { - /// Consul's node name [default: None] - pub node_name: Option, - /// Consul's REST URL [default: "http://127.0.0.1:8500"] - pub url: Option, + /// Consul's node name [default: None] + pub node_name: Option, + /// Consul's REST URL [default: "http://127.0.0.1:8500"] + pub url: Option, } /// ACME configuration options #[derive(Clone, Default, Deserialize)] pub struct ConfigOptsAcme { - /// Whether the ACME module is enabled [default: false] - #[serde(default)] - pub enable: bool, + /// Whether the ACME module is enabled [default: false] + #[serde(default)] + pub enable: bool, - /// The default domain holder's e-mail [default: None] - pub email: Option, + /// The default domain holder's e-mail [default: None] + pub email: Option, } /// Firewall configuration options #[derive(Clone, Default, Deserialize)] pub struct ConfigOptsFirewall { - /// Whether the firewall module is enabled [default: false] - #[serde(default)] - pub enable: bool, + /// Whether the firewall module is enabled [default: false] + #[serde(default)] + pub enable: bool, - /// Refresh time for firewall rules [default: 300] - pub refresh_time: Option, + /// Refresh time for firewall rules [default: 300] + pub refresh_time: Option, } /// IGD configuration options #[derive(Clone, Default, Deserialize)] pub struct ConfigOptsIgd { - /// Whether the IGD module is enabled [default: false] - #[serde(default)] - pub enable: bool, + /// Whether the IGD module is enabled [default: false] + #[serde(default)] + pub enable: bool, - /// This node's private IP address [default: None] - pub private_ip: Option, - /// Expiration time for IGD rules [default: 60] - pub expiration_time: Option, - /// Refresh time for IGD rules [default: 300] - pub refresh_time: Option, + /// This node's private IP address [default: None] + pub private_ip: Option, + /// Expiration time for IGD rules [default: 60] + pub expiration_time: Option, + /// Refresh time for IGD rules [default: 300] + pub refresh_time: Option, } /// Model of all potential configuration options pub struct ConfigOpts { - pub consul: ConfigOptsConsul, - pub acme: ConfigOptsAcme, - pub firewall: ConfigOptsFirewall, - pub igd: ConfigOptsIgd, + pub consul: ConfigOptsConsul, + pub acme: ConfigOptsAcme, + pub firewall: ConfigOptsFirewall, + pub igd: ConfigOptsIgd, } impl ConfigOpts { - pub fn from_env() -> Result { - let consul: ConfigOptsConsul = envy::prefixed("DIPLONAT_CONSUL_").from_env()?; - let acme: ConfigOptsAcme = envy::prefixed("DIPLONAT_ACME_").from_env()?; - let firewall: ConfigOptsFirewall = envy::prefixed("DIPLONAT_FIREWALL_").from_env()?; - let igd: ConfigOptsIgd = envy::prefixed("DIPLONAT_IGD_").from_env()?; + pub fn from_env() -> Result { + let consul: ConfigOptsConsul = envy::prefixed("DIPLONAT_CONSUL_").from_env()?; + let acme: ConfigOptsAcme = envy::prefixed("DIPLONAT_ACME_").from_env()?; + let firewall: ConfigOptsFirewall = envy::prefixed("DIPLONAT_FIREWALL_").from_env()?; + let igd: ConfigOptsIgd = envy::prefixed("DIPLONAT_IGD_").from_env()?; - RuntimeConfig::new(Self { - consul, - acme, - firewall, - igd, - }) - } + RuntimeConfig::new(Self { + consul, + acme, + firewall, + igd, + }) + } - // Currently only used in tests - #[allow(dead_code)] - pub fn from_iter(iter: Iter) -> Result - where Iter: IntoIterator { - 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 igd: ConfigOptsIgd = envy::prefixed("DIPLONAT_IGD_").from_iter(iter.clone())?; + // Currently only used in tests + #[allow(dead_code)] + pub fn from_iter(iter: Iter) -> Result + where + Iter: IntoIterator, + { + 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 igd: ConfigOptsIgd = envy::prefixed("DIPLONAT_IGD_").from_iter(iter.clone())?; - RuntimeConfig::new(Self { - consul, - acme, - firewall, - igd, - }) - } -} \ No newline at end of file + RuntimeConfig::new(Self { + consul, + acme, + firewall, + igd, + }) + } +} diff --git a/src/config/options_test.rs b/src/config/options_test.rs index 98fc625..0dfaaac 100644 --- a/src/config/options_test.rs +++ b/src/config/options_test.rs @@ -5,28 +5,43 @@ use crate::config::*; // Environment variables are set for the entire process and // tests are run whithin the same process. -// => We cannot test ConfigOpts::from_env(), +// => We cannot test ConfigOpts::from_env(), // because tests modify each other's environment. // This is why we only test ConfigOpts::from_iter(iter). fn minimal_valid_options() -> HashMap { - let mut opts = HashMap::new(); - opts.insert("DIPLONAT_CONSUL_NODE_NAME".to_string(), "consul_node".to_string()); - opts + let mut opts = HashMap::new(); + opts.insert( + "DIPLONAT_CONSUL_NODE_NAME".to_string(), + "consul_node".to_string(), + ); + opts } fn all_valid_options() -> HashMap { - let mut opts = minimal_valid_options(); - 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_FIREWALL_ENABLE".to_string(), "true".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_EXPIRATION_TIME".to_string(), "60".to_string()); - opts.insert("DIPLONAT_IGD_REFRESH_TIME".to_string(), "10".to_string()); - opts + let mut opts = minimal_valid_options(); + 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_FIREWALL_ENABLE".to_string(), "true".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_EXPIRATION_TIME".to_string(), "60".to_string()); + opts.insert("DIPLONAT_IGD_REFRESH_TIME".to_string(), "10".to_string()); + opts } // #[test] @@ -39,107 +54,108 @@ fn all_valid_options() -> HashMap { #[test] #[should_panic] fn err_empty_env() { - std::env::remove_var("DIPLONAT_CONSUL_NODE_NAME"); - let opts: HashMap = HashMap::new(); - ConfigOpts::from_iter(opts).unwrap(); + std::env::remove_var("DIPLONAT_CONSUL_NODE_NAME"); + let opts: HashMap = HashMap::new(); + ConfigOpts::from_iter(opts).unwrap(); } #[test] fn ok_minimal_valid_options() { - let opts = minimal_valid_options(); - let rt_config = ConfigOpts::from_iter(opts.clone()).unwrap(); + let opts = minimal_valid_options(); + let rt_config = ConfigOpts::from_iter(opts.clone()).unwrap(); - assert_eq!( - &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!(rt_config.acme.is_none()); - assert!(rt_config.firewall.is_none()); - assert!(rt_config.igd.is_none()); - /*assert_eq!( - rt_config.firewall.refresh_time, - Duration::from_secs(REFRESH_TIME.into()) - ); - assert_eq!( - &rt_config.igd.private_ip, - opts.get(&"DIPLONAT_PRIVATE_IP".to_string()).unwrap() - ); - assert_eq!( - rt_config.igd.expiration_time, - Duration::from_secs(EXPIRATION_TIME.into()) - ); - assert_eq!( - rt_config.igd.refresh_time, - Duration::from_secs(REFRESH_TIME.into()) - );*/ + assert_eq!( + &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!(rt_config.acme.is_none()); + assert!(rt_config.firewall.is_none()); + assert!(rt_config.igd.is_none()); + /*assert_eq!( + rt_config.firewall.refresh_time, + Duration::from_secs(REFRESH_TIME.into()) + ); + assert_eq!( + &rt_config.igd.private_ip, + opts.get(&"DIPLONAT_PRIVATE_IP".to_string()).unwrap() + ); + assert_eq!( + rt_config.igd.expiration_time, + Duration::from_secs(EXPIRATION_TIME.into()) + ); + assert_eq!( + rt_config.igd.refresh_time, + Duration::from_secs(REFRESH_TIME.into()) + );*/ } #[test] #[should_panic] fn err_invalid_igd_options() { - let mut opts = minimal_valid_options(); - opts.insert("DIPLONAT_IGD_ENABLE".to_string(), "true".to_string()); - opts.insert("DIPLONAT_IGD_EXPIRATION_TIME".to_string(), "60".to_string()); - opts.insert("DIPLONAT_IGD_REFRESH_TIME".to_string(), "60".to_string()); - ConfigOpts::from_iter(opts).unwrap(); + let mut opts = minimal_valid_options(); + opts.insert("DIPLONAT_IGD_ENABLE".to_string(), "true".to_string()); + opts.insert("DIPLONAT_IGD_EXPIRATION_TIME".to_string(), "60".to_string()); + opts.insert("DIPLONAT_IGD_REFRESH_TIME".to_string(), "60".to_string()); + ConfigOpts::from_iter(opts).unwrap(); } #[test] fn ok_all_valid_options() { - let opts = all_valid_options(); - let rt_config = ConfigOpts::from_iter(opts.clone()).unwrap(); + let opts = 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::().unwrap() - .into()); - let igd_expiration_time = Duration::from_secs( - opts.get(&"DIPLONAT_IGD_EXPIRATION_TIME".to_string()).unwrap() - .parse::().unwrap() - .into()); - let igd_refresh_time = Duration::from_secs( - opts.get(&"DIPLONAT_IGD_REFRESH_TIME".to_string()).unwrap() - .parse::().unwrap() - .into()); + let firewall_refresh_time = Duration::from_secs( + opts + .get(&"DIPLONAT_FIREWALL_REFRESH_TIME".to_string()) + .unwrap() + .parse::() + .unwrap() + .into(), + ); + let igd_expiration_time = Duration::from_secs( + opts + .get(&"DIPLONAT_IGD_EXPIRATION_TIME".to_string()) + .unwrap() + .parse::() + .unwrap() + .into(), + ); + let igd_refresh_time = Duration::from_secs( + opts + .get(&"DIPLONAT_IGD_REFRESH_TIME".to_string()) + .unwrap() + .parse::() + .unwrap() + .into(), + ); - assert_eq!( - &rt_config.consul.node_name, - opts.get(&"DIPLONAT_CONSUL_NODE_NAME".to_string()).unwrap() - ); - assert_eq!( - &rt_config.consul.url, - opts.get(&"DIPLONAT_CONSUL_URL".to_string()).unwrap() - ); + assert_eq!( + &rt_config.consul.node_name, + opts.get(&"DIPLONAT_CONSUL_NODE_NAME".to_string()).unwrap() + ); + assert_eq!( + &rt_config.consul.url, + opts.get(&"DIPLONAT_CONSUL_URL".to_string()).unwrap() + ); - assert!(rt_config.acme.is_some()); - let acme = rt_config.acme.unwrap(); - assert_eq!( - &acme.email, - opts.get(&"DIPLONAT_ACME_EMAIL".to_string()).unwrap()); + assert!(rt_config.acme.is_some()); + let acme = rt_config.acme.unwrap(); + assert_eq!( + &acme.email, + 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!(rt_config.firewall.is_some()); + let firewall = rt_config.firewall.unwrap(); + assert_eq!(firewall.refresh_time, firewall_refresh_time); - assert!(rt_config.igd.is_some()); - let igd = rt_config.igd.unwrap(); - assert_eq!( - &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 - ); -} \ No newline at end of file + assert!(rt_config.igd.is_some()); + let igd = rt_config.igd.unwrap(); + assert_eq!( + &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); +} diff --git a/src/config/runtime.rs b/src/config/runtime.rs index f83a6b5..4e718a0 100644 --- a/src/config/runtime.rs +++ b/src/config/runtime.rs @@ -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) @@ -13,118 +15,116 @@ use crate::config::{ConfigOpts, ConfigOptsConsul, ConfigOptsAcme, ConfigOptsFire #[derive(Debug)] pub struct RuntimeConfigConsul { - pub node_name: String, - pub url: String, + pub node_name: String, + pub url: String, } #[derive(Debug)] pub struct RuntimeConfigAcme { - pub email: String, + pub email: String, } #[derive(Debug)] pub struct RuntimeConfigFirewall { - pub refresh_time: Duration, + pub refresh_time: Duration, } #[derive(Debug)] pub struct RuntimeConfigIgd { - pub private_ip: String, - pub expiration_time: Duration, - pub refresh_time: Duration, + pub private_ip: String, + pub expiration_time: Duration, + pub refresh_time: Duration, } #[derive(Debug)] pub struct RuntimeConfig { - pub consul: RuntimeConfigConsul, - pub acme: Option, - pub firewall: Option, - pub igd: Option, + pub consul: RuntimeConfigConsul, + pub acme: Option, + pub firewall: Option, + pub igd: Option, } impl RuntimeConfig { - pub fn new(opts: ConfigOpts) -> Result { - let consul = RuntimeConfigConsul::new(opts.consul.clone())?; - let acme = RuntimeConfigAcme::new(opts.acme.clone())?; - let firewall = RuntimeConfigFirewall::new(opts.firewall.clone())?; - let igd = RuntimeConfigIgd::new(opts.igd.clone())?; + pub fn new(opts: ConfigOpts) -> Result { + let consul = RuntimeConfigConsul::new(opts.consul.clone())?; + let acme = RuntimeConfigAcme::new(opts.acme.clone())?; + let firewall = RuntimeConfigFirewall::new(opts.firewall.clone())?; + let igd = RuntimeConfigIgd::new(opts.igd.clone())?; - Ok(Self { - acme, - consul, - firewall, - igd, - }) - } + Ok(Self { + acme, + consul, + firewall, + igd, + }) + } } impl RuntimeConfigConsul { - pub(super) fn new(opts: ConfigOptsConsul) -> Result { - let node_name = opts.node_name.expect( - "'DIPLONAT_CONSUL_NODE_NAME' is required"); - let url = opts.url.unwrap_or(super::CONSUL_URL.to_string()); + pub(super) fn new(opts: ConfigOptsConsul) -> Result { + let node_name = opts + .node_name + .expect("'DIPLONAT_CONSUL_NODE_NAME' is required"); + let url = opts.url.unwrap_or(super::CONSUL_URL.to_string()); - Ok(Self { - node_name, - url, - }) - } + Ok(Self { node_name, url }) + } } impl RuntimeConfigAcme { - pub fn new(opts: ConfigOptsAcme) -> Result> { - if !opts.enable { - return Ok(None); - } + pub fn new(opts: ConfigOptsAcme) -> Result> { + if !opts.enable { + 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"); - Ok(Some(Self { - email, - })) - } + Ok(Some(Self { email })) + } } impl RuntimeConfigFirewall { - pub(super) fn new(opts: ConfigOptsFirewall) -> Result> { - if !opts.enable { - return Ok(None); - } + pub(super) fn new(opts: ConfigOptsFirewall) -> Result> { + if !opts.enable { + 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 })) + } } impl RuntimeConfigIgd { - pub(super) fn new(opts: ConfigOptsIgd) -> Result> { - if !opts.enable { - return Ok(None); - } + pub(super) fn new(opts: ConfigOptsIgd) -> Result> { + if !opts.enable { + return Ok(None); + } - 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()); + 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()); - if refresh_time.as_secs() * 2 > expiration_time.as_secs() { - return Err(anyhow!( + if refresh_time.as_secs() * 2 > expiration_time.as_secs() { + return Err(anyhow!( "IGD expiration time (currently: {}s) must be at least twice bigger than refresh time (currently: {}s)", expiration_time.as_secs(), refresh_time.as_secs())); - } + } - Ok(Some(Self { - private_ip, - expiration_time, - refresh_time, - })) - } -} \ No newline at end of file + Ok(Some(Self { + private_ip, + expiration_time, + refresh_time, + })) + } +} diff --git a/src/consul.rs b/src/consul.rs index 9a91782..4e4f79c 100644 --- a/src/consul.rs +++ b/src/consul.rs @@ -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 + pub Tags: Vec, } #[derive(Serialize, Deserialize, Debug)] pub struct CatalogNode { - pub Services: HashMap + pub Services: HashMap, } pub struct Consul { client: reqwest::Client, url: String, - idx: Option + idx: Option, } 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 { 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::()?), - 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); } } diff --git a/src/consul_actor.rs b/src/consul_actor.rs index dcbd79e..dc5bb5e 100644 --- a/src/consul_actor.rs +++ b/src/consul_actor.rs @@ -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), - udp_port(HashSet) + udp_port(HashSet), } #[derive(Serialize, Deserialize, Debug)] pub enum DiplonatConsul { - diplonat(Vec) + diplonat(Vec), } pub struct ConsulActor { @@ -31,13 +31,16 @@ pub struct ConsulActor { node: String, retries: u32, - tx_open_ports: watch::Sender + tx_open_ports: watch::Sender, } 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 { @@ -57,9 +60,9 @@ fn to_parameters(catalog: &consul::CatalogNode) -> Vec { } fn to_open_ports(params: &Vec) -> messages::PublicExposedPorts { - let mut op = 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) -> 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; } diff --git a/src/diplonat.rs b/src/diplonat.rs index 6334e5b..3b5793d 100644 --- a/src/diplonat.rs +++ b/src/diplonat.rs @@ -1,4 +1,4 @@ -use anyhow::{Result, anyhow}; +use anyhow::{anyhow, Result}; use tokio::try_join; use crate::config::ConfigOpts; @@ -17,22 +17,17 @@ impl Diplonat { pub async fn new() -> Result { let config = ConfigOpts::from_env()?; println!("{:#?}", config); - + let consul_actor = ConsulActor::new(config.consul); - 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 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?; 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(()), } }, )?; diff --git a/src/fw.rs b/src/fw.rs index a71dd1c..e18a301 100644 --- a/src/fw.rs +++ b/src/fw.rs @@ -1,6 +1,6 @@ use std::collections::HashSet; -use anyhow::{Result,Context}; +use anyhow::{Context, Result}; use iptables; use log::*; use regex::Regex; @@ -8,76 +8,92 @@ 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)?; - // 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(()) + 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")?; - } + for p in ports.tcp_ports { + 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")?; - } + for p in ports.udp_ports { + ipt + .append( + "filter", + "DIPLONAT", + &format!("-p udp --dport {} -j ACCEPT", p), + ) + .context("Failed to insert port rule")?; + } - Ok(()) + Ok(()) } pub fn get_opened_ports(ipt: &iptables::IPTables) -> Result { - let mut ports = messages::PublicExposedPorts { - tcp_ports: HashSet::new(), - udp_ports: HashSet::new() - }; + let mut ports = messages::PublicExposedPorts { + tcp_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")?; - 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 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")?; + 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::()?; - let proto = String::from(raw_proto.as_str()); - let number = String::from(raw_port.as_str()).parse::()?; - - if proto == "tcp" { - ports.tcp_ports.insert(number); - } else { - ports.udp_ports.insert(number); - } - - } else { - error!("Unexpected rule found in DIPLONAT chain") - } - - }, - _ => {} + if proto == "tcp" { + ports.tcp_ports.insert(number); + } else { + ports.udp_ports.insert(number); + } + } else { + error!("Unexpected rule found in DIPLONAT chain") } + } + _ => {} } + } - Ok(ports) + Ok(ports) } 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")?; - - if ipt.exists("filter", "INPUT", "-j DIPLONAT")? { - ipt.delete("filter", "INPUT", "-j DIPLONAT").context("Failed to delete jump rule")?; - } - - ipt.delete_chain("filter", "DIPLONAT").context("Failed to delete chain")?; + if ipt.chain_exists("filter", "DIPLONAT")? { + 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")?; } - Ok(()) -} + ipt + .delete_chain("filter", "DIPLONAT") + .context("Failed to delete chain")?; + } + Ok(()) +} diff --git a/src/fw_actor.rs b/src/fw_actor.rs index 29e6473..d41a97f 100644 --- a/src/fw_actor.rs +++ b/src/fw_actor.rs @@ -4,18 +4,15 @@ use anyhow::Result; use iptables; use log::*; use tokio::{ - select, - sync::watch, - time::{ - Duration, - self, -}}; + select, + sync::watch, + time::{self, Duration}, +}; use crate::config::RuntimeConfigFirewall; use crate::fw; use crate::messages; - pub struct FirewallActor { pub ipt: iptables::IPTables, @@ -26,13 +23,16 @@ pub struct FirewallActor { } impl FirewallActor { - pub async fn new(config: Option, rxp: &watch::Receiver) -> Result> { + pub async fn new( + config: Option, + rxp: &watch::Receiver, + ) -> Result> { if config.is_none() { return Ok(None); } let config = config.unwrap(); - let ctx = Self { + let ctx = Self { ipt: iptables::new(false)?, last_ports: messages::PublicExposedPorts::new(), refresh: config.refresh_time, @@ -40,7 +40,7 @@ impl FirewallActor { }; fw::setup(&ctx.ipt)?; - + return Ok(Some(ctx)); } @@ -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::>(); - let diff_udp = self.last_ports.udp_ports.difference(&curr_opened_ports.udp_ports).copied().collect::>(); + let diff_tcp = self + .last_ports + .tcp_ports + .difference(&curr_opened_ports.tcp_ports) + .copied() + .collect::>(); + let diff_udp = self + .last_ports + .udp_ports + .difference(&curr_opened_ports.udp_ports) + .copied() + .collect::>(); let ports_to_open = messages::PublicExposedPorts { - tcp_ports: diff_tcp, - udp_ports: diff_udp + tcp_ports: diff_tcp, + udp_ports: diff_udp, }; fw::open_ports(&self.ipt, ports_to_open)?; return Ok(()); } - } - diff --git a/src/igd_actor.rs b/src/igd_actor.rs index 19a7e93..928bb04 100644 --- a/src/igd_actor.rs +++ b/src/igd_actor.rs @@ -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, -}}; + select, + sync::watch, + time::{self, Duration}, +}; use crate::config::RuntimeConfigIgd; use crate::messages; @@ -26,18 +24,21 @@ pub struct IgdActor { } impl IgdActor { - pub async fn new(config: Option, rxp: &watch::Receiver) -> Result> { + pub async fn new( + config: Option, + rxp: &watch::Receiver, + ) -> Result> { if config.is_none() { return Ok(None); } let config = config.unwrap(); let gw = search_gateway(Default::default()) - .await - .context("Failed to find IGD gateway")?; + .await + .context("Failed to find IGD gateway")?; info!("IGD gateway: {}", gw); - let ctx = Self { + let ctx = Self { expire: config.expiration_time, gateway: gw, last_ports: messages::PublicExposedPorts::new(), @@ -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 { @@ -72,15 +75,26 @@ 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::TCP, &self.last_ports.tcp_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::().context("Invalid socket address")?; - self.gateway.add_port(*proto, *port, service, self.expire.as_secs() as u32, "diplonat").await?; + let service = service_str + .parse::() + .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); } } diff --git a/src/main.rs b/src/main.rs index 720edf8..99d38f5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,8 +7,8 @@ mod fw_actor; mod igd_actor; mod messages; -use log::*; use diplonat::Diplonat; +use log::*; #[tokio::main] async fn main() { diff --git a/src/messages.rs b/src/messages.rs index 09a7c14..63f16b0 100644 --- a/src/messages.rs +++ b/src/messages.rs @@ -3,14 +3,14 @@ use std::collections::HashSet; #[derive(Debug, Clone, PartialEq, Eq)] pub struct PublicExposedPorts { pub tcp_ports: HashSet, - pub udp_ports: HashSet + pub udp_ports: HashSet, } impl PublicExposedPorts { pub fn new() -> Self { return Self { tcp_ports: HashSet::new(), - udp_ports: HashSet::new() - } + udp_ports: HashSet::new(), + }; } } -- 2.43.0 From ffab1f6c4b5c10a4a4ab1800bca14458d9968326 Mon Sep 17 00:00:00 2001 From: Quentin Dufour Date: Mon, 20 Sep 2021 12:08:10 +0200 Subject: [PATCH 04/10] Fix code style --- src/fw.rs | 1 - src/fw_actor.rs | 1 - 2 files changed, 2 deletions(-) diff --git a/src/fw.rs b/src/fw.rs index 3cb4afb..e18a301 100644 --- a/src/fw.rs +++ b/src/fw.rs @@ -97,4 +97,3 @@ pub fn cleanup(ipt: &iptables::IPTables) -> Result<()> { Ok(()) } - diff --git a/src/fw_actor.rs b/src/fw_actor.rs index c32ee5a..d41a97f 100644 --- a/src/fw_actor.rs +++ b/src/fw_actor.rs @@ -93,4 +93,3 @@ impl FirewallActor { return Ok(()); } } - -- 2.43.0 From 900ba20fbe09393745a6107177b381c512a4d38c Mon Sep 17 00:00:00 2001 From: Quentin Dufour Date: Mon, 20 Sep 2021 16:07:58 +0200 Subject: [PATCH 05/10] Use the new formatting rules --- .rustfmt.toml | 73 ++++++++++++++++++++++++++++++++++++++ src/config/options.rs | 6 ++-- src/config/options_test.rs | 35 +++++++++--------- src/config/runtime.rs | 18 +++++----- src/consul.rs | 8 ++--- src/consul_actor.rs | 24 ++++++------- src/diplonat.rs | 13 ++++--- src/fw_actor.rs | 10 +++--- src/igd_actor.rs | 12 +++---- src/messages.rs | 2 +- 10 files changed, 131 insertions(+), 70 deletions(-) diff --git a/.rustfmt.toml b/.rustfmt.toml index f0fa07f..37dfe2c 100644 --- a/.rustfmt.toml +++ b/.rustfmt.toml @@ -1,2 +1,75 @@ +unstable_features = true + +array_width = 60 +attr_fn_like_width = 70 +binop_separator = "Front" +blank_lines_lower_bound = 0 +blank_lines_upper_bound = 1 +brace_style = "SameLineWhere" +chain_width = 60 +color = "Auto" +combine_control_expr = true +comment_width = 80 +condense_wildcard_suffixes = true +control_brace_style = "AlwaysSameLine" +disable_all_formatting = false +empty_item_single_line = true +enum_discrim_align_threshold = 0 +error_on_line_overflow = true +error_on_unformatted = true +fn_args_layout = "Tall" +fn_call_width = 60 +fn_single_line = true +force_explicit_abi = true +force_multiline_blocks = false +format_code_in_doc_comments = true +# format_generated_files = true +format_macro_matchers = true +format_macro_bodies = true +format_strings = true hard_tabs = false +#hex_literal_case = "Lower" +hide_parse_errors = false +ignore = [] +imports_indent = "Block" +imports_layout = "Mixed" +indent_style = "Block" +inline_attribute_width = 0 +license_template_path = "" +match_arm_blocks = true +match_arm_leading_pipes = "Never" +match_block_trailing_comma = false +max_width = 100 +merge_derives = true +imports_granularity = "Crate" +newline_style = "Unix" +normalize_comments = true +normalize_doc_attributes = true +overflow_delimited_expr = false +remove_nested_parens = true +reorder_impl_items = true +reorder_imports = true +group_imports = "StdExternalCrate" +reorder_modules = true +report_fixme = "Unnumbered" +report_todo = "Unnumbered" +required_version = "1.4.37" +skip_children = false +single_line_if_else_max_width = 50 +space_after_colon = true +space_before_colon = false +#space_around_ranges = false +struct_field_align_threshold = 0 +struct_lit_single_line = true +struct_lit_width = 18 +struct_variant_width = 35 tab_spaces = 2 +trailing_comma = "Vertical" +trailing_semicolon = false +type_punctuation_density = "Wide" +use_field_init_shorthand = false +use_small_heuristics = "Off" +use_try_shorthand = true +version = "Two" +where_single_line = true +wrap_comments = true diff --git a/src/config/options.rs b/src/config/options.rs index 638a2f3..7f4eefe 100644 --- a/src/config/options.rs +++ b/src/config/options.rs @@ -8,7 +8,7 @@ use crate::config::RuntimeConfig; // This file parses the options that can be declared in the environment. // runtime.rs applies business logic and builds RuntimeConfig structs. -// - Note for the future - +// Note for the future // There is no *need* to have a 'DIPLONAT_XXX_*' prefix for all config options. // If some config options are shared by several modules, a ConfigOptsBase could // contain them, and parse the 'DIPLONAT_*' prefix directly. @@ -87,9 +87,7 @@ impl ConfigOpts { // Currently only used in tests #[allow(dead_code)] pub fn from_iter(iter: Iter) -> Result - where - Iter: IntoIterator, - { + where Iter: IntoIterator { 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 = diff --git a/src/config/options_test.rs b/src/config/options_test.rs index 3d74d7d..6072c43 100644 --- a/src/config/options_test.rs +++ b/src/config/options_test.rs @@ -1,5 +1,4 @@ -use std::collections::HashMap; -use std::time::Duration; +use std::{collections::HashMap, time::Duration}; use crate::config::*; @@ -65,22 +64,22 @@ fn ok_minimal_valid_options() { assert!(rt_config.acme.is_none()); assert!(rt_config.firewall.is_none()); assert!(rt_config.igd.is_none()); - /*assert_eq!( - rt_config.firewall.refresh_time, - Duration::from_secs(REFRESH_TIME.into()) - ); - assert_eq!( - &rt_config.igd.private_ip, - opts.get(&"DIPLONAT_PRIVATE_IP".to_string()).unwrap() - ); - assert_eq!( - rt_config.igd.expiration_time, - Duration::from_secs(EXPIRATION_TIME.into()) - ); - assert_eq!( - rt_config.igd.refresh_time, - Duration::from_secs(REFRESH_TIME.into()) - );*/ + // assert_eq!( + // rt_config.firewall.refresh_time, + // Duration::from_secs(REFRESH_TIME.into()) + // ); + // assert_eq!( + // &rt_config.igd.private_ip, + // opts.get(&"DIPLONAT_PRIVATE_IP".to_string()).unwrap() + // ); + // assert_eq!( + // rt_config.igd.expiration_time, + // Duration::from_secs(EXPIRATION_TIME.into()) + // ); + // assert_eq!( + // rt_config.igd.refresh_time, + // Duration::from_secs(REFRESH_TIME.into()) + // ); } #[test] diff --git a/src/config/runtime.rs b/src/config/runtime.rs index 4e718a0..0924b16 100644 --- a/src/config/runtime.rs +++ b/src/config/runtime.rs @@ -8,8 +8,8 @@ use crate::config::{ // This code is inspired by the Trunk crate (https://github.com/thedodd/trunk) -// In this file, we take ConfigOpts and transform them into ready-to-use RuntimeConfig. -// We apply default values and business logic. +// In this file, we take ConfigOpts and transform them into ready-to-use +// RuntimeConfig. We apply default values and business logic. // Consul config is mandatory, all the others are optional. @@ -74,7 +74,7 @@ impl RuntimeConfigConsul { impl RuntimeConfigAcme { pub fn new(opts: ConfigOptsAcme) -> Result> { if !opts.enable { - return Ok(None); + return Ok(None) } let email = opts @@ -88,7 +88,7 @@ impl RuntimeConfigAcme { impl RuntimeConfigFirewall { pub(super) fn new(opts: ConfigOptsFirewall) -> Result> { if !opts.enable { - return Ok(None); + return Ok(None) } let refresh_time = Duration::from_secs(opts.refresh_time.unwrap_or(super::REFRESH_TIME).into()); @@ -100,7 +100,7 @@ impl RuntimeConfigFirewall { impl RuntimeConfigIgd { pub(super) fn new(opts: ConfigOptsIgd) -> Result> { if !opts.enable { - return Ok(None); + return Ok(None) } let private_ip = opts @@ -116,9 +116,11 @@ impl RuntimeConfigIgd { if refresh_time.as_secs() * 2 > expiration_time.as_secs() { return Err(anyhow!( - "IGD expiration time (currently: {}s) must be at least twice bigger than refresh time (currently: {}s)", - expiration_time.as_secs(), - refresh_time.as_secs())); + "IGD expiration time (currently: {}s) must be at least twice bigger than refresh time \ + (currently: {}s)", + expiration_time.as_secs(), + refresh_time.as_secs() + )) } Ok(Some(Self { diff --git a/src/consul.rs b/src/consul.rs index 4e4f79c..4e6fe5d 100644 --- a/src/consul.rs +++ b/src/consul.rs @@ -25,12 +25,10 @@ impl Consul { client: reqwest::Client::new(), url: url.to_string(), idx: None, - }; + } } - pub fn watch_node_reset(&mut self) -> () { - self.idx = None; - } + pub fn watch_node_reset(&mut self) -> () { self.idx = None; } pub async fn watch_node(&mut self, host: &str) -> Result { let url = match self.idx { @@ -45,6 +43,6 @@ impl Consul { }; let resp: CatalogNode = http.json().await?; - return Ok(resp); + return Ok(resp) } } diff --git a/src/consul_actor.rs b/src/consul_actor.rs index dc5bb5e..bc35307 100644 --- a/src/consul_actor.rs +++ b/src/consul_actor.rs @@ -1,17 +1,12 @@ -use std::cmp; -use std::collections::HashSet; -use std::time::Duration; +use std::{cmp, collections::HashSet, time::Duration}; use anyhow::Result; use log::*; use serde::{Deserialize, Serialize}; use serde_lexpr::{error, from_str}; -use tokio::sync::watch; -use tokio::time::delay_for; +use tokio::{sync::watch, time::delay_for}; -use crate::config::RuntimeConfigConsul; -use crate::consul; -use crate::messages; +use crate::{config::RuntimeConfigConsul, consul, messages}; #[derive(Serialize, Deserialize, Debug)] pub enum DiplonatParameter { @@ -36,11 +31,12 @@ pub struct ConsulActor { 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 + // 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, - )); + )) } fn to_parameters(catalog: &consul::CatalogNode) -> Vec { @@ -56,7 +52,7 @@ fn to_parameters(catalog: &consul::CatalogNode) -> Vec { } } - return r; + return r } fn to_open_ports(params: &Vec) -> messages::PublicExposedPorts { @@ -75,7 +71,7 @@ fn to_open_ports(params: &Vec) -> messages::PublicExposedPorts { } } - return op; + return op } impl ConsulActor { @@ -91,7 +87,7 @@ impl ConsulActor { retries: 0, rx_open_ports: rx, tx_open_ports: tx, - }; + } } pub async fn listen(&mut self) -> Result<()> { @@ -108,7 +104,7 @@ impl ConsulActor { e ); delay_for(will_retry_in).await; - continue; + continue } }; self.retries = 0; diff --git a/src/diplonat.rs b/src/diplonat.rs index 3b5793d..798a678 100644 --- a/src/diplonat.rs +++ b/src/diplonat.rs @@ -1,10 +1,9 @@ use anyhow::{anyhow, Result}; use tokio::try_join; -use crate::config::ConfigOpts; -use crate::consul_actor::ConsulActor; -use crate::fw_actor::FirewallActor; -use crate::igd_actor::IgdActor; +use crate::{ + config::ConfigOpts, consul_actor::ConsulActor, fw_actor::FirewallActor, igd_actor::IgdActor, +}; pub struct Diplonat { consul: ConsulActor, @@ -27,7 +26,7 @@ impl Diplonat { if firewall_actor.is_none() && igd_actor.is_none() { return Err(anyhow!( "At least enable *one* module, otherwise it's boring!" - )); + )) } let ctx = Self { @@ -36,7 +35,7 @@ impl Diplonat { igd: igd_actor, }; - return Ok(ctx); + return Ok(ctx) } pub async fn listen(&mut self) -> Result<()> { @@ -59,6 +58,6 @@ impl Diplonat { }, )?; - return Ok(()); + return Ok(()) } } diff --git a/src/fw_actor.rs b/src/fw_actor.rs index d41a97f..0333785 100644 --- a/src/fw_actor.rs +++ b/src/fw_actor.rs @@ -9,9 +9,7 @@ use tokio::{ time::{self, Duration}, }; -use crate::config::RuntimeConfigFirewall; -use crate::fw; -use crate::messages; +use crate::{config::RuntimeConfigFirewall, fw, messages}; pub struct FirewallActor { pub ipt: iptables::IPTables, @@ -28,7 +26,7 @@ impl FirewallActor { rxp: &watch::Receiver, ) -> Result> { if config.is_none() { - return Ok(None); + return Ok(None) } let config = config.unwrap(); @@ -41,7 +39,7 @@ impl FirewallActor { fw::setup(&ctx.ipt)?; - return Ok(Some(ctx)); + return Ok(Some(ctx)) } pub async fn listen(&mut self) -> Result<()> { @@ -90,6 +88,6 @@ impl FirewallActor { fw::open_ports(&self.ipt, ports_to_open)?; - return Ok(()); + return Ok(()) } } diff --git a/src/igd_actor.rs b/src/igd_actor.rs index 928bb04..e965ce2 100644 --- a/src/igd_actor.rs +++ b/src/igd_actor.rs @@ -1,8 +1,7 @@ use std::net::SocketAddrV4; use anyhow::{Context, Result}; -use igd::aio::*; -use igd::PortMappingProtocol; +use igd::{aio::*, PortMappingProtocol}; use log::*; use tokio::{ select, @@ -10,8 +9,7 @@ use tokio::{ time::{self, Duration}, }; -use crate::config::RuntimeConfigIgd; -use crate::messages; +use crate::{config::RuntimeConfigIgd, messages}; pub struct IgdActor { expire: Duration, @@ -29,7 +27,7 @@ impl IgdActor { rxp: &watch::Receiver, ) -> Result> { if config.is_none() { - return Ok(None); + return Ok(None) } let config = config.unwrap(); @@ -47,7 +45,7 @@ impl IgdActor { rx_ports: rxp.clone(), }; - return Ok(Some(ctx)); + return Ok(Some(ctx)) } pub async fn listen(&mut self) -> Result<()> { @@ -99,6 +97,6 @@ impl IgdActor { } } - return Ok(()); + return Ok(()) } } diff --git a/src/messages.rs b/src/messages.rs index 63f16b0..b622be1 100644 --- a/src/messages.rs +++ b/src/messages.rs @@ -11,6 +11,6 @@ impl PublicExposedPorts { return Self { tcp_ports: HashSet::new(), udp_ports: HashSet::new(), - }; + } } } -- 2.43.0 From beefcd7adea5bb932d8393024076287b5713b16c Mon Sep 17 00:00:00 2001 From: adrien Date: Wed, 22 Sep 2021 12:47:14 +0200 Subject: [PATCH 06/10] WIP: removed useless '.clone()' in runtime.rs and changed config test file name to config_test.rs --- src/config/{options_test.rs => config_test.rs} | 0 src/config/mod.rs | 4 ++-- src/config/runtime.rs | 8 ++++---- 3 files changed, 6 insertions(+), 6 deletions(-) rename src/config/{options_test.rs => config_test.rs} (100%) diff --git a/src/config/options_test.rs b/src/config/config_test.rs similarity index 100% rename from src/config/options_test.rs rename to src/config/config_test.rs diff --git a/src/config/mod.rs b/src/config/mod.rs index 9dddd04..a11fab4 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -1,6 +1,6 @@ -mod options; #[cfg(test)] -mod options_test; +mod config_test; +mod options; mod runtime; pub use options::{ diff --git a/src/config/runtime.rs b/src/config/runtime.rs index 0924b16..1cc409c 100644 --- a/src/config/runtime.rs +++ b/src/config/runtime.rs @@ -46,10 +46,10 @@ pub struct RuntimeConfig { impl RuntimeConfig { pub fn new(opts: ConfigOpts) -> Result { - let consul = RuntimeConfigConsul::new(opts.consul.clone())?; - let acme = RuntimeConfigAcme::new(opts.acme.clone())?; - let firewall = RuntimeConfigFirewall::new(opts.firewall.clone())?; - let igd = RuntimeConfigIgd::new(opts.igd.clone())?; + let consul = RuntimeConfigConsul::new(opts.consul)?; + let acme = RuntimeConfigAcme::new(opts.acme)?; + let firewall = RuntimeConfigFirewall::new(opts.firewall)?; + let igd = RuntimeConfigIgd::new(opts.igd)?; Ok(Self { acme, -- 2.43.0 From f5ac36e21f4ae2f720070b5fff550954c7af0b25 Mon Sep 17 00:00:00 2001 From: adrien Date: Wed, 22 Sep 2021 13:12:57 +0200 Subject: [PATCH 07/10] WIP: rewrote runtime.rs using match to convey intent better --- src/config/runtime.rs | 62 +++++++++++++++++++++++++++++++------------ 1 file changed, 45 insertions(+), 17 deletions(-) diff --git a/src/config/runtime.rs b/src/config/runtime.rs index 1cc409c..89acdfd 100644 --- a/src/config/runtime.rs +++ b/src/config/runtime.rs @@ -62,10 +62,14 @@ impl RuntimeConfig { impl RuntimeConfigConsul { pub(super) fn new(opts: ConfigOptsConsul) -> Result { - let node_name = opts - .node_name - .expect("'DIPLONAT_CONSUL_NODE_NAME' is required"); - let url = opts.url.unwrap_or(super::CONSUL_URL.to_string()); + let node_name = match opts.node_name { + Some(n) => n, + _ => return Err(anyhow!("'DIPLONAT_CONSUL_NODE_NAME' is required")), + }; + let url = match opts.url { + Some(url) => url, + _ => super::CONSUL_URL.to_string(), + }; Ok(Self { node_name, url }) } @@ -75,11 +79,16 @@ impl RuntimeConfigAcme { pub fn new(opts: ConfigOptsAcme) -> Result> { if !opts.enable { return Ok(None) - } + }; - let email = opts - .email - .expect("'DIPLONAT_ACME_EMAIL' is required if ACME is enabled"); + let email = match opts.email { + Some(email) => email, + _ => { + return Err(anyhow!( + "'DIPLONAT_ACME_EMAIL' is required if ACME is enabled" + )) + } + }; Ok(Some(Self { email })) } @@ -91,7 +100,13 @@ 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( + match opts.refresh_time { + Some(t) => t, + _ => super::REFRESH_TIME, + } + .into(), + ); Ok(Some(Self { refresh_time })) } @@ -103,16 +118,29 @@ 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 = match opts.private_ip { + Some(ip) => ip, + _ => { + return Err(anyhow!( + "'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(), + match opts.expiration_time { + Some(t) => t.into(), + _ => super::EXPIRATION_TIME, + } + .into(), + ); + let refresh_time = Duration::from_secs( + match opts.refresh_time { + Some(t) => t, + _ => super::REFRESH_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!( -- 2.43.0 From 006f84e3b0fff2a744a108570692351b927e0e73 Mon Sep 17 00:00:00 2001 From: adrien Date: Wed, 22 Sep 2021 14:47:58 +0200 Subject: [PATCH 08/10] WIP: rewrote body of 'new(...)' for FirewallActor and IgdActor using match; clearer intent --- src/diplonat.rs | 2 +- src/fw_actor.rs | 29 ++++++++++++++--------------- src/igd_actor.rs | 36 +++++++++++++++++------------------- 3 files changed, 32 insertions(+), 35 deletions(-) diff --git a/src/diplonat.rs b/src/diplonat.rs index 798a678..7ad327d 100644 --- a/src/diplonat.rs +++ b/src/diplonat.rs @@ -19,7 +19,7 @@ impl Diplonat { 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)?; let igd_actor = IgdActor::new(config.igd, &consul_actor.rx_open_ports).await?; diff --git a/src/fw_actor.rs b/src/fw_actor.rs index 0333785..97de1c3 100644 --- a/src/fw_actor.rs +++ b/src/fw_actor.rs @@ -21,25 +21,24 @@ pub struct FirewallActor { } impl FirewallActor { - pub async fn new( + pub fn new( config: Option, rxp: &watch::Receiver, ) -> Result> { - if config.is_none() { - return Ok(None) + match config { + None => Ok(None), + Some(c) => { + let ctx = Self { + ipt: iptables::new(false)?, + last_ports: messages::PublicExposedPorts::new(), + refresh: c.refresh_time, + rx_ports: rxp.clone(), + }; + fw::setup(&ctx.ipt)?; + + Ok(Some(ctx)) + } } - let config = config.unwrap(); - - let ctx = Self { - ipt: iptables::new(false)?, - last_ports: messages::PublicExposedPorts::new(), - refresh: config.refresh_time, - rx_ports: rxp.clone(), - }; - - fw::setup(&ctx.ipt)?; - - return Ok(Some(ctx)) } pub async fn listen(&mut self) -> Result<()> { diff --git a/src/igd_actor.rs b/src/igd_actor.rs index e965ce2..ffd71dd 100644 --- a/src/igd_actor.rs +++ b/src/igd_actor.rs @@ -26,26 +26,24 @@ impl IgdActor { config: Option, rxp: &watch::Receiver, ) -> Result> { - if config.is_none() { - return Ok(None) + match config { + None => Ok(None), + Some(c) => { + let gw = search_gateway(Default::default()) + .await + .context("Failed to find IGD gateway")?; + info!("IGD gateway: {}", gw); + + Ok(Some(Self { + expire: c.expiration_time, + gateway: gw, + last_ports: messages::PublicExposedPorts::new(), + private_ip: c.private_ip, + refresh: c.refresh_time, + rx_ports: rxp.clone(), + })) + } } - let config = config.unwrap(); - - let gw = search_gateway(Default::default()) - .await - .context("Failed to find IGD gateway")?; - info!("IGD gateway: {}", gw); - - let ctx = Self { - expire: config.expiration_time, - gateway: gw, - last_ports: messages::PublicExposedPorts::new(), - private_ip: config.private_ip, - refresh: config.refresh_time, - rx_ports: rxp.clone(), - }; - - return Ok(Some(ctx)) } pub async fn listen(&mut self) -> Result<()> { -- 2.43.0 From a464aabe256a817d5872a1b246ccb383a5faf114 Mon Sep 17 00:00:00 2001 From: adrien Date: Wed, 22 Sep 2021 15:52:50 +0200 Subject: [PATCH 09/10] WIP: I'm lost with Boxes and opaque types --- src/diplonat.rs | 52 +++++++++++++++---------------------------------- 1 file changed, 16 insertions(+), 36 deletions(-) diff --git a/src/diplonat.rs b/src/diplonat.rs index 7ad327d..d98b6ed 100644 --- a/src/diplonat.rs +++ b/src/diplonat.rs @@ -1,4 +1,7 @@ +use std::pin::Pin; + use anyhow::{anyhow, Result}; +use futures::future::{try_join_all, Future}; use tokio::try_join; use crate::{ @@ -6,10 +9,7 @@ use crate::{ }; pub struct Diplonat { - consul: ConsulActor, - - firewall: Option, - igd: Option, + actors: Vec>>>>, } impl Diplonat { @@ -18,46 +18,26 @@ impl Diplonat { println!("{:#?}", config); let consul_actor = ConsulActor::new(config.consul); + let mut actors = vec![Box::pin(consul_actor.listen())]; - let firewall_actor = FirewallActor::new(config.firewall, &consul_actor.rx_open_ports)?; + if let Some(actor) = FirewallActor::new(config.firewall, &consul_actor.rx_open_ports)? { + actors.push(Box::pin(actor.listen())); + } + if let Some(actor) = IgdActor::new(config.igd, &consul_actor.rx_open_ports).await? { + actors.push(Box::pin(actor.listen())); + } - let igd_actor = IgdActor::new(config.igd, &consul_actor.rx_open_ports).await?; - - if firewall_actor.is_none() && igd_actor.is_none() { + if actors.len() == 1 { return Err(anyhow!( "At least enable *one* module, otherwise it's boring!" )) } - let ctx = Self { - consul: consul_actor, - firewall: firewall_actor, - igd: igd_actor, - }; - - return Ok(ctx) + Ok(Self { actors }) } - pub async fn listen(&mut self) -> Result<()> { - let firewall = &mut self.firewall; - let igd = &mut self.igd; - - try_join!( - self.consul.listen(), - async { - match firewall { - Some(x) => x.listen().await, - None => Ok(()), - } - }, - async { - match igd { - Some(x) => x.listen().await, - None => Ok(()), - } - }, - )?; - - return Ok(()) + pub async fn listen(&self) -> Result<()> { + try_join_all(self.actors); + Ok(()) } } -- 2.43.0 From 6015d85b7331306dcd4f989b682442f064df9553 Mon Sep 17 00:00:00 2001 From: adrien Date: Wed, 22 Sep 2021 17:02:25 +0200 Subject: [PATCH 10/10] WIP: looks better, but the Diplonat new function is still full of shit - I'll investigate --- src/diplonat.rs | 23 +++++++++++++++-------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/src/diplonat.rs b/src/diplonat.rs index d98b6ed..884e8ec 100644 --- a/src/diplonat.rs +++ b/src/diplonat.rs @@ -2,14 +2,16 @@ use std::pin::Pin; use anyhow::{anyhow, Result}; use futures::future::{try_join_all, Future}; -use tokio::try_join; +// use tokio::try_join; + +type ActorTask = Pin>>>; use crate::{ config::ConfigOpts, consul_actor::ConsulActor, fw_actor::FirewallActor, igd_actor::IgdActor, }; pub struct Diplonat { - actors: Vec>>>>, + actors: Vec, } impl Diplonat { @@ -17,14 +19,19 @@ impl Diplonat { let config = ConfigOpts::from_env()?; println!("{:#?}", config); - let consul_actor = ConsulActor::new(config.consul); - let mut actors = vec![Box::pin(consul_actor.listen())]; + let mut consul_actor = ConsulActor::new(config.consul); + let consul_rx = consul_actor.rx_open_ports.clone(); + let actor_task: ActorTask = Box::pin(consul_actor.listen()); - if let Some(actor) = FirewallActor::new(config.firewall, &consul_actor.rx_open_ports)? { - actors.push(Box::pin(actor.listen())); + let mut actors = vec![actor_task]; + + if let Some(mut actor) = FirewallActor::new(config.firewall, &consul_rx)? { + let actor_task: ActorTask = Box::pin(actor.listen()); + actors.push(actor_task); } - if let Some(actor) = IgdActor::new(config.igd, &consul_actor.rx_open_ports).await? { - actors.push(Box::pin(actor.listen())); + if let Some(mut actor) = IgdActor::new(config.igd, &consul_rx).await? { + let actor_task: ActorTask = Box::pin(actor.listen()); + actors.push(actor_task); } if actors.len() == 1 { -- 2.43.0