1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
use std::collections::HashSet;
use anyhow::Result;
use iptables;
use log::*;
use tokio::{
select,
sync::watch,
time::{self, Duration},
};
use crate::{fw, messages};
pub struct FirewallActor {
pub ipt: iptables::IPTables,
rx_ports: watch::Receiver<messages::PublicExposedPorts>,
last_ports: messages::PublicExposedPorts,
refresh: Duration,
}
impl FirewallActor {
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(),
refresh: _refresh,
};
fw::setup(&ctx.ipt)?;
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! {
Some(ports) = self.rx_ports.recv() => Some(ports),
_ = 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<()> {
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 ports_to_open = messages::PublicExposedPorts {
tcp_ports: diff_tcp,
udp_ports: diff_udp,
};
fw::open_ports(&self.ipt, ports_to_open)?;
return Ok(())
}
}
|
