netget/server/openvpn/
peer.rs

1//! Tracking for OpenVPN peers seen by the control-channel server.
2//!
3//! A "peer" here is a source address that sent a session reset. It is transport
4//! state only — nothing is persisted, and the table is swept of idle entries so
5//! a scan cannot pin every slot forever.
6
7use crate::server::connection::ConnectionId;
8use crate::utils::clock::Instant;
9use std::collections::HashMap;
10use std::net::SocketAddr;
11use std::time::Duration;
12use tokio::sync::RwLock;
13
14/// Whether this server has decided to answer a peer.
15///
16/// The three states are kept distinct on purpose: `Rejected` and "no decision
17/// came back" both stop us answering, but only `Accepted` may ever send bytes,
18/// so there is no state in which an absent decision reads as approval.
19#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20pub enum PeerAdmission {
21    /// A decision is in flight; the peer has not been answered.
22    Deciding,
23    /// Answered with `P_CONTROL_HARD_RESET_SERVER_V2`.
24    Accepted,
25    /// Refused, or no usable decision was produced. Never answered.
26    Rejected,
27}
28
29/// A peer of the OpenVPN control-channel server.
30#[derive(Clone)]
31pub struct Peer {
32    pub connection_id: ConnectionId,
33    pub addr: SocketAddr,
34    /// The client's OpenVPN session id, from its reset packet.
35    pub session_id: u64,
36    /// The key id the client used, echoed back in our replies.
37    pub key_id: u8,
38    pub admission: PeerAdmission,
39
40    /// Whether a non-empty control payload has been seen, so the "feeding it to
41    /// the TLS session" notice is logged once rather than per retransmission.
42    pub saw_control_payload: bool,
43    /// Same, for undecryptable data packets.
44    pub saw_data_packet: bool,
45
46    pub bytes_sent: u64,
47    pub bytes_received: u64,
48    pub last_activity: Instant,
49}
50
51impl Peer {
52    pub fn new(connection_id: ConnectionId, addr: SocketAddr, session_id: u64, key_id: u8) -> Self {
53        Peer {
54            connection_id,
55            addr,
56            session_id,
57            key_id,
58            admission: PeerAdmission::Deciding,
59            saw_control_payload: false,
60            saw_data_packet: false,
61            bytes_sent: 0,
62            bytes_received: 0,
63            last_activity: Instant::now(),
64        }
65    }
66
67    pub fn record_sent(&mut self, bytes: u64) {
68        self.bytes_sent = self.bytes_sent.saturating_add(bytes);
69        self.last_activity = Instant::now();
70    }
71
72    pub fn record_received(&mut self, bytes: u64) {
73        self.bytes_received = self.bytes_received.saturating_add(bytes);
74        self.last_activity = Instant::now();
75    }
76}
77
78/// In-memory peer table.
79pub struct PeerManager {
80    peers: RwLock<HashMap<SocketAddr, Peer>>,
81}
82
83impl PeerManager {
84    pub fn new() -> Self {
85        PeerManager {
86            peers: RwLock::new(HashMap::new()),
87        }
88    }
89
90    pub async fn add_peer(&self, peer: Peer) {
91        self.peers.write().await.insert(peer.addr, peer);
92    }
93
94    pub async fn get_peer(&self, addr: &SocketAddr) -> Option<Peer> {
95        self.peers.read().await.get(addr).cloned()
96    }
97
98    pub async fn update_peer<F>(&self, addr: &SocketAddr, f: F)
99    where
100        F: FnOnce(&mut Peer),
101    {
102        if let Some(peer) = self.peers.write().await.get_mut(addr) {
103            f(peer);
104        }
105    }
106
107    /// Mutate a peer and return whatever the closure produced, or `None` if the
108    /// peer is unknown.
109    pub async fn update_peer_returning<F, T>(&self, addr: &SocketAddr, f: F) -> Option<T>
110    where
111        F: FnOnce(&mut Peer) -> T,
112    {
113        self.peers.write().await.get_mut(addr).map(f)
114    }
115
116    pub async fn set_admission(&self, addr: &SocketAddr, admission: PeerAdmission) {
117        self.update_peer(addr, |p| {
118            p.admission = admission;
119            p.last_activity = Instant::now();
120        })
121        .await;
122    }
123
124    /// Refresh a peer's liveness without changing anything else.
125    pub async fn touch(&self, addr: &SocketAddr) {
126        self.update_peer(addr, |p| p.last_activity = Instant::now())
127            .await;
128    }
129
130    pub async fn remove_peer(&self, addr: &SocketAddr) -> Option<Peer> {
131        self.peers.write().await.remove(addr)
132    }
133
134    /// Drop peers idle for longer than `max_idle` and return them.
135    pub async fn remove_idle_peers(&self, max_idle: Duration) -> Vec<Peer> {
136        let now = Instant::now();
137        let mut peers = self.peers.write().await;
138        let expired: Vec<SocketAddr> = peers
139            .iter()
140            .filter(|(_, p)| now.duration_since(p.last_activity) > max_idle)
141            .map(|(addr, _)| *addr)
142            .collect();
143        expired
144            .into_iter()
145            .filter_map(|addr| peers.remove(&addr))
146            .collect()
147    }
148
149    pub async fn get_all_peers(&self) -> Vec<Peer> {
150        self.peers.read().await.values().cloned().collect()
151    }
152
153    pub async fn count(&self) -> usize {
154        self.peers.read().await.len()
155    }
156}
157
158impl Default for PeerManager {
159    fn default() -> Self {
160        Self::new()
161    }
162}
netget/server/openvpn/peer.rs

netget/server/openvpn/
peer.rs
