hopr_session_server_forwarder/
lib.rs

1//! HOPR session server that bridges TCP/UDP sockets from the Session Exit node to a destination.
2
3pub mod config;
4
5use std::net::SocketAddr;
6
7use hopr_api::types::crypto::prelude::OffchainKeypair;
8use hopr_transport_session::{IncomingSession, ServiceId, SessionTarget, transfer_session};
9use hopr_utils::{
10    network_types::{
11        prelude::ForeignDataMode,
12        udp::{ConnectedUdpStream, UdpStreamParallelism},
13    },
14    parallelize::cpu::spawn_blocking,
15};
16
17use crate::config::SessionIpForwardingConfig;
18
19#[cfg(all(feature = "telemetry", not(test)))]
20lazy_static::lazy_static! {
21    static ref METRIC_ACTIVE_TARGETS: hopr_api::types::telemetry::MultiGauge = hopr_api::types::telemetry::MultiGauge::new(
22        "hopr_session_hoprd_target_connections",
23        "Number of currently active HOPR session target connections on this Exit node",
24        &["type"]
25    ).unwrap();
26}
27
28/// Size of the buffer for forwarding data to/from a TCP stream.
29pub const HOPR_TCP_BUFFER_SIZE: usize = 4096;
30
31/// Size of the buffer for forwarding data to/from a UDP stream.
32pub const HOPR_UDP_BUFFER_SIZE: usize = 16384;
33
34/// Size of the queue (back-pressure) for data incoming from a UDP stream.
35pub const HOPR_UDP_QUEUE_SIZE: usize = 8192;
36
37/// Error type for [`HoprServerIpForwardingReactor`].
38#[derive(Debug, thiserror::Error)]
39pub enum ForwarderError {
40    #[error("{0}")]
41    General(String),
42}
43
44impl ForwarderError {
45    fn general(s: impl std::fmt::Display) -> Self {
46        Self::General(s.to_string())
47    }
48}
49
50/// Implementation of `HoprSessionServer` that facilitates
51/// bridging of TCP or UDP sockets from the Session Exit node to a destination.
52#[derive(Debug, Clone)]
53pub struct HoprServerIpForwardingReactor {
54    keypair: OffchainKeypair,
55    cfg: SessionIpForwardingConfig,
56}
57
58impl HoprServerIpForwardingReactor {
59    pub fn new(keypair: OffchainKeypair, cfg: SessionIpForwardingConfig) -> Self {
60        Self { keypair, cfg }
61    }
62
63    fn all_ips_allowed(&self, addrs: &[SocketAddr]) -> bool {
64        if self.cfg.use_target_allow_list {
65            for addr in addrs {
66                if !self.cfg.target_allow_list.contains(addr) {
67                    tracing::error!(%addr, "address not allowed by the target allow list, denying the target");
68                    return false;
69                }
70                tracing::debug!(%addr, "address allowed by the target allow list, accepting the target");
71            }
72        }
73        true
74    }
75}
76
77pub const SERVICE_ID_LOOPBACK: ServiceId = 0;
78
79#[async_trait::async_trait]
80impl hopr_api::node::HoprSessionServer for HoprServerIpForwardingReactor {
81    type Error = ForwarderError;
82    type Session = IncomingSession;
83
84    #[tracing::instrument(level = "debug", skip(self, session))]
85    async fn process(&self, mut session: IncomingSession) -> Result<(), ForwarderError> {
86        let session_id = *session.session.id();
87        match session.target {
88            SessionTarget::UdpStream(udp_target) => {
89                let kp = self.keypair.clone();
90                let udp_target = spawn_blocking(move || udp_target.unseal(&kp), "udp_unseal")
91                    .await
92                    .map_err(|e| ForwarderError::general(format!("failed to spawn unseal task: {e}")))?
93                    .map_err(|e| ForwarderError::general(format!("cannot unseal target: {e}")))?;
94
95                tracing::debug!(
96                    session_id = ?session_id,
97                    %udp_target,
98                    "binding socket to the UDP server"
99                );
100
101                // In UDP, it is impossible to determine if the target is viable,
102                // so we just take the first resolved address.
103                let resolved_udp_target = udp_target
104                    .clone()
105                    .resolve_tokio()
106                    .await
107                    .map_err(|e| ForwarderError::general(format!("failed to resolve DNS name {udp_target}: {e}")))?
108                    .first()
109                    .ok_or_else(|| ForwarderError::general(format!("failed to resolve DNS name {udp_target}")))?
110                    .to_owned();
111                tracing::debug!(
112                    ?session_id,
113                    %udp_target,
114                    resolution = ?resolved_udp_target,
115                    "UDP target resolved"
116                );
117
118                if !self.all_ips_allowed(&[resolved_udp_target]) {
119                    return Err(ForwarderError::general(format!(
120                        "denied target address {resolved_udp_target}"
121                    )));
122                }
123
124                let mut udp_bridge = ConnectedUdpStream::builder()
125                    .with_buffer_size(HOPR_UDP_BUFFER_SIZE)
126                    .with_counterparty(resolved_udp_target)
127                    .with_foreign_data_mode(ForeignDataMode::Error)
128                    .with_queue_size(HOPR_UDP_QUEUE_SIZE)
129                    .with_receiver_parallelism(
130                        self.cfg
131                            .udp_rx_parallelism
132                            .map(UdpStreamParallelism::Specific)
133                            .unwrap_or(UdpStreamParallelism::Auto),
134                    )
135                    .build(("0.0.0.0", 0))
136                    .map_err(|e| {
137                        ForwarderError::general(format!("could not bridge the incoming session to {udp_target}: {e}"))
138                    })?;
139
140                tracing::debug!(
141                    ?session_id,
142                    %udp_target,
143                    "bridging the session to the UDP server"
144                );
145
146                tokio::task::spawn(async move {
147                    #[cfg(all(feature = "telemetry", not(test)))]
148                    let _g = hopr_api::types::telemetry::MultiGaugeGuard::new(&METRIC_ACTIVE_TARGETS, &["udp"], 1.0);
149
150                    // The Session forwards the termination to the udp_bridge, terminating
151                    // the UDP socket.
152                    match transfer_session(&mut session.session, &mut udp_bridge, HOPR_UDP_BUFFER_SIZE, None).await {
153                        Ok((session_to_stream_bytes, stream_to_session_bytes)) => tracing::info!(
154                            ?session_id,
155                            session_to_stream_bytes,
156                            stream_to_session_bytes,
157                            %udp_target,
158                            "server bridged session to UDP ended"
159                        ),
160                        Err(e) => tracing::error!(
161                            ?session_id,
162                            %udp_target,
163                            error = %e,
164                            "UDP server stream is closed"
165                        ),
166                    }
167                });
168
169                Ok(())
170            }
171            SessionTarget::TcpStream(tcp_target) => {
172                let kp = self.keypair.clone();
173                let tcp_target = spawn_blocking(move || tcp_target.unseal(&kp), "tcp_unseal")
174                    .await
175                    .map_err(|e| ForwarderError::general(format!("failed to spawn unseal task: {e}")))?
176                    .map_err(|e| ForwarderError::general(format!("cannot unseal target: {e}")))?;
177
178                tracing::debug!(?session_id, %tcp_target, "creating a connection to the TCP server");
179
180                // TCP is able to determine which of the resolved multiple addresses is viable,
181                // and therefore we can pass all of them.
182                let resolved_tcp_targets =
183                    tcp_target.clone().resolve_tokio().await.map_err(|e| {
184                        ForwarderError::general(format!("failed to resolve DNS name {tcp_target}: {e}"))
185                    })?;
186                tracing::debug!(
187                    ?session_id,
188                    %tcp_target,
189                    resolution = ?resolved_tcp_targets,
190                    "TCP target resolved"
191                );
192
193                if !self.all_ips_allowed(&resolved_tcp_targets) {
194                    return Err(ForwarderError::general(format!(
195                        "denied target address {resolved_tcp_targets:?}"
196                    )));
197                }
198
199                let strategy = tokio_retry::strategy::FixedInterval::new(self.cfg.tcp_target_retry_delay)
200                    .take(self.cfg.max_tcp_target_retries as usize);
201
202                let mut tcp_bridge = tokio_retry::Retry::start(strategy, || {
203                    tokio::net::TcpStream::connect(resolved_tcp_targets.as_slice())
204                })
205                .await
206                .map_err(|e| {
207                    ForwarderError::general(format!("could not bridge the incoming session to {tcp_target}: {e}"))
208                })?;
209
210                tcp_bridge.set_nodelay(true).map_err(|e| {
211                    ForwarderError::general(format!(
212                        "could not set the TCP_NODELAY option for the bridged session to {tcp_target}: {e}",
213                    ))
214                })?;
215
216                tracing::debug!(
217                    ?session_id,
218                    %tcp_target,
219                    "bridging the session to the TCP server"
220                );
221
222                tokio::task::spawn(async move {
223                    #[cfg(all(feature = "telemetry", not(test)))]
224                    let _g = hopr_api::types::telemetry::MultiGaugeGuard::new(&METRIC_ACTIVE_TARGETS, &["tcp"], 1.0);
225
226                    match transfer_session(&mut session.session, &mut tcp_bridge, HOPR_TCP_BUFFER_SIZE, None).await {
227                        Ok((session_to_stream_bytes, stream_to_session_bytes)) => tracing::info!(
228                            ?session_id,
229                            session_to_stream_bytes,
230                            stream_to_session_bytes,
231                            %tcp_target,
232                            "server bridged session to TCP ended"
233                        ),
234                        Err(error) => tracing::error!(
235                            ?session_id,
236                            %tcp_target,
237                            %error,
238                            "TCP server stream is closed"
239                        ),
240                    }
241                });
242
243                Ok(())
244            }
245            SessionTarget::ExitNode(SERVICE_ID_LOOPBACK) => {
246                tracing::debug!(?session_id, "bridging the session to the loopback service");
247                let (mut reader, mut writer) = tokio::io::split(session.session);
248
249                #[cfg(all(feature = "telemetry", not(test)))]
250                let _g = hopr_api::types::telemetry::MultiGaugeGuard::new(&METRIC_ACTIVE_TARGETS, &["loopback"], 1.0);
251
252                // Uses 4 kB buffer for copying
253                match tokio::io::copy(&mut reader, &mut writer).await {
254                    Ok(copied) => tracing::info!(?session_id, copied, "server loopback session service ended"),
255                    Err(error) => tracing::error!(
256                        ?session_id,
257                        %error,
258                        "server loopback session service ended with an error"
259                    ),
260                }
261
262                Ok(())
263            }
264            SessionTarget::ExitNode(_) => Err(ForwarderError::General(
265                "server does not support internal session processing".into(),
266            )),
267        }
268    }
269}
hopr_session_server_forwarder/lib.rs

hopr_session_server_forwarder/
lib.rs
