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

Remove .rustfmt.toml and move to standard rustfmt format (4 spaces)
All checks were successful
continuous-integration/drone/push Build is passing
Details
continuous-integration/drone/pr Build is passing
Details

Browse source
This commit is contained in:
Alex 2023-04-04 18:48:52 +02:00
parent 2d39adcabb
commit b3f76f272a
14 changed files with 772 additions and 854 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

75
.rustfmt.toml
View file

@ -1,75 +0,0 @@
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

4
src/config/mod.rs
View file

@ -5,8 +5,8 @@ mod runtime;
pub use options::{ConfigOpts, ConfigOptsAcme, ConfigOptsBase, ConfigOptsConsul}; pub use options::{ConfigOpts, ConfigOptsAcme, ConfigOptsBase, ConfigOptsConsul};
pub use runtime::{ pub use runtime::{
RuntimeConfig, RuntimeConfigAcme, RuntimeConfigConsul, RuntimeConfigFirewall, RuntimeConfigIgd, RuntimeConfig, RuntimeConfigAcme, RuntimeConfigConsul, RuntimeConfigFirewall, RuntimeConfigIgd,
RuntimeConfigStun, RuntimeConfigStun,
}; };
pub const EXPIRATION_TIME: u16 = 300; pub const EXPIRATION_TIME: u16 = 300;

109
src/config/options.rs
View file

@ -11,79 +11,80 @@ use crate::config::RuntimeConfig;
/// Base configuration options /// Base configuration options
#[derive(Clone, Default, Deserialize)] #[derive(Clone, Default, Deserialize)]
pub struct ConfigOptsBase { pub struct ConfigOptsBase {
/// This node's private IP address [default: None] /// This node's private IP address [default: None]
pub private_ip: Option<String>, pub private_ip: Option<String>,
/// Expiration time for IGD rules [default: 60] /// Expiration time for IGD rules [default: 60]
pub expiration_time: Option<u16>, pub expiration_time: Option<u16>,
/// Refresh time for IGD and Firewall rules [default: 300] /// Refresh time for IGD and Firewall rules [default: 300]
pub refresh_time: Option<u16>, pub refresh_time: Option<u16>,
/// STUN server [default: stun.nextcloud.com:443] /// STUN server [default: stun.nextcloud.com:443]
pub stun_server: Option<String>, pub stun_server: Option<String>,
} }
/// ACME configuration options /// ACME configuration options
#[derive(Clone, Default, Deserialize)] #[derive(Clone, Default, Deserialize)]
pub struct ConfigOptsAcme { pub struct ConfigOptsAcme {
/// Whether ACME is enabled [default: false] /// Whether ACME is enabled [default: false]
#[serde(default)] #[serde(default)]
pub enable: bool, pub enable: bool,
/// The default domain holder's e-mail [default: None] /// The default domain holder's e-mail [default: None]
pub email: Option<String>, pub email: Option<String>,
} }
/// Consul configuration options /// Consul configuration options
#[derive(Clone, Default, Deserialize)] #[derive(Clone, Default, Deserialize)]
pub struct ConfigOptsConsul { pub struct ConfigOptsConsul {
/// Consul's node name [default: None] /// Consul's node name [default: None]
pub node_name: Option<String>, pub node_name: Option<String>,
/// Consul's REST URL [default: "http://127.0.0.1:8500"] /// Consul's REST URL [default: "http://127.0.0.1:8500"]
pub url: Option<String>, pub url: Option<String>,
/// Consul's CA certificate [default: None] /// Consul's CA certificate [default: None]
pub ca_cert: Option<String>, pub ca_cert: Option<String>,
/// Skip TLS verification for Consul server [default: false] /// Skip TLS verification for Consul server [default: false]
#[serde(default)] #[serde(default)]
pub tls_skip_verify: bool, pub tls_skip_verify: bool,
/// Consul's client certificate [default: None] /// Consul's client certificate [default: None]
pub client_cert: Option<String>, pub client_cert: Option<String>,
/// Consul's client key [default: None] /// Consul's client key [default: None]
pub client_key: Option<String>, pub client_key: Option<String>,
} }
/// Model of all potential configuration options /// Model of all potential configuration options
pub struct ConfigOpts { pub struct ConfigOpts {
pub base: ConfigOptsBase, pub base: ConfigOptsBase,
pub acme: ConfigOptsAcme, pub acme: ConfigOptsAcme,
pub consul: ConfigOptsConsul, pub consul: ConfigOptsConsul,
} }
impl ConfigOpts { impl ConfigOpts {
pub fn from_env() -> Result<RuntimeConfig> { pub fn from_env() -> Result<RuntimeConfig> {
let base: ConfigOptsBase = envy::prefixed("DIPLONAT_").from_env()?; let base: ConfigOptsBase = envy::prefixed("DIPLONAT_").from_env()?;
let consul: ConfigOptsConsul = envy::prefixed("DIPLONAT_CONSUL_").from_env()?; let consul: ConfigOptsConsul = envy::prefixed("DIPLONAT_CONSUL_").from_env()?;
let acme: ConfigOptsAcme = envy::prefixed("DIPLONAT_ACME_").from_env()?; let acme: ConfigOptsAcme = envy::prefixed("DIPLONAT_ACME_").from_env()?;
RuntimeConfig::new(Self { RuntimeConfig::new(Self {
base: base, base: base,
consul: consul, consul: consul,
acme: acme, acme: acme,
}) })
} }
// Currently only used in tests // Currently only used in tests
#[cfg(test)] #[cfg(test)]
pub fn from_iter<Iter: Clone>(iter: Iter) -> Result<RuntimeConfig> pub fn from_iter<Iter: Clone>(iter: Iter) -> Result<RuntimeConfig>
where where
Iter: IntoIterator<Item = (String, String)>, Iter: IntoIterator<Item = (String, String)>,
{ {
let base: ConfigOptsBase = envy::prefixed("DIPLONAT_").from_iter(iter.clone())?; let base: ConfigOptsBase = envy::prefixed("DIPLONAT_").from_iter(iter.clone())?;
let consul: ConfigOptsConsul = envy::prefixed("DIPLONAT_CONSUL_").from_iter(iter.clone())?; let consul: ConfigOptsConsul =
let acme: ConfigOptsAcme = envy::prefixed("DIPLONAT_ACME_").from_iter(iter.clone())?; envy::prefixed("DIPLONAT_CONSUL_").from_iter(iter.clone())?;
let acme: ConfigOptsAcme = envy::prefixed("DIPLONAT_ACME_").from_iter(iter.clone())?;
RuntimeConfig::new(Self { RuntimeConfig::new(Self {
base: base, base: base,
consul: consul, consul: consul,
acme: acme, acme: acme,
}) })
} }
} }

182
src/config/options_test.rs
View file

@ -9,120 +9,118 @@ use crate::config::*;
// This is why we only test ConfigOpts::from_iter(iter). // This is why we only test ConfigOpts::from_iter(iter).
fn minimal_valid_options() -> HashMap<String, String> { fn minimal_valid_options() -> HashMap<String, String> {
let mut opts = HashMap::new(); let mut opts = HashMap::new();
opts.insert( opts.insert(
"DIPLONAT_CONSUL_NODE_NAME".to_string(), "DIPLONAT_CONSUL_NODE_NAME".to_string(),
"consul_node".to_string(), "consul_node".to_string(),
); );
opts opts
} }
fn all_valid_options() -> HashMap<String, String> { fn all_valid_options() -> HashMap<String, String> {
let mut opts = minimal_valid_options(); let mut opts = minimal_valid_options();
opts.insert("DIPLONAT_EXPIRATION_TIME".to_string(), "30".to_string()); opts.insert("DIPLONAT_EXPIRATION_TIME".to_string(), "30".to_string());
opts.insert( opts.insert(
"DIPLONAT_STUN_SERVER".to_string(), "DIPLONAT_STUN_SERVER".to_string(),
"stun.nextcloud.com:443".to_string(), "stun.nextcloud.com:443".to_string(),
); );
opts.insert( opts.insert(
"DIPLONAT_PRIVATE_IP".to_string(), "DIPLONAT_PRIVATE_IP".to_string(),
"172.123.43.555".to_string(), "172.123.43.555".to_string(),
); );
opts.insert("DIPLONAT_REFRESH_TIME".to_string(), "10".to_string()); opts.insert("DIPLONAT_REFRESH_TIME".to_string(), "10".to_string());
opts.insert( opts.insert(
"DIPLONAT_CONSUL_URL".to_string(), "DIPLONAT_CONSUL_URL".to_string(),
"http://127.0.0.1:9999".to_string(), "http://127.0.0.1:9999".to_string(),
); );
opts.insert("DIPLONAT_ACME_ENABLE".to_string(), "true".to_string()); opts.insert("DIPLONAT_ACME_ENABLE".to_string(), "true".to_string());
opts.insert( opts.insert(
"DIPLONAT_ACME_EMAIL".to_string(), "DIPLONAT_ACME_EMAIL".to_string(),
"bozo@bozo.net".to_string(), "bozo@bozo.net".to_string(),
); );
opts opts
} }
#[test] #[test]
#[should_panic] #[should_panic]
fn err_empty_env() { fn err_empty_env() {
std::env::remove_var("DIPLONAT_CONSUL_NODE_NAME"); std::env::remove_var("DIPLONAT_CONSUL_NODE_NAME");
ConfigOpts::from_env().unwrap(); ConfigOpts::from_env().unwrap();
} }
#[test] #[test]
fn ok_from_iter_minimal_valid_options() { fn ok_from_iter_minimal_valid_options() {
let opts = minimal_valid_options(); let opts = minimal_valid_options();
let rt_config = ConfigOpts::from_iter(opts.clone()).unwrap(); let rt_config = ConfigOpts::from_iter(opts.clone()).unwrap();
assert!(rt_config.acme.is_none()); assert!(rt_config.acme.is_none());
assert_eq!( assert_eq!(
&rt_config.consul.node_name, &rt_config.consul.node_name,
opts.get(&"DIPLONAT_CONSUL_NODE_NAME".to_string()).unwrap() opts.get(&"DIPLONAT_CONSUL_NODE_NAME".to_string()).unwrap()
); );
assert_eq!(rt_config.consul.url, CONSUL_URL.to_string()); assert_eq!(rt_config.consul.url, CONSUL_URL.to_string());
assert_eq!( assert_eq!(
rt_config.firewall.refresh_time, rt_config.firewall.refresh_time,
Duration::from_secs(REFRESH_TIME.into()) Duration::from_secs(REFRESH_TIME.into())
); );
assert!(rt_config.igd.private_ip.is_none()); assert!(rt_config.igd.private_ip.is_none());
assert_eq!( assert_eq!(
rt_config.igd.expiration_time, rt_config.igd.expiration_time,
Duration::from_secs(EXPIRATION_TIME.into()) Duration::from_secs(EXPIRATION_TIME.into())
); );
assert_eq!( assert_eq!(
rt_config.igd.refresh_time, rt_config.igd.refresh_time,
Duration::from_secs(REFRESH_TIME.into()) Duration::from_secs(REFRESH_TIME.into())
); );
} }
#[test] #[test]
#[should_panic] #[should_panic]
fn err_from_iter_invalid_refresh_time() { fn err_from_iter_invalid_refresh_time() {
let mut opts = minimal_valid_options(); let mut opts = minimal_valid_options();
opts.insert("DIPLONAT_EXPIRATION_TIME".to_string(), "60".to_string()); opts.insert("DIPLONAT_EXPIRATION_TIME".to_string(), "60".to_string());
opts.insert("DIPLONAT_REFRESH_TIME".to_string(), "60".to_string()); opts.insert("DIPLONAT_REFRESH_TIME".to_string(), "60".to_string());
ConfigOpts::from_iter(opts).unwrap(); ConfigOpts::from_iter(opts).unwrap();
} }
#[test] #[test]
fn ok_from_iter_all_valid_options() { fn ok_from_iter_all_valid_options() {
let opts = all_valid_options(); let opts = all_valid_options();
let rt_config = ConfigOpts::from_iter(opts.clone()).unwrap(); let rt_config = ConfigOpts::from_iter(opts.clone()).unwrap();
let expiration_time = Duration::from_secs( let expiration_time = Duration::from_secs(
opts opts.get(&"DIPLONAT_EXPIRATION_TIME".to_string())
.get(&"DIPLONAT_EXPIRATION_TIME".to_string()) .unwrap()
.unwrap() .parse::<u64>()
.parse::<u64>() .unwrap()
.unwrap() .into(),
.into(), );
); let refresh_time = Duration::from_secs(
let refresh_time = Duration::from_secs( opts.get(&"DIPLONAT_REFRESH_TIME".to_string())
opts .unwrap()
.get(&"DIPLONAT_REFRESH_TIME".to_string()) .parse::<u64>()
.unwrap() .unwrap()
.parse::<u64>() .into(),
.unwrap() );
.into(),
);
assert!(rt_config.acme.is_some()); assert!(rt_config.acme.is_some());
assert_eq!( assert_eq!(
&rt_config.acme.unwrap().email, &rt_config.acme.unwrap().email,
opts.get(&"DIPLONAT_ACME_EMAIL".to_string()).unwrap() opts.get(&"DIPLONAT_ACME_EMAIL".to_string()).unwrap()
); );
assert_eq!( assert_eq!(
&rt_config.consul.node_name, &rt_config.consul.node_name,
opts.get(&"DIPLONAT_CONSUL_NODE_NAME".to_string()).unwrap() opts.get(&"DIPLONAT_CONSUL_NODE_NAME".to_string()).unwrap()
); );
assert_eq!( assert_eq!(
&rt_config.consul.url, &rt_config.consul.url,
opts.get(&"DIPLONAT_CONSUL_URL".to_string()).unwrap() opts.get(&"DIPLONAT_CONSUL_URL".to_string()).unwrap()
); );
assert_eq!(rt_config.firewall.refresh_time, refresh_time); assert_eq!(rt_config.firewall.refresh_time, refresh_time);
assert_eq!( assert_eq!(
&rt_config.igd.private_ip.unwrap(), &rt_config.igd.private_ip.unwrap(),
opts.get(&"DIPLONAT_PRIVATE_IP".to_string()).unwrap() opts.get(&"DIPLONAT_PRIVATE_IP".to_string()).unwrap()
); );
assert_eq!(rt_config.igd.expiration_time, expiration_time); assert_eq!(rt_config.igd.expiration_time, expiration_time);
assert_eq!(rt_config.igd.refresh_time, refresh_time); assert_eq!(rt_config.igd.refresh_time, refresh_time);
} }

243
src/config/runtime.rs
View file

@ -14,179 +14,182 @@ use crate::config::{ConfigOpts, ConfigOptsAcme, ConfigOptsBase, ConfigOptsConsul
#[derive(Debug)] #[derive(Debug)]
pub struct RuntimeConfigAcme { pub struct RuntimeConfigAcme {
pub email: String, pub email: String,
} }
#[derive(Debug)] #[derive(Debug)]
pub struct RuntimeConfigConsul { pub struct RuntimeConfigConsul {
pub node_name: String, pub node_name: String,
pub url: String, pub url: String,
pub tls: Option<(Option<reqwest::Certificate>, bool, reqwest::Identity)>, pub tls: Option<(Option<reqwest::Certificate>, bool, reqwest::Identity)>,
} }
#[derive(Debug)] #[derive(Debug)]
pub struct RuntimeConfigFirewall { pub struct RuntimeConfigFirewall {
pub refresh_time: Duration, pub refresh_time: Duration,
} }
#[derive(Debug)] #[derive(Debug)]
pub struct RuntimeConfigIgd { pub struct RuntimeConfigIgd {
pub private_ip: Option<String>, pub private_ip: Option<String>,
pub expiration_time: Duration, pub expiration_time: Duration,
pub refresh_time: Duration, pub refresh_time: Duration,
} }
#[derive(Debug)] #[derive(Debug)]
pub struct RuntimeConfigStun { pub struct RuntimeConfigStun {
pub stun_server_v4: SocketAddr, pub stun_server_v4: SocketAddr,
pub stun_server_v6: SocketAddr, pub stun_server_v6: SocketAddr,
pub refresh_time: Duration, pub refresh_time: Duration,
} }
#[derive(Debug)] #[derive(Debug)]
pub struct RuntimeConfig { pub struct RuntimeConfig {
pub acme: Option<RuntimeConfigAcme>, pub acme: Option<RuntimeConfigAcme>,
pub consul: RuntimeConfigConsul, pub consul: RuntimeConfigConsul,
pub firewall: RuntimeConfigFirewall, pub firewall: RuntimeConfigFirewall,
pub igd: RuntimeConfigIgd, pub igd: RuntimeConfigIgd,
pub stun: RuntimeConfigStun, pub stun: RuntimeConfigStun,
} }
impl RuntimeConfig { impl RuntimeConfig {
pub fn new(opts: ConfigOpts) -> Result<Self> { pub fn new(opts: ConfigOpts) -> Result<Self> {
let acme = RuntimeConfigAcme::new(opts.acme)?; let acme = RuntimeConfigAcme::new(opts.acme)?;
let consul = RuntimeConfigConsul::new(opts.consul)?; let consul = RuntimeConfigConsul::new(opts.consul)?;
let firewall = RuntimeConfigFirewall::new(&opts.base)?; let firewall = RuntimeConfigFirewall::new(&opts.base)?;
let igd = RuntimeConfigIgd::new(&opts.base)?; let igd = RuntimeConfigIgd::new(&opts.base)?;
let stun = RuntimeConfigStun::new(&opts.base)?; let stun = RuntimeConfigStun::new(&opts.base)?;
Ok(Self { Ok(Self {
acme, acme,
consul, consul,
firewall, firewall,
igd, igd,
stun, stun,
}) })
} }
} }
impl RuntimeConfigAcme { impl RuntimeConfigAcme {
pub fn new(opts: ConfigOptsAcme) -> Result<Option<Self>> { pub fn new(opts: ConfigOptsAcme) -> Result<Option<Self>> {
if !opts.enable { if !opts.enable {
return Ok(None); return Ok(None);
} }
let email = opts.email.expect( let email = opts.email.expect(
"'DIPLONAT_ACME_EMAIL' environment variable is required if 'DIPLONAT_ACME_ENABLE' == 'true'", "'DIPLONAT_ACME_EMAIL' environment variable is required if 'DIPLONAT_ACME_ENABLE' == 'true'",
); );
Ok(Some(Self { email })) Ok(Some(Self { email }))
} }
} }
impl RuntimeConfigConsul { impl RuntimeConfigConsul {
pub(super) fn new(opts: ConfigOptsConsul) -> Result<Self> { pub(super) fn new(opts: ConfigOptsConsul) -> Result<Self> {
let node_name = opts let node_name = opts
.node_name .node_name
.expect("'DIPLONAT_CONSUL_NODE_NAME' environment variable is required"); .expect("'DIPLONAT_CONSUL_NODE_NAME' environment variable is required");
let url = opts.url.unwrap_or(super::CONSUL_URL.to_string()); let url = opts.url.unwrap_or(super::CONSUL_URL.to_string());
let tls = match (&opts.client_cert, &opts.client_key) { let tls = match (&opts.client_cert, &opts.client_key) {
(Some(client_cert), Some(client_key)) => { (Some(client_cert), Some(client_key)) => {
let cert = match &opts.ca_cert { let cert = match &opts.ca_cert {
Some(ca_cert) => { Some(ca_cert) => {
let mut ca_cert_buf = vec![]; let mut ca_cert_buf = vec![];
File::open(ca_cert)?.read_to_end(&mut ca_cert_buf)?; File::open(ca_cert)?.read_to_end(&mut ca_cert_buf)?;
Some(reqwest::Certificate::from_pem(&ca_cert_buf[..])?) Some(reqwest::Certificate::from_pem(&ca_cert_buf[..])?)
} }
None => None, None => None,
};
let mut client_cert_buf = vec![];
File::open(client_cert)?.read_to_end(&mut client_cert_buf)?;
let mut client_key_buf = vec![];
File::open(client_key)?.read_to_end(&mut client_key_buf)?;
let ident = reqwest::Identity::from_pem(
&[&client_cert_buf[..], &client_key_buf[..]].concat()[..],
)?;
Some((cert, opts.tls_skip_verify, ident))
}
(None, None) => None,
_ => bail!("Incomplete TLS configuration parameters"),
}; };
let mut client_cert_buf = vec![]; Ok(Self {
File::open(client_cert)?.read_to_end(&mut client_cert_buf)?; node_name,
url,
let mut client_key_buf = vec![]; tls,
File::open(client_key)?.read_to_end(&mut client_key_buf)?; })
}
let ident =
reqwest::Identity::from_pem(&[&client_cert_buf[..], &client_key_buf[..]].concat()[..])?;
Some((cert, opts.tls_skip_verify, ident))
}
(None, None) => None,
_ => bail!("Incomplete TLS configuration parameters"),
};
Ok(Self {
node_name,
url,
tls,
})
}
} }
impl RuntimeConfigFirewall { impl RuntimeConfigFirewall {
pub(super) fn new(opts: &ConfigOptsBase) -> Result<Self> { pub(super) fn new(opts: &ConfigOptsBase) -> Result<Self> {
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(Self { refresh_time }) Ok(Self { refresh_time })
} }
} }
impl RuntimeConfigIgd { impl RuntimeConfigIgd {
pub(super) fn new(opts: &ConfigOptsBase) -> Result<Self> { pub(super) fn new(opts: &ConfigOptsBase) -> Result<Self> {
let private_ip = opts.private_ip.clone(); let private_ip = opts.private_ip.clone();
let expiration_time = Duration::from_secs( let expiration_time = Duration::from_secs(
opts opts.expiration_time
.expiration_time .unwrap_or(super::EXPIRATION_TIME)
.unwrap_or(super::EXPIRATION_TIME) .into(),
.into(), );
); let refresh_time =
let refresh_time = Duration::from_secs(opts.refresh_time.unwrap_or(super::REFRESH_TIME).into()); Duration::from_secs(opts.refresh_time.unwrap_or(super::REFRESH_TIME).into());
if refresh_time.as_secs() * 2 > expiration_time.as_secs() { if refresh_time.as_secs() * 2 > expiration_time.as_secs() {
return Err(anyhow!( return Err(anyhow!(
"IGD expiration time (currently: {}s) must be at least twice bigger than refresh time \ "IGD expiration time (currently: {}s) must be at least twice bigger than refresh time \
(currently: {}s)", (currently: {}s)",
expiration_time.as_secs(), expiration_time.as_secs(),
refresh_time.as_secs() refresh_time.as_secs()
)); ));
} }
Ok(Self { Ok(Self {
private_ip, private_ip,
expiration_time, expiration_time,
refresh_time, refresh_time,
}) })
} }
} }
impl RuntimeConfigStun { impl RuntimeConfigStun {
pub(super) fn new(opts: &ConfigOptsBase) -> Result<Self> { pub(super) fn new(opts: &ConfigOptsBase) -> Result<Self> {
let mut stun_server_v4 = None; let mut stun_server_v4 = None;
let mut stun_server_v6 = None; let mut stun_server_v6 = None;
for addr in opts for addr in opts
.stun_server .stun_server
.as_deref() .as_deref()
.unwrap_or(super::STUN_SERVER) .unwrap_or(super::STUN_SERVER)
.to_socket_addrs()? .to_socket_addrs()?
{ {
if addr.is_ipv4() { if addr.is_ipv4() {
stun_server_v4 = Some(addr); stun_server_v4 = Some(addr);
} }
if addr.is_ipv6() { if addr.is_ipv6() {
stun_server_v6 = Some(addr); stun_server_v6 = Some(addr);
} }
}
let refresh_time =
Duration::from_secs(opts.refresh_time.unwrap_or(super::REFRESH_TIME).into());
Ok(Self {
stun_server_v4: stun_server_v4
.ok_or(anyhow!("Unable to resolve STUN server's IPv4 address"))?,
stun_server_v6: stun_server_v6
.ok_or(anyhow!("Unable to resolve STUN server's IPv6 address"))?,
refresh_time,
})
} }
let refresh_time = Duration::from_secs(opts.refresh_time.unwrap_or(super::REFRESH_TIME).into());
Ok(Self {
stun_server_v4: stun_server_v4
.ok_or(anyhow!("Unable to resolve STUN server's IPv4 address"))?,
stun_server_v6: stun_server_v6
.ok_or(anyhow!("Unable to resolve STUN server's IPv6 address"))?,
refresh_time,
})
}
} }

122
src/consul.rs
View file

@ -7,80 +7,80 @@ use crate::config::RuntimeConfigConsul;
#[derive(Serialize, Deserialize, Debug)] #[derive(Serialize, Deserialize, Debug)]
pub struct ServiceEntry { pub struct ServiceEntry {
#[serde(rename = "Tags")] #[serde(rename = "Tags")]
pub tags: Vec<String>, pub tags: Vec<String>,
} }
#[derive(Serialize, Deserialize, Debug, Default)] #[derive(Serialize, Deserialize, Debug, Default)]
pub struct CatalogNode { pub struct CatalogNode {
#[serde(rename = "Services")] #[serde(rename = "Services")]
pub services: HashMap<String, ServiceEntry>, pub services: HashMap<String, ServiceEntry>,
} }
pub struct Consul { pub struct Consul {
client: reqwest::Client, client: reqwest::Client,
url: String, url: String,
idx: Option<u64>, idx: Option<u64>,
} }
impl Consul { impl Consul {
pub fn new(config: &RuntimeConfigConsul) -> Self { pub fn new(config: &RuntimeConfigConsul) -> Self {
let client = if let Some((ca, skip_verify, ident)) = config.tls.clone() { let client = if let Some((ca, skip_verify, ident)) = config.tls.clone() {
if skip_verify { if skip_verify {
reqwest::Client::builder() reqwest::Client::builder()
.use_rustls_tls() .use_rustls_tls()
.danger_accept_invalid_certs(true) .danger_accept_invalid_certs(true)
.identity(ident) .identity(ident)
.build() .build()
.expect("Unable to build reqwest client") .expect("Unable to build reqwest client")
} else if let Some(ca) = ca { } else if let Some(ca) = ca {
reqwest::Client::builder() reqwest::Client::builder()
.use_rustls_tls() .use_rustls_tls()
.add_root_certificate(ca) .add_root_certificate(ca)
.identity(ident) .identity(ident)
.build() .build()
.expect("Unable to build reqwest client") .expect("Unable to build reqwest client")
} else { } else {
reqwest::Client::builder() reqwest::Client::builder()
.use_rustls_tls() .use_rustls_tls()
.identity(ident) .identity(ident)
.build() .build()
.expect("Unable to build reqwest client") .expect("Unable to build reqwest client")
} }
} else { } else {
reqwest::Client::new() reqwest::Client::new()
}; };
return Self { return Self {
client, client,
url: config.url.clone(), url: config.url.clone(),
idx: None, idx: None,
}; };
} }
pub fn watch_node_reset(&mut self) -> () { pub fn watch_node_reset(&mut self) -> () {
self.idx = None; self.idx = None;
} }
pub async fn watch_node(&mut self, host: &str) -> Result<CatalogNode> { pub async fn watch_node(&mut self, host: &str) -> Result<CatalogNode> {
let url = match self.idx { let url = match self.idx {
Some(i) => format!("{}/v1/catalog/node/{}?index={}", self.url, host, i), 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?; let http = self.client.get(&url).send().await?;
self.idx = match http.headers().get("X-Consul-Index") { self.idx = match http.headers().get("X-Consul-Index") {
Some(v) => Some(v.to_str()?.parse::<u64>()?), 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: Option<CatalogNode> = http.json().await?; let resp: Option<CatalogNode> = http.json().await?;
return Ok(resp.unwrap_or_default()); return Ok(resp.unwrap_or_default());
} }
pub async fn kv_put(&self, key: &str, bytes: Vec<u8>) -> Result<()> { pub async fn kv_put(&self, key: &str, bytes: Vec<u8>) -> Result<()> {
let url = format!("{}/v1/kv/{}", self.url, key); let url = format!("{}/v1/kv/{}", self.url, key);
let http = self.client.put(&url).body(bytes).send().await?; let http = self.client.put(&url).body(bytes).send().await?;
http.error_for_status()?; http.error_for_status()?;
Ok(()) Ok(())
} }
} }

156
src/consul_actor.rs
View file

@ -11,110 +11,110 @@ use crate::{consul, messages};
#[derive(Serialize, Deserialize, Debug)] #[derive(Serialize, Deserialize, Debug)]
pub enum DiplonatParameter { pub enum DiplonatParameter {
#[serde(rename = "tcp_port")] #[serde(rename = "tcp_port")]
TcpPort(HashSet<u16>), TcpPort(HashSet<u16>),
#[serde(rename = "udp_port")] #[serde(rename = "udp_port")]
UdpPort(HashSet<u16>), UdpPort(HashSet<u16>),
} }
#[derive(Serialize, Deserialize, Debug)] #[derive(Serialize, Deserialize, Debug)]
pub enum DiplonatConsul { pub enum DiplonatConsul {
#[serde(rename = "diplonat")] #[serde(rename = "diplonat")]
Diplonat(Vec<DiplonatParameter>), Diplonat(Vec<DiplonatParameter>),
} }
pub struct ConsulActor { pub struct ConsulActor {
pub rx_open_ports: watch::Receiver<messages::PublicExposedPorts>, pub rx_open_ports: watch::Receiver<messages::PublicExposedPorts>,
consul: consul::Consul, consul: consul::Consul,
node: String, node: String,
retries: u32, 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 { 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 // 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 // eg. 1.2^32 = 341 seconds ~= 5 minutes - ie. after 32 retries we wait 5
// minutes // minutes
return Duration::from_secs(cmp::min( return Duration::from_secs(cmp::min(
max_time.as_secs(), max_time.as_secs(),
1.2f64.powf(retries as f64) as u64, 1.2f64.powf(retries as f64) as u64,
)); ));
} }
fn to_parameters(catalog: &consul::CatalogNode) -> Vec<DiplonatConsul> { fn to_parameters(catalog: &consul::CatalogNode) -> Vec<DiplonatConsul> {
let mut r = Vec::new(); let mut r = Vec::new();
for (_, service_info) in &catalog.services { for (_, service_info) in &catalog.services {
for tag in &service_info.tags { for tag in &service_info.tags {
let diplo_conf: error::Result<DiplonatConsul> = from_str(tag); let diplo_conf: error::Result<DiplonatConsul> = from_str(tag);
match diplo_conf { match diplo_conf {
Ok(conf) => r.push(conf), Ok(conf) => r.push(conf),
Err(e) => debug!("Failed to parse entry {}. {}", tag, e), Err(e) => debug!("Failed to parse entry {}. {}", tag, e),
}; };
}
} }
}
return r; return r;
} }
fn to_open_ports(params: &Vec<DiplonatConsul>) -> messages::PublicExposedPorts { fn to_open_ports(params: &Vec<DiplonatConsul>) -> messages::PublicExposedPorts {
let mut op = messages::PublicExposedPorts { let mut op = messages::PublicExposedPorts {
tcp_ports: HashSet::new(), tcp_ports: HashSet::new(),
udp_ports: HashSet::new(), udp_ports: HashSet::new(),
}; };
for conf in params { for conf in params {
let DiplonatConsul::Diplonat(c) = conf; let DiplonatConsul::Diplonat(c) = conf;
for parameter in c { for parameter in c {
match parameter { match parameter {
DiplonatParameter::TcpPort(p) => op.tcp_ports.extend(p), DiplonatParameter::TcpPort(p) => op.tcp_ports.extend(p),
DiplonatParameter::UdpPort(p) => op.udp_ports.extend(p), DiplonatParameter::UdpPort(p) => op.udp_ports.extend(p),
}; };
}
} }
}
return op; return op;
} }
impl ConsulActor { impl ConsulActor {
pub fn new(config: &RuntimeConfigConsul, node: &str) -> Self { pub fn new(config: &RuntimeConfigConsul, node: &str) -> Self {
let (tx, rx) = watch::channel(messages::PublicExposedPorts { let (tx, rx) = watch::channel(messages::PublicExposedPorts {
tcp_ports: HashSet::new(), tcp_ports: HashSet::new(),
udp_ports: HashSet::new(), udp_ports: HashSet::new(),
}); });
return Self { return Self {
consul: consul::Consul::new(config), consul: consul::Consul::new(config),
rx_open_ports: rx, rx_open_ports: rx,
tx_open_ports: tx, tx_open_ports: tx,
node: node.to_string(), node: node.to_string(),
retries: 0, retries: 0,
}; };
} }
pub async fn listen(&mut self) -> Result<()> { pub async fn listen(&mut self) -> Result<()> {
loop { loop {
let catalog = match self.consul.watch_node(&self.node).await { let catalog = match self.consul.watch_node(&self.node).await {
Ok(c) => c, Ok(c) => c,
Err(e) => { Err(e) => {
self.consul.watch_node_reset(); self.consul.watch_node_reset();
self.retries = cmp::min(std::u32::MAX - 1, self.retries) + 1; self.retries = cmp::min(std::u32::MAX - 1, self.retries) + 1;
let will_retry_in = retry_to_time(self.retries, Duration::from_secs(600)); let will_retry_in = retry_to_time(self.retries, Duration::from_secs(600));
error!( error!(
"Failed to query consul. Will retry in {}s. {}", "Failed to query consul. Will retry in {}s. {}",
will_retry_in.as_secs(), will_retry_in.as_secs(),
e e
); );
sleep(will_retry_in).await; sleep(will_retry_in).await;
continue; continue;
} }
}; };
self.retries = 0; self.retries = 0;
let msg = to_open_ports(&to_parameters(&catalog)); let msg = to_open_ports(&to_parameters(&catalog));
debug!("Extracted configuration: {:#?}", msg); debug!("Extracted configuration: {:#?}", msg);
self.tx_open_ports.send(msg)?; self.tx_open_ports.send(msg)?;
}
} }
}
} }

72
src/diplonat.rs
View file

@ -2,54 +2,54 @@ use anyhow::Result;
use tokio::try_join; use tokio::try_join;
use crate::{ use crate::{
config::ConfigOpts, consul_actor::ConsulActor, fw_actor::FirewallActor, igd_actor::IgdActor, config::ConfigOpts, consul_actor::ConsulActor, fw_actor::FirewallActor, igd_actor::IgdActor,
stun_actor::StunActor, stun_actor::StunActor,
}; };
pub struct Diplonat { pub struct Diplonat {
consul: ConsulActor, consul: ConsulActor,
firewall: FirewallActor, firewall: FirewallActor,
igd: IgdActor, igd: IgdActor,
stun: StunActor, stun: StunActor,
} }
impl Diplonat { impl Diplonat {
pub async fn new() -> Result<Self> { pub async fn new() -> Result<Self> {
let rt_cfg = ConfigOpts::from_env()?; let rt_cfg = ConfigOpts::from_env()?;
println!("{:#?}", rt_cfg); println!("{:#?}", rt_cfg);
let ca = ConsulActor::new(&rt_cfg.consul, &rt_cfg.consul.node_name); let ca = ConsulActor::new(&rt_cfg.consul, &rt_cfg.consul.node_name);
let fw = FirewallActor::new(rt_cfg.firewall.refresh_time, &ca.rx_open_ports).await?; let fw = FirewallActor::new(rt_cfg.firewall.refresh_time, &ca.rx_open_ports).await?;
let ia = IgdActor::new( let ia = IgdActor::new(
rt_cfg.igd.private_ip.as_ref().map(String::as_str), rt_cfg.igd.private_ip.as_ref().map(String::as_str),
rt_cfg.igd.refresh_time, rt_cfg.igd.refresh_time,
rt_cfg.igd.expiration_time, rt_cfg.igd.expiration_time,
&ca.rx_open_ports, &ca.rx_open_ports,
) )
.await?; .await?;
let sa = StunActor::new(&rt_cfg.consul, &rt_cfg.stun, &rt_cfg.consul.node_name); let sa = StunActor::new(&rt_cfg.consul, &rt_cfg.stun, &rt_cfg.consul.node_name);
let ctx = Self { let ctx = Self {
consul: ca, consul: ca,
igd: ia, igd: ia,
firewall: fw, firewall: fw,
stun: sa, stun: sa,
}; };
Ok(ctx) Ok(ctx)
} }
pub async fn listen(&mut self) -> Result<()> { pub async fn listen(&mut self) -> Result<()> {
try_join!( try_join!(
self.consul.listen(), self.consul.listen(),
self.igd.listen(), self.igd.listen(),
self.firewall.listen(), self.firewall.listen(),
self.stun.listen(), self.stun.listen(),
)?; )?;
Ok(()) Ok(())
} }
} }

135
src/fw.rs
View file

@ -8,96 +8,89 @@ use regex::Regex;
use crate::messages; use crate::messages;
pub fn setup(ipt: &iptables::IPTables) -> Result<()> { pub fn setup(ipt: &iptables::IPTables) -> Result<()> {
// ensure we start from a clean state without any rule already set // ensure we start from a clean state without any rule already set
cleanup(ipt)?; cleanup(ipt)?;
info!("{}: creating DIPLONAT chain using", ipt.cmd); info!("{}: creating DIPLONAT chain using", ipt.cmd);
ipt ipt.new_chain("filter", "DIPLONAT")
.new_chain("filter", "DIPLONAT") .context("Failed to create new chain")?;
.context("Failed to create new chain")?; ipt.insert_unique("filter", "INPUT", "-j DIPLONAT", 1)
ipt .context("Failed to insert jump rule")?;
.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<()> { pub fn open_ports(ipt: &iptables::IPTables, ports: messages::PublicExposedPorts) -> Result<()> {
for p in ports.tcp_ports { for p in ports.tcp_ports {
info!("{}: opening TCP port {}", ipt.cmd, p); info!("{}: opening TCP port {}", ipt.cmd, p);
ipt ipt.append(
.append( "filter",
"filter", "DIPLONAT",
"DIPLONAT", &format!("-p tcp --dport {} -j ACCEPT", p),
&format!("-p tcp --dport {} -j ACCEPT", p), )
) .context("Failed to insert port rule")?;
.context("Failed to insert port rule")?; }
}
for p in ports.udp_ports { for p in ports.udp_ports {
info!("{}: opening UDP port {}", ipt.cmd, p); info!("{}: opening UDP port {}", ipt.cmd, p);
ipt ipt.append(
.append( "filter",
"filter", "DIPLONAT",
"DIPLONAT", &format!("-p udp --dport {} -j ACCEPT", p),
&format!("-p udp --dport {} -j ACCEPT", p), )
) .context("Failed to insert port rule")?;
.context("Failed to insert port rule")?; }
}
Ok(()) Ok(())
} }
pub fn get_opened_ports(ipt: &iptables::IPTables) -> Result<messages::PublicExposedPorts> { pub fn get_opened_ports(ipt: &iptables::IPTables) -> Result<messages::PublicExposedPorts> {
let mut ports = messages::PublicExposedPorts { let mut ports = messages::PublicExposedPorts {
tcp_ports: HashSet::new(), tcp_ports: HashSet::new(),
udp_ports: HashSet::new(), udp_ports: HashSet::new(),
}; };
let list = ipt.list("filter", "DIPLONAT")?; let list = ipt.list("filter", "DIPLONAT")?;
let re = Regex::new(r"\-A.*? \-p (\w+).*\-\-dport (\d+).*?\-j ACCEPT") let re = Regex::new(r"\-A.*? \-p (\w+).*\-\-dport (\d+).*?\-j ACCEPT")
.context("Regex matching open ports encountered an unexpected rule")?; .context("Regex matching open ports encountered an unexpected rule")?;
for i in list { for i in list {
let caps = re.captures(&i); let caps = re.captures(&i);
match caps { match caps {
Some(c) => { Some(c) => {
if let (Some(raw_proto), Some(raw_port)) = (c.get(1), c.get(2)) { if let (Some(raw_proto), Some(raw_port)) = (c.get(1), c.get(2)) {
let proto = String::from(raw_proto.as_str()); let proto = String::from(raw_proto.as_str());
let number = String::from(raw_port.as_str()).parse::<u16>()?; let number = String::from(raw_port.as_str()).parse::<u16>()?;
if proto == "tcp" { if proto == "tcp" {
ports.tcp_ports.insert(number); ports.tcp_ports.insert(number);
} else { } else {
ports.udp_ports.insert(number); ports.udp_ports.insert(number);
} }
} else { } else {
error!("Unexpected rule found in DIPLONAT chain") error!("Unexpected rule found in DIPLONAT chain")
}
}
_ => {}
} }
}
_ => {}
} }
}
Ok(ports) Ok(ports)
} }
pub fn cleanup(ipt: &iptables::IPTables) -> Result<()> { pub fn cleanup(ipt: &iptables::IPTables) -> Result<()> {
if ipt.chain_exists("filter", "DIPLONAT")? { if ipt.chain_exists("filter", "DIPLONAT")? {
info!("{}: removing old DIPLONAT chain", ipt.cmd); info!("{}: removing old DIPLONAT chain", ipt.cmd);
ipt ipt.flush_chain("filter", "DIPLONAT")
.flush_chain("filter", "DIPLONAT") .context("Failed to flush the DIPLONAT chain")?;
.context("Failed to flush the DIPLONAT chain")?;
if ipt.exists("filter", "INPUT", "-j DIPLONAT")? { if ipt.exists("filter", "INPUT", "-j DIPLONAT")? {
ipt ipt.delete("filter", "INPUT", "-j DIPLONAT")
.delete("filter", "INPUT", "-j DIPLONAT") .context("Failed to delete jump rule")?;
.context("Failed to delete jump rule")?; }
ipt.delete_chain("filter", "DIPLONAT")
.context("Failed to delete chain")?;
} }
ipt Ok(())
.delete_chain("filter", "DIPLONAT")
.context("Failed to delete chain")?;
}
Ok(())
} }

144
src/fw_actor.rs
View file

@ -4,88 +4,88 @@ use anyhow::Result;
use iptables; use iptables;
use log::*; use log::*;
use tokio::{ use tokio::{
select, select,
sync::watch, sync::watch,
time::{self, Duration}, time::{self, Duration},
}; };
use crate::{fw, messages}; use crate::{fw, messages};
pub struct FirewallActor { pub struct FirewallActor {
pub ipt_v4: iptables::IPTables, pub ipt_v4: iptables::IPTables,
pub ipt_v6: iptables::IPTables, pub ipt_v6: iptables::IPTables,
rx_ports: watch::Receiver<messages::PublicExposedPorts>, rx_ports: watch::Receiver<messages::PublicExposedPorts>,
last_ports: messages::PublicExposedPorts, last_ports: messages::PublicExposedPorts,
refresh: Duration, refresh: Duration,
} }
impl FirewallActor { impl FirewallActor {
pub async fn new( pub async fn new(
refresh: Duration, refresh: Duration,
rxp: &watch::Receiver<messages::PublicExposedPorts>, rxp: &watch::Receiver<messages::PublicExposedPorts>,
) -> Result<Self> { ) -> Result<Self> {
let ctx = Self { let ctx = Self {
ipt_v4: iptables::new(false)?, ipt_v4: iptables::new(false)?,
ipt_v6: iptables::new(true)?, ipt_v6: iptables::new(true)?,
rx_ports: rxp.clone(), rx_ports: rxp.clone(),
last_ports: messages::PublicExposedPorts::new(), last_ports: messages::PublicExposedPorts::new(),
refresh, refresh,
}; };
fw::setup(&ctx.ipt_v4)?; fw::setup(&ctx.ipt_v4)?;
fw::setup(&ctx.ipt_v6)?; fw::setup(&ctx.ipt_v6)?;
return Ok(ctx); return Ok(ctx);
}
pub async fn listen(&mut self) -> Result<()> {
let mut interval = time::interval(self.refresh);
loop {
// 1. Wait for an event
let new_ports = select! {
_ = self.rx_ports.changed() => Some(self.rx_ports.borrow().clone()),
_ = interval.tick() => None,
else => return Ok(()) // Sender dropped, terminate loop.
};
// 2. Update last ports if needed
if let Some(p) = new_ports {
self.last_ports = p;
}
// 3. Update firewall rules
match self.do_fw_update().await {
Ok(()) => debug!("Successfully updated firewall rules"),
Err(e) => error!("An error occured while updating firewall rules. {}", e),
}
}
}
pub async fn do_fw_update(&self) -> Result<()> {
for ipt in [&self.ipt_v4, &self.ipt_v6] {
let curr_opened_ports = fw::get_opened_ports(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 ports_to_open = messages::PublicExposedPorts {
tcp_ports: diff_tcp,
udp_ports: diff_udp,
};
fw::open_ports(ipt, ports_to_open)?;
} }
return Ok(()); pub async fn listen(&mut self) -> Result<()> {
} let mut interval = time::interval(self.refresh);
loop {
// 1. Wait for an event
let new_ports = select! {
_ = self.rx_ports.changed() => Some(self.rx_ports.borrow().clone()),
_ = interval.tick() => None,
else => return Ok(()) // Sender dropped, terminate loop.
};
// 2. Update last ports if needed
if let Some(p) = new_ports {
self.last_ports = p;
}
// 3. Update firewall rules
match self.do_fw_update().await {
Ok(()) => debug!("Successfully updated firewall rules"),
Err(e) => error!("An error occured while updating firewall rules. {}", e),
}
}
}
pub async fn do_fw_update(&self) -> Result<()> {
for ipt in [&self.ipt_v4, &self.ipt_v6] {
let curr_opened_ports = fw::get_opened_ports(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 ports_to_open = messages::PublicExposedPorts {
tcp_ports: diff_tcp,
udp_ports: diff_udp,
};
fw::open_ports(ipt, ports_to_open)?;
}
return Ok(());
}
} }

213
src/igd_actor.rs
View file

@ -4,124 +4,123 @@ use anyhow::{Context, Result};
use igd::{aio::*, PortMappingProtocol}; use igd::{aio::*, PortMappingProtocol};
use log::*; use log::*;
use tokio::{ use tokio::{
select, select,
sync::watch, sync::watch,
time::{self, Duration}, time::{self, Duration},
}; };
use crate::messages; use crate::messages;
pub struct IgdActor { pub struct IgdActor {
last_ports: messages::PublicExposedPorts, last_ports: messages::PublicExposedPorts,
rx_ports: watch::Receiver<messages::PublicExposedPorts>, rx_ports: watch::Receiver<messages::PublicExposedPorts>,
gateway: Gateway, gateway: Gateway,
refresh: Duration, refresh: Duration,
expire: Duration, expire: Duration,
private_ip: String, private_ip: String,
} }
impl IgdActor { impl IgdActor {
pub async fn new( pub async fn new(
priv_ip: Option<&str>, priv_ip: Option<&str>,
refresh: Duration, refresh: Duration,
expire: Duration, expire: Duration,
rxp: &watch::Receiver<messages::PublicExposedPorts>, rxp: &watch::Receiver<messages::PublicExposedPorts>,
) -> Result<Self> { ) -> Result<Self> {
let gw = search_gateway(Default::default()) let gw = search_gateway(Default::default())
.await .await
.context("Failed to find IGD gateway")?; .context("Failed to find IGD gateway")?;
info!("IGD gateway: {}", gw); info!("IGD gateway: {}", gw);
let private_ip = if let Some(ip) = priv_ip { let private_ip = if let Some(ip) = priv_ip {
info!("Using private IP from config: {}", ip); info!("Using private IP from config: {}", ip);
ip.to_string() ip.to_string()
} else { } else {
info!("Trying to automatically detect private IP"); info!("Trying to automatically detect private IP");
let gwa = gw.addr.ip().octets(); let gwa = gw.addr.ip().octets();
let cmplen = match gwa { let cmplen = match gwa {
[192, 168, _, _] => 3, [192, 168, _, _] => 3,
[10, _, _, _] => 2, [10, _, _, _] => 2,
_ => panic!( _ => panic!(
"Gateway IP does not appear to be in a local network ({})", "Gateway IP does not appear to be in a local network ({})",
gw.addr.ip() gw.addr.ip()
), ),
}; };
#[allow(unused_parens)] #[allow(unused_parens)]
let public_ip = get_if_addrs::get_if_addrs()? let public_ip = get_if_addrs::get_if_addrs()?
.into_iter() .into_iter()
.map(|i| i.addr.ip()) .map(|i| i.addr.ip())
.filter(|a| match a { .filter(|a| match a {
std::net::IpAddr::V4(a4) => (a4.octets()[..cmplen] == gwa[..cmplen]), std::net::IpAddr::V4(a4) => (a4.octets()[..cmplen] == gwa[..cmplen]),
_ => false, _ => false,
}) })
.next() .next()
.expect("No interface has an IP on same subnet as gateway") .expect("No interface has an IP on same subnet as gateway")
.to_string(); .to_string();
info!("Found private IP: {}", public_ip); info!("Found private IP: {}", public_ip);
public_ip public_ip
}; };
let ctx = Self { let ctx = Self {
gateway: gw, gateway: gw,
rx_ports: rxp.clone(), rx_ports: rxp.clone(),
private_ip, private_ip,
refresh: refresh, refresh: refresh,
expire: expire, expire: expire,
last_ports: messages::PublicExposedPorts::new(), last_ports: messages::PublicExposedPorts::new(),
}; };
return Ok(ctx); return Ok(ctx);
}
pub async fn listen(&mut self) -> Result<()> {
let mut interval = time::interval(self.refresh);
loop {
// 1. Wait for an event
let new_ports = select! {
_ = self.rx_ports.changed() => Some(self.rx_ports.borrow().clone()),
_ = interval.tick() => None,
else => return Ok(()) // Sender dropped, terminate loop.
};
// 2. Update last ports if needed
if let Some(p) = new_ports {
self.last_ports = p;
}
// 3. Flush IGD requests
match self.do_igd().await {
Ok(()) => debug!("Successfully updated IGD"),
Err(e) => error!("An error occured while updating IGD. {}", e),
}
}
}
pub async fn do_igd(&self) -> Result<()> {
let actions = [
(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?;
debug!("IGD request successful for {:#?} {}", proto, service);
}
} }
return Ok(()); pub async fn listen(&mut self) -> Result<()> {
} let mut interval = time::interval(self.refresh);
loop {
// 1. Wait for an event
let new_ports = select! {
_ = self.rx_ports.changed() => Some(self.rx_ports.borrow().clone()),
_ = interval.tick() => None,
else => return Ok(()) // Sender dropped, terminate loop.
};
// 2. Update last ports if needed
if let Some(p) = new_ports {
self.last_ports = p;
}
// 3. Flush IGD requests
match self.do_igd().await {
Ok(()) => debug!("Successfully updated IGD"),
Err(e) => error!("An error occured while updating IGD. {}", e),
}
}
}
pub async fn do_igd(&self) -> Result<()> {
let actions = [
(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?;
debug!("IGD request successful for {:#?} {}", proto, service);
}
}
return Ok(());
}
} }

16
src/main.rs
View file

@ -13,13 +13,13 @@ use log::*;
#[tokio::main] #[tokio::main]
async fn main() { async fn main() {
pretty_env_logger::init(); pretty_env_logger::init();
info!("Starting Diplonat"); info!("Starting Diplonat");
Diplonat::new() Diplonat::new()
.await .await
.expect("Setup failed") .expect("Setup failed")
.listen() .listen()
.await .await
.expect("A runtime error occured"); .expect("A runtime error occured");
} }

16
src/messages.rs
View file

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

139
src/stun_actor.rs
View file

@ -9,96 +9,95 @@ use crate::config::{RuntimeConfigConsul, RuntimeConfigStun};
use crate::consul; use crate::consul;
pub struct StunActor { pub struct StunActor {
node: String, node: String,
consul: consul::Consul, consul: consul::Consul,
stun_server_v4: SocketAddr, stun_server_v4: SocketAddr,
stun_server_v6: SocketAddr, stun_server_v6: SocketAddr,
refresh_time: Duration, refresh_time: Duration,
} }
#[derive(Serialize, Deserialize, Debug)] #[derive(Serialize, Deserialize, Debug)]
pub struct AutodiscoverResult { pub struct AutodiscoverResult {
pub timestamp: u64, pub timestamp: u64,
pub address: IpAddr, pub address: IpAddr,
} }
impl StunActor { impl StunActor {
pub fn new( pub fn new(
consul_config: &RuntimeConfigConsul, consul_config: &RuntimeConfigConsul,
stun_config: &RuntimeConfigStun, stun_config: &RuntimeConfigStun,
node: &str, node: &str,
) -> Self { ) -> Self {
assert!(stun_config.stun_server_v4.is_ipv4()); assert!(stun_config.stun_server_v4.is_ipv4());
assert!(stun_config.stun_server_v6.is_ipv6()); assert!(stun_config.stun_server_v6.is_ipv6());
Self { Self {
consul: consul::Consul::new(consul_config), consul: consul::Consul::new(consul_config),
node: node.to_string(), node: node.to_string(),
stun_server_v4: stun_config.stun_server_v4, stun_server_v4: stun_config.stun_server_v4,
stun_server_v6: stun_config.stun_server_v6, stun_server_v6: stun_config.stun_server_v6,
refresh_time: stun_config.refresh_time, refresh_time: stun_config.refresh_time,
}
} }
}
pub async fn listen(&mut self) -> Result<()> { pub async fn listen(&mut self) -> Result<()> {
loop { loop {
if let Err(e) = self.autodiscover_ip(self.stun_server_v4).await { if let Err(e) = self.autodiscover_ip(self.stun_server_v4).await {
error!("Unable to autodiscover IPv4 address: {}", e); error!("Unable to autodiscover IPv4 address: {}", e);
} }
if let Err(e) = self.autodiscover_ip(self.stun_server_v6).await { if let Err(e) = self.autodiscover_ip(self.stun_server_v6).await {
error!("Unable to autodiscover IPv6 address: {}", e); error!("Unable to autodiscover IPv6 address: {}", e);
} }
tokio::time::sleep(self.refresh_time).await; tokio::time::sleep(self.refresh_time).await;
}
} }
}
async fn autodiscover_ip(&self, stun_server: SocketAddr) -> Result<()> { async fn autodiscover_ip(&self, stun_server: SocketAddr) -> Result<()> {
let binding_addr = match stun_server.is_ipv4() { let binding_addr = match stun_server.is_ipv4() {
true => "0.0.0.0:34791".parse().unwrap(), true => "0.0.0.0:34791".parse().unwrap(),
false => "[::]:34792".parse().unwrap(), false => "[::]:34792".parse().unwrap(),
}; };
let discovered_addr = get_mapped_addr(stun_server, binding_addr).await?.ip(); let discovered_addr = get_mapped_addr(stun_server, binding_addr).await?.ip();
let consul_key = match stun_server.is_ipv4() { let consul_key = match stun_server.is_ipv4() {
true => { true => {
debug!("Autodiscovered IPv4: {}", discovered_addr); debug!("Autodiscovered IPv4: {}", discovered_addr);
format!("diplonat/autodiscovery/ipv4/{}", self.node) format!("diplonat/autodiscovery/ipv4/{}", self.node)
} }
false => { false => {
debug!("Autodiscovered IPv6: {}", discovered_addr); debug!("Autodiscovered IPv6: {}", discovered_addr);
format!("diplonat/autodiscovery/ipv6/{}", self.node) format!("diplonat/autodiscovery/ipv6/{}", self.node)
} }
}; };
self self.consul
.consul .kv_put(
.kv_put( &consul_key,
&consul_key, serde_json::to_vec(&AutodiscoverResult {
serde_json::to_vec(&AutodiscoverResult { timestamp: SystemTime::now()
timestamp: SystemTime::now() .duration_since(SystemTime::UNIX_EPOCH)?
.duration_since(SystemTime::UNIX_EPOCH)? .as_secs(),
.as_secs(), address: discovered_addr,
address: discovered_addr, })?,
})?, )
) .await?;
.await?;
Ok(()) Ok(())
} }
} }
async fn get_mapped_addr(stun_server: SocketAddr, binding_addr: SocketAddr) -> Result<SocketAddr> { async fn get_mapped_addr(stun_server: SocketAddr, binding_addr: SocketAddr) -> Result<SocketAddr> {
use stun_client::*; use stun_client::*;
let mut client = Client::new(binding_addr, None).await.unwrap(); let mut client = Client::new(binding_addr, None).await.unwrap();
let res = client.binding_request(stun_server, None).await.unwrap(); let res = client.binding_request(stun_server, None).await.unwrap();
if res.get_class() != Class::SuccessResponse { if res.get_class() != Class::SuccessResponse {
bail!("STUN server did not responde with a success response"); bail!("STUN server did not responde with a success response");
} }
let xor_mapped_addr = Attribute::get_xor_mapped_address(&res) let xor_mapped_addr = Attribute::get_xor_mapped_address(&res)
.ok_or(anyhow!("no XorMappedAddress found in STUN response"))?; .ok_or(anyhow!("no XorMappedAddress found in STUN response"))?;
Ok(xor_mapped_addr) Ok(xor_mapped_addr)
} }