adrien/diplonat
1
0
Fork 0
forked from Deuxfleurs/diplonat
Code Issues Pull requests Projects Releases Wiki Activity

Merge pull request 'added rustfmt, a guide about this, and a CI job to enforce code quality' (#10) from adrien/diplonat:meta/formating into main

Browse source 
Reviewed-on: Deuxfleurs/diplonat#10
This commit is contained in:
Quentin 2021-09-17 10:06:51 +02:00
commit 2bbc910999
16 changed files with 457 additions and 334 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
.drone.yml Normal file
View file

@ -0,0 +1,24 @@
---
kind: pipeline
name: default
workspace:
base: /drone/diplonat
steps:
- name: code style
image: rust:1.47
commands:
- rustup component add rustfmt
- cargo fmt --all -- --check
# - name: code quality
# image: rust:1.47
# commands:
# - cargo clippy -- --deny warnings
# - name: test
# image: rust:1.47
# commands:
# - cargo build --verbose --all
# - cargo test --verbose --all

2
.rustfmt.toml Normal file
View file

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

34
CONTRIBUTING.md Normal file
View file

@ -0,0 +1,34 @@
# Contributing to Diplonat
## Development guidelines
### Code formatting
[Our CI pipeline](./.drone.yml) features a verification of the code format, using [rustfmt](https://github.com/rust-lang/rustfmt).
#### Installing rustfmt
You can run `rustfmt` with Rust 1.24 and above.
To install:
```
rustup component add rustfmt
```
#### Usage
To run on Diplonat, launch the following in the root directory:
```
cargo fmt --all
```
This will format the whole repository using the settigs defined in [`.rustfmt.toml`](./.rustfmt.toml): soft tabs of 2 spaces.
#### Auto-format code
You can automate formatting in a number of ways:
* [Setup your IDE to use `rustfmt`](https://github.com/rust-lang/rustfmt#running-rustfmt-from-your-editor).
* Setup a git hook to run `rustfmt` before each commit.

4
README.md
View file

@ -48,6 +48,10 @@ export RUST_LOG=debug
cargo run
```
## Contributing
Refer to [CONTRIBUTING.md](./CONTRIBUTING.md).
## Design Guidelines
Diplonat is made of a set of Components.

6
src/config/mod.rs
View file

@ -4,8 +4,10 @@ mod options_test;
mod runtime;
pub use options::{ConfigOpts, ConfigOptsAcme, ConfigOptsBase, ConfigOptsConsul};
pub use runtime::{RuntimeConfig, RuntimeConfigAcme, RuntimeConfigConsul, RuntimeConfigFirewall, RuntimeConfigIgd};
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";
pub const CONSUL_URL: &str = "http://127.0.0.1:8500";

86
src/config/options.rs
View file

@ -11,66 +11,68 @@ use crate::config::RuntimeConfig;
/// Base configuration options
#[derive(Clone, Default, Deserialize)]
pub struct ConfigOptsBase {
/// 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 and Firewall rules [default: 300]
pub refresh_time: Option<u16>,
/// 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 and Firewall rules [default: 300]
pub refresh_time: Option<u16>,
}
/// ACME configuration options
#[derive(Clone, Default, Deserialize)]
pub struct ConfigOptsAcme {
/// Whether ACME is enabled [default: false]
#[serde(default)]
pub enable: bool,
/// Whether ACME is enabled [default: false]
#[serde(default)]
pub enable: bool,
/// The default domain holder's e-mail [default: None]
pub email: Option<String>,
/// The default domain holder's e-mail [default: None]
pub email: Option<String>,
}
/// 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>,
}
/// Model of all potential configuration options
pub struct ConfigOpts {
pub base: ConfigOptsBase,
pub acme: ConfigOptsAcme,
pub consul: ConfigOptsConsul,
pub base: ConfigOptsBase,
pub acme: ConfigOptsAcme,
pub consul: ConfigOptsConsul,
}
impl ConfigOpts {
pub fn from_env() -> Result<RuntimeConfig> {
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()?;
pub fn from_env() -> Result<RuntimeConfig> {
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()?;
RuntimeConfig::new(Self {
base: base,
consul: consul,
acme: acme,
})
}
RuntimeConfig::new(Self {
base: base,
consul: consul,
acme: acme,
})
}
// 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 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())?;
// 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 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())?;
RuntimeConfig::new(Self {
base: base,
consul: consul,
acme: acme,
})
}
}
RuntimeConfig::new(Self {
base: base,
consul: consul,
acme: acme,
})
}
}

179
src/config/options_test.rs
View file

@ -5,116 +5,125 @@ 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_PRIVATE_IP".to_string(), "172.123.43.555".to_string());
opts.insert("DIPLONAT_CONSUL_NODE_NAME".to_string(), "consul_node".to_string());
opts
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<String, String> {
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
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
}
#[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();
std::env::remove_var("DIPLONAT_PRIVATE_IP");
std::env::remove_var("DIPLONAT_CONSUL_NODE_NAME");
ConfigOpts::from_env().unwrap();
}
#[test]
fn ok_from_iter_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!(rt_config.acme.is_none());
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_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!(rt_config.acme.is_none());
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_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_from_iter_invalid_refresh_time() {
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());
ConfigOpts::from_iter(opts).unwrap();
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());
ConfigOpts::from_iter(opts).unwrap();
}
#[test]
fn ok_from_iter_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 expiration_time = Duration::from_secs(
opts.get(&"DIPLONAT_EXPIRATION_TIME".to_string()).unwrap()
.parse::<u64>().unwrap()
.into());
let refresh_time = Duration::from_secs(
opts.get(&"DIPLONAT_REFRESH_TIME".to_string()).unwrap()
.parse::<u64>().unwrap()
.into());
let expiration_time = Duration::from_secs(
opts
.get(&"DIPLONAT_EXPIRATION_TIME".to_string())
.unwrap()
.parse::<u64>()
.unwrap()
.into(),
);
let refresh_time = Duration::from_secs(
opts
.get(&"DIPLONAT_REFRESH_TIME".to_string())
.unwrap()
.parse::<u64>()
.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()
);
assert_eq!(
&rt_config.consul.url,
opts.get(&"DIPLONAT_CONSUL_URL".to_string()).unwrap()
);
assert_eq!(
rt_config.firewall.refresh_time,
refresh_time
);
assert_eq!(
&rt_config.igd.private_ip,
opts.get(&"DIPLONAT_PRIVATE_IP".to_string()).unwrap()
);
assert_eq!(
rt_config.igd.expiration_time,
expiration_time
);
assert_eq!(
rt_config.igd.refresh_time,
refresh_time
);
}
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()
);
assert_eq!(
&rt_config.consul.url,
opts.get(&"DIPLONAT_CONSUL_URL".to_string()).unwrap()
);
assert_eq!(rt_config.firewall.refresh_time, refresh_time);
assert_eq!(
&rt_config.igd.private_ip,
opts.get(&"DIPLONAT_PRIVATE_IP".to_string()).unwrap()
);
assert_eq!(rt_config.igd.expiration_time, expiration_time);
assert_eq!(rt_config.igd.refresh_time, refresh_time);
}

134
src/config/runtime.rs
View file

@ -1,6 +1,6 @@
use std::time::Duration;
use anyhow::{Result, anyhow};
use anyhow::{anyhow, Result};
use crate::config::{ConfigOpts, ConfigOptsAcme, ConfigOptsBase, ConfigOptsConsul};
@ -11,111 +11,109 @@ use crate::config::{ConfigOpts, ConfigOptsAcme, ConfigOptsBase, ConfigOptsConsul
#[derive(Debug)]
pub struct RuntimeConfigAcme {
pub email: String,
pub email: String,
}
#[derive(Debug)]
pub struct RuntimeConfigConsul {
pub node_name: String,
pub url: String,
pub node_name: String,
pub url: 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 acme: Option<RuntimeConfigAcme>,
pub consul: RuntimeConfigConsul,
pub firewall: RuntimeConfigFirewall,
pub igd: RuntimeConfigIgd,
pub acme: Option<RuntimeConfigAcme>,
pub consul: RuntimeConfigConsul,
pub firewall: RuntimeConfigFirewall,
pub igd: RuntimeConfigIgd,
}
impl RuntimeConfig {
pub fn new(opts: ConfigOpts) -> Result<Self> {
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())?;
pub fn new(opts: ConfigOpts) -> Result<Self> {
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())?;
Ok(Self {
acme,
consul,
firewall,
igd,
})
}
Ok(Self {
acme,
consul,
firewall,
igd,
})
}
}
impl RuntimeConfigAcme {
pub fn new(opts: ConfigOptsAcme) -> Result<Option<Self>> {
if !opts.enable {
return Ok(None);
}
pub fn new(opts: ConfigOptsAcme) -> Result<Option<Self>> {
if !opts.enable {
return Ok(None);
}
let email = opts.email.expect(
"'DIPLONAT_ACME_EMAIL' environment variable is required \
if 'DIPLONAT_ACME_ENABLE' == 'true'");
let email = opts.email.expect(
"'DIPLONAT_ACME_EMAIL' environment variable is required \
if 'DIPLONAT_ACME_ENABLE' == 'true'",
);
Ok(Some(Self {
email,
}))
}
Ok(Some(Self { email }))
}
}
impl RuntimeConfigConsul {
pub(super) fn new(opts: ConfigOptsConsul) -> Result<Self> {
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());
pub(super) fn new(opts: ConfigOptsConsul) -> Result<Self> {
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,
})
}
Ok(Self { node_name, url })
}
}
impl RuntimeConfigFirewall {
pub(super) fn new(opts: ConfigOptsBase) -> Result<Self> {
let refresh_time = Duration::from_secs(
opts.refresh_time.unwrap_or(super::REFRESH_TIME).into());
pub(super) fn new(opts: ConfigOptsBase) -> Result<Self> {
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 {
pub(super) fn new(opts: ConfigOptsBase) -> Result<Self> {
let private_ip = opts.private_ip.expect(
"'DIPLONAT_PRIVATE_IP' environment variable is required");
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());
pub(super) fn new(opts: ConfigOptsBase) -> Result<Self> {
let private_ip = opts
.private_ip
.expect("'DIPLONAT_PRIVATE_IP' environment variable is required");
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(Self {
private_ip,
expiration_time,
refresh_time,
})
}
}
Ok(Self {
private_ip,
expiration_time,
refresh_time,
})
}
}

18
src/consul.rs
View file

@ -1,21 +1,21 @@
use serde::{Serialize, Deserialize};
use anyhow::{anyhow, Result};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use anyhow::{Result, anyhow};
#[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 {
@ -23,7 +23,7 @@ impl Consul {
return Self {
client: reqwest::Client::new(),
url: url.to_string(),
idx: None
idx: None,
};
}
@ -34,16 +34,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);
}
}

41
src/consul_actor.rs
View file

@ -1,24 +1,24 @@
use std::cmp;
use std::time::Duration;
use crate::consul;
use crate::messages;
use anyhow::Result;
use log::*;
use serde::{Deserialize, Serialize};
use serde_lexpr::{error, from_str};
use std::cmp;
use std::collections::HashSet;
use std::time::Duration;
use tokio::sync::watch;
use tokio::time::delay_for;
use anyhow::Result;
use serde::{Serialize, Deserialize};
use serde_lexpr::{from_str,error};
use crate::messages;
use crate::consul;
use std::collections::HashSet;
#[derive(Serialize, Deserialize, Debug)]
pub enum DiplonatParameter {
tcp_port(HashSet<u16>),
udp_port(HashSet<u16>)
udp_port(HashSet<u16>),
}
#[derive(Serialize, Deserialize, Debug)]
pub enum DiplonatConsul {
diplonat(Vec<DiplonatParameter>)
diplonat(Vec<DiplonatParameter>),
}
pub struct ConsulActor {
@ -27,13 +27,16 @@ pub struct ConsulActor {
consul: consul::Consul,
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> {
@ -53,9 +56,9 @@ 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()
udp_ports: HashSet::new(),
};
for conf in params {
@ -73,9 +76,9 @@ fn to_open_ports(params: &Vec<DiplonatConsul>) -> messages::PublicExposedPorts {
impl ConsulActor {
pub fn new(url: &str, node: &str) -> 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 {
@ -95,7 +98,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;
}

22
src/diplonat.rs
View file

@ -16,25 +16,23 @@ impl Diplonat {
pub async fn new() -> Result<Self> {
let rt_cfg = ConfigOpts::from_env()?;
println!("{:#?}", rt_cfg);
let ca = ConsulActor::new(&rt_cfg.consul.url, &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(
&rt_cfg.igd.private_ip,
rt_cfg.igd.refresh_time,
rt_cfg.igd.expiration_time,
&ca.rx_open_ports
).await?;
&rt_cfg.igd.private_ip,
rt_cfg.igd.refresh_time,
rt_cfg.igd.expiration_time,
&ca.rx_open_ports,
)
.await?;
let ctx = Self {
consul: ca,
igd: ia,
firewall: fw
firewall: fw,
};
return Ok(ctx);

126
src/fw.rs
View file

@ -1,81 +1,97 @@
use crate::messages;
use anyhow::{Context, Result};
use iptables;
use log::*;
use regex::Regex;
use std::collections::HashSet;
use crate::messages;
use anyhow::{Result,Context};
use log::*;
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<messages::PublicExposedPorts> {
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::<u16>()?;
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")
}
},
_ => {}
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(())
}

51
src/fw_actor.rs
View file

@ -1,28 +1,29 @@
use anyhow::Result;
use tokio::{
select,
sync::watch,
time::{
self,
Duration
}};
use log::*;
use tokio::{
select,
sync::watch,
time::{self, Duration},
};
use iptables;
use crate::messages;
use crate::fw;
use crate::messages;
use iptables;
use std::collections::HashSet;
pub struct FirewallActor {
pub ipt: iptables::IPTables,
rx_ports: watch::Receiver<messages::PublicExposedPorts>,
last_ports: messages::PublicExposedPorts,
refresh: Duration
refresh: Duration,
}
impl FirewallActor {
pub async fn new(_refresh: Duration, rxp: &watch::Receiver<messages::PublicExposedPorts>) -> Result<Self> {
let ctx = Self {
pub async fn new(
_refresh: Duration,
rxp: &watch::Receiver<messages::PublicExposedPorts>,
) -> Result<Self> {
let ctx = Self {
ipt: iptables::new(false)?,
rx_ports: rxp.clone(),
last_ports: messages::PublicExposedPorts::new(),
@ -30,7 +31,7 @@ impl FirewallActor {
};
fw::setup(&ctx.ipt)?;
return Ok(ctx);
}
@ -45,7 +46,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 {
@ -58,18 +61,26 @@ impl FirewallActor {
pub async fn do_fw_update(&self) -> Result<()> {
let curr_opened_ports = fw::get_opened_ports(&self.ipt)?;
let diff_tcp = self.last_ports.tcp_ports.difference(&curr_opened_ports.tcp_ports).copied().collect::<HashSet<u16>>();
let diff_udp = self.last_ports.udp_ports.difference(&curr_opened_ports.udp_ports).copied().collect::<HashSet<u16>>();
let diff_tcp = self
.last_ports
.tcp_ports
.difference(&curr_opened_ports.tcp_ports)
.copied()
.collect::<HashSet<u16>>();
let diff_udp = self
.last_ports
.udp_ports
.difference(&curr_opened_ports.udp_ports)
.copied()
.collect::<HashSet<u16>>();
let ports_to_open = messages::PublicExposedPorts {
tcp_ports: diff_tcp,
udp_ports: diff_udp
tcp_ports: diff_tcp,
udp_ports: diff_udp,
};
fw::open_ports(&self.ipt, ports_to_open)?;
return Ok(());
}
}

56
src/igd_actor.rs
View file

@ -1,16 +1,14 @@
use crate::messages;
use anyhow::{Context, Result};
use igd::aio::*;
use igd::PortMappingProtocol;
use std::net::SocketAddrV4;
use log::*;
use anyhow::{Result, Context};
use std::net::SocketAddrV4;
use tokio::{
select,
sync::watch,
time::{
self,
Duration
}};
use crate::messages;
select,
sync::watch,
time::{self, Duration},
};
pub struct IgdActor {
last_ports: messages::PublicExposedPorts,
@ -18,23 +16,28 @@ pub struct IgdActor {
gateway: Gateway,
refresh: Duration,
expire: Duration,
private_ip: String
private_ip: String,
}
impl IgdActor {
pub async fn new(priv_ip: &str, refresh: Duration, expire: Duration, rxp: &watch::Receiver<messages::PublicExposedPorts>) -> Result<Self> {
pub async fn new(
priv_ip: &str,
refresh: Duration,
expire: Duration,
rxp: &watch::Receiver<messages::PublicExposedPorts>,
) -> Result<Self> {
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 {
gateway: gw,
rx_ports: rxp.clone(),
private_ip: priv_ip.to_string(),
refresh: refresh,
expire: expire,
last_ports: messages::PublicExposedPorts::new()
last_ports: messages::PublicExposedPorts::new(),
};
return Ok(ctx);
@ -51,7 +54,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 {
@ -63,15 +68,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);
}
}

2
src/main.rs
View file

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

6
src/messages.rs
View file

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