diff options
-rw-r--r-- | .rustfmt.toml | 2 | ||||
-rw-r--r-- | src/acme_actor.rs | 27 | ||||
-rw-r--r-- | src/config/mod.rs | 10 | ||||
-rw-r--r-- | src/config/options.rs | 130 | ||||
-rw-r--r-- | src/config/options_test.rs | 216 | ||||
-rw-r--r-- | src/config/runtime.rs | 175 | ||||
-rw-r--r-- | src/consul.rs | 18 | ||||
-rw-r--r-- | src/consul_actor.rs | 23 | ||||
-rw-r--r-- | src/diplonat.rs | 36 | ||||
-rw-r--r-- | src/fw.rs | 132 | ||||
-rw-r--r-- | src/fw_actor.rs | 46 | ||||
-rw-r--r-- | src/igd_actor.rs | 46 | ||||
-rw-r--r-- | src/main.rs | 2 | ||||
-rw-r--r-- | src/messages.rs | 6 |
14 files changed, 465 insertions, 404 deletions
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/acme_actor.rs b/src/acme_actor.rs index 3b6aa50..0d7aec4 100644 --- a/src/acme_actor.rs +++ b/src/acme_actor.rs @@ -1,12 +1,10 @@ use anyhow::Result; use log::*; use tokio::{ - select, - sync::watch, - time::{ - Duration, - self, -}}; + select, + sync::watch, + time::{self, Duration}, +}; use crate::config::RuntimeConfigAcme; use crate::messages; @@ -21,18 +19,17 @@ pub struct AcmeActor { impl AcmeActor { pub async fn new( - config: Option<RuntimeConfigAcme>, - rxp: &watch::Receiver<messages::PublicExposedPorts> - ) -> Result<Option<Self>> { - + config: Option<RuntimeConfigAcme>, + rxp: &watch::Receiver<messages::PublicExposedPorts>, + ) -> Result<Option<Self>> { if config.is_none() { return Ok(None); } let config = config.unwrap(); - let ctx = Self { + let ctx = Self { email: config.email, - last_ports: messages::PublicExposedPorts::new(), + last_ports: messages::PublicExposedPorts::new(), refresh: config.refresh_time, rx_ports: rxp.clone(), }; @@ -51,7 +48,9 @@ impl AcmeActor { }; // 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_acme().await { @@ -67,4 +66,4 @@ impl AcmeActor { Ok(()) } -}
\ No newline at end of file +} 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 54e948d..6d636aa 100644 --- a/src/config/options.rs +++ b/src/config/options.rs @@ -12,95 +12,97 @@ 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<String>, - /// Consul's REST URL [default: "http://127.0.0.1:8500"] - pub url: Option<String>, + /// Consul's node name [default: None] + pub node_name: Option<String>, + /// Consul's REST URL [default: "http://127.0.0.1:8500"] + pub url: Option<String>, } /// 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<String>, - /// Refresh time for firewall rules [default: 300] - pub refresh_time: Option<u16>, + /// 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<String>, + /// Refresh time for firewall rules [default: 300] + pub refresh_time: Option<u16>, } /// 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<u16>, + /// Refresh time for firewall rules [default: 300] + pub refresh_time: Option<u16>, } /// 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<String>, - /// Expiration time for IGD rules [default: 60] - pub expiration_time: Option<u16>, - /// Refresh time for IGD rules [default: 300] - pub refresh_time: Option<u16>, + /// 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<String>, + /// Expiration time for IGD rules [default: 60] + pub expiration_time: Option<u16>, + /// Refresh time for IGD rules [default: 300] + pub refresh_time: Option<u16>, } /// 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<RuntimeConfig> { - 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, - }) - } - - // Currently only used in tests - #[allow(dead_code)] - pub fn from_iter<Iter: Clone>(iter: Iter) -> Result<RuntimeConfig> - 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 igd: ConfigOptsIgd = envy::prefixed("DIPLONAT_IGD_").from_iter(iter.clone())?; - - RuntimeConfig::new(Self { - consul, - acme, - firewall, - igd, - }) - } -}
\ No newline at end of file + pub fn from_env() -> Result<RuntimeConfig> { + 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, + }) + } + + // Currently only used in tests + #[allow(dead_code)] + pub fn from_iter<Iter: Clone>(iter: Iter) -> Result<RuntimeConfig> + 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 igd: ConfigOptsIgd = envy::prefixed("DIPLONAT_IGD_").from_iter(iter.clone())?; + + 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<String, String> { - 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<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_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<String, String> { #[test] #[should_panic] fn err_empty_env() { - std::env::remove_var("DIPLONAT_CONSUL_NODE_NAME"); - let opts: HashMap<String, String> = HashMap::new(); - ConfigOpts::from_iter(opts).unwrap(); + std::env::remove_var("DIPLONAT_CONSUL_NODE_NAME"); + let opts: HashMap<String, String> = 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::<u64>().unwrap() - .into()); - let igd_expiration_time = Duration::from_secs( - 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()); + let firewall_refresh_time = Duration::from_secs( + 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(), + ); + let igd_refresh_time = Duration::from_secs( + opts + .get(&"DIPLONAT_IGD_REFRESH_TIME".to_string()) + .unwrap() + .parse::<u64>() + .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 ee7f682..ecdf315 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,122 +15,121 @@ 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 refresh_time: Duration, + pub email: String, + pub refresh_time: Duration, } #[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<RuntimeConfigAcme>, - pub firewall: Option<RuntimeConfigFirewall>, - pub igd: Option<RuntimeConfigIgd>, + pub consul: RuntimeConfigConsul, + pub acme: Option<RuntimeConfigAcme>, + pub firewall: Option<RuntimeConfigFirewall>, + pub igd: Option<RuntimeConfigIgd>, } impl RuntimeConfig { - pub fn new(opts: ConfigOpts) -> Result<Self> { - 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, - }) - } + pub fn new(opts: ConfigOpts) -> Result<Self> { + 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, + }) + } } impl RuntimeConfigConsul { - pub(super) fn new(opts: ConfigOptsConsul) -> Result<Self> { - 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, - }) - } + pub(super) fn new(opts: ConfigOptsConsul) -> Result<Self> { + 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<Option<Self>> { - if !opts.enable { - return Ok(None); - } - - let email = opts.email.expect( - "'DIPLONAT_ACME_EMAIL' is required if ACME is enabled"); - let refresh_time = Duration::from_secs( - opts.refresh_time.unwrap_or(super::REFRESH_TIME).into()); - - Ok(Some(Self { - email, - refresh_time, - })) - } + pub fn new(opts: ConfigOptsAcme) -> Result<Option<Self>> { + if !opts.enable { + return Ok(None); + } + + let email = opts + .email + .expect("'DIPLONAT_ACME_EMAIL' is required if ACME is enabled"); + let refresh_time = Duration::from_secs(opts.refresh_time.unwrap_or(super::REFRESH_TIME).into()); + + Ok(Some(Self { + email, + refresh_time, + })) + } } impl RuntimeConfigFirewall { - pub(super) fn new(opts: ConfigOptsFirewall) -> Result<Option<Self>> { - if !opts.enable { - return Ok(None); - } - - let refresh_time = Duration::from_secs( - opts.refresh_time.unwrap_or(super::REFRESH_TIME).into()); - - Ok(Some(Self { - refresh_time, - })) - } + pub(super) fn new(opts: ConfigOptsFirewall) -> Result<Option<Self>> { + if !opts.enable { + return Ok(None); + } + + let refresh_time = Duration::from_secs(opts.refresh_time.unwrap_or(super::REFRESH_TIME).into()); + + Ok(Some(Self { refresh_time })) + } } impl RuntimeConfigIgd { - pub(super) fn new(opts: ConfigOptsIgd) -> Result<Option<Self>> { - 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()); - - if refresh_time.as_secs() * 2 > expiration_time.as_secs() { - return Err(anyhow!( + pub(super) fn new(opts: ConfigOptsIgd) -> Result<Option<Self>> { + 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()); + + 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<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); } } diff --git a/src/consul_actor.rs b/src/consul_actor.rs index fd4d05f..136b248 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; @@ -18,12 +18,12 @@ use crate::messages; pub enum DiplonatParameter { TcpPort(HashSet<u16>), UdpPort(HashSet<u16>), - Acme(HashSet<String>) + Acme(HashSet<String>), } #[derive(Serialize, Deserialize, Debug)] pub enum DiplonatConsul { - diplonat(Vec<DiplonatParameter>) + diplonat(Vec<DiplonatParameter>), } pub struct ConsulActor { @@ -33,13 +33,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 { + // let mut op = messages::PublicExposedPorts { // tcp_ports: HashSet::new(), // udp_ports: HashSet::new() // }; @@ -104,7 +107,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 b2310f9..ba7bce0 100644 --- a/src/diplonat.rs +++ b/src/diplonat.rs @@ -1,10 +1,10 @@ -use anyhow::{Result, anyhow}; +use anyhow::{anyhow, Result}; use log::debug; use tokio::try_join; +use crate::acme_actor::AcmeActor; use crate::config::ConfigOpts; use crate::consul_actor::ConsulActor; -use crate::acme_actor::AcmeActor; use crate::fw_actor::FirewallActor; use crate::igd_actor::IgdActor; @@ -20,29 +20,19 @@ impl Diplonat { pub async fn new() -> Result<Self> { let config = ConfigOpts::from_env()?; debug!("{:#?}", config); - + let consul_actor = ConsulActor::new(config.consul); - let acme_actor = AcmeActor::new( - config.acme, - &consul_actor.rx_open_ports - ).await?; + let acme_actor = AcmeActor::new(config.acme, &consul_actor.rx_open_ports).await?; + + 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 acme_actor.is_none() && - firewall_actor.is_none() && - igd_actor.is_none() { + if acme_actor.is_none() && 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 { @@ -65,19 +55,19 @@ impl Diplonat { async { match acme { Some(x) => x.listen().await, - None => Ok(()) + None => Ok(()), } }, async { match firewall { Some(x) => x.listen().await, - None => Ok(()) + None => Ok(()), } }, async { match igd { Some(x) => x.listen().await, - None => Ok(()) + None => Ok(()), } }, )?; @@ -1,6 +1,6 @@ // use std::collections::HashSet; -use anyhow::{Result,Context}; +use anyhow::{Context, Result}; use iptables; use log::*; use regex::Regex; @@ -8,77 +8,93 @@ 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.udp_ports { - ipt.append("filter", "DIPLONAT", &format!("-p udp --dport {} -j ACCEPT", p)).context("Failed to insert port rule")?; - } - - Ok(()) + 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")?; + } + + Ok(()) } pub fn get_opened_ports(ipt: &iptables::IPTables) -> Result<messages::PublicExposedPorts> { - let mut ports = messages::PublicExposedPorts::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 proto = String::from(raw_proto.as_str()); - let number = String::from(raw_port.as_str()).parse::<u16>()?; - - if proto == "tcp" { - ports.tcp_ports.insert(number); - } else { - ports.udp_ports.insert(number); - } - - } else { - error!("Unexpected rule found in DIPLONAT chain") - } - - }, - _ => {} + let mut ports = messages::PublicExposedPorts::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 proto = String::from(raw_proto.as_str()); + let number = String::from(raw_port.as_str()).parse::<u16>()?; + + 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 9000048..fa905c0 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<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); } 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,19 +70,27 @@ 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, - acme: HashSet::new() + tcp_ports: diff_tcp, + udp_ports: diff_udp, + acme: HashSet::new(), }; 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<RuntimeConfigIgd>, rxp: &watch::Receiver<messages::PublicExposedPorts>) -> Result<Option<Self>> { + pub async fn new( + config: Option<RuntimeConfigIgd>, + rxp: &watch::Receiver<messages::PublicExposedPorts>, + ) -> Result<Option<Self>> { 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::<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); } } diff --git a/src/main.rs b/src/main.rs index 48467d6..abcc628 100644 --- a/src/main.rs +++ b/src/main.rs @@ -8,8 +8,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 86ba1b2..ea1629a 100644 --- a/src/messages.rs +++ b/src/messages.rs @@ -4,7 +4,7 @@ use std::collections::HashSet; pub struct PublicExposedPorts { pub tcp_ports: HashSet<u16>, pub udp_ports: HashSet<u16>, - pub acme: HashSet<String> + pub acme: HashSet<String>, } impl PublicExposedPorts { @@ -12,7 +12,7 @@ impl PublicExposedPorts { return Self { tcp_ports: HashSet::new(), udp_ports: HashSet::new(), - acme: HashSet::new() - } + acme: HashSet::new(), + }; } } |