syncweb_core/net/
relay.rs

1use std::time::Duration;
2
3use iroh::PublicKey;
4use serde::{Deserialize, Serialize};
5use tokio::{
6    io::{AsyncReadExt, AsyncWriteExt},
7    net::TcpStream,
8    sync::Mutex,
9    time::timeout,
10};
11
12use crate::error::{Result, SyncwebError};
13use crate::node::identity::DeviceId;
14
15const MAX_FRAME_SIZE: usize = 1024 * 1024;
16
17#[derive(Clone, Copy, Debug, Eq, PartialEq)]
18#[non_exhaustive]
19pub struct JoinRelayRequest {
20    pub device_id: DeviceId,
21}
22
23impl JoinRelayRequest {
24    #[must_use]
25    pub const fn new(device_id: DeviceId) -> Self {
26        Self { device_id }
27    }
28}
29
30#[derive(Clone, Copy, Debug, Eq, PartialEq)]
31#[non_exhaustive]
32pub struct SessionInvitation {
33    pub session_key: [u8; 32],
34    pub server_socket: bool,
35}
36
37impl SessionInvitation {
38    #[must_use]
39    pub const fn new(session_key: [u8; 32], server_socket: bool) -> Self {
40        Self {
41            session_key,
42            server_socket,
43        }
44    }
45}
46
47#[derive(Clone, Copy, Debug, Eq, PartialEq)]
48#[non_exhaustive]
49pub struct JoinSessionRequest {
50    pub session_key: [u8; 32],
51    pub device_id: DeviceId,
52}
53
54impl JoinSessionRequest {
55    #[must_use]
56    pub const fn new(session_key: [u8; 32], device_id: DeviceId) -> Self {
57        Self { session_key, device_id }
58    }
59}
60
61#[derive(Clone, Copy, Debug, Eq, PartialEq)]
62#[non_exhaustive]
63pub enum RelayMessage {
64    JoinRelayRequest(JoinRelayRequest),
65    SessionInvitation(SessionInvitation),
66    JoinSessionRequest(JoinSessionRequest),
67    ResponseSuccess,
68    ResponseNotFound,
69    RelayFull,
70}
71
72impl RelayMessage {
73    #[must_use]
74    pub fn encode(&self) -> Vec<u8> {
75        let mut bytes = Vec::with_capacity(66);
76        match self {
77            Self::JoinRelayRequest(request) => {
78                bytes.push(1);
79                bytes.extend_from_slice(request.device_id.node_id().as_bytes());
80            }
81            Self::SessionInvitation(invitation) => {
82                bytes.push(2);
83                bytes.extend_from_slice(&invitation.session_key);
84                bytes.push(u8::from(invitation.server_socket));
85            }
86            Self::JoinSessionRequest(request) => {
87                bytes.push(3);
88                bytes.extend_from_slice(&request.session_key);
89                bytes.extend_from_slice(request.device_id.node_id().as_bytes());
90            }
91            Self::ResponseSuccess => bytes.push(4),
92            Self::ResponseNotFound => bytes.push(5),
93            Self::RelayFull => bytes.push(6),
94        }
95        bytes
96    }
97
98    /// # Errors
99    ///
100    /// Returns an error if the bytes cannot be decoded into a relay message.
101    pub fn decode(bytes: &[u8]) -> Result<Self> {
102        let (tag, body) = bytes
103            .split_first()
104            .ok_or_else(|| SyncwebError::RelayDecode("message is missing a type tag".to_owned()))?;
105        match (*tag, body) {
106            (1, device_id) if device_id.len() == 32 => Ok(Self::JoinRelayRequest(JoinRelayRequest {
107                device_id: device_id_from_bytes(device_id)?,
108            })),
109            (2, [session_key_bytes @ .., server_socket]) if session_key_bytes.len() == 32 => {
110                if *server_socket > 1 {
111                    return Err(SyncwebError::RelayDecode(
112                        "invalid session invitation socket flag".to_owned(),
113                    ));
114                }
115                let session_key = session_key_bytes
116                    .try_into()
117                    .map_err(|error| SyncwebError::RelayDecode(format!("invalid session key length: {error}")))?;
118                Ok(Self::SessionInvitation(SessionInvitation {
119                    session_key,
120                    server_socket: *server_socket == 1,
121                }))
122            }
123            (3, body_bytes) if body_bytes.len() == 64 => {
124                let session_key_bytes = body_bytes
125                    .get(..32)
126                    .ok_or_else(|| SyncwebError::RelayDecode("invalid body length".to_owned()))?;
127                let device_id = body_bytes
128                    .get(32..)
129                    .ok_or_else(|| SyncwebError::RelayDecode("invalid body length".to_owned()))?;
130                let session_key = session_key_bytes
131                    .try_into()
132                    .map_err(|error| SyncwebError::RelayDecode(format!("invalid session key length: {error}")))?;
133                Ok(Self::JoinSessionRequest(JoinSessionRequest {
134                    session_key,
135                    device_id: device_id_from_bytes(device_id)?,
136                }))
137            }
138            (4, []) => Ok(Self::ResponseSuccess),
139            (5, []) => Ok(Self::ResponseNotFound),
140            (6, []) => Ok(Self::RelayFull),
141            _ => Err(SyncwebError::RelayDecode("invalid message encoding".to_owned())),
142        }
143    }
144}
145
146fn device_id_from_bytes(bytes: &[u8]) -> Result<DeviceId> {
147    let arr: [u8; 32] = bytes
148        .try_into()
149        .map_err(|error| SyncwebError::RelayDecode(format!("device ID must be 32 bytes: {error}")))?;
150    let public_key = PublicKey::from_bytes(&arr).map_err(|error| SyncwebError::RelayDecode(error.to_string()))?;
151    Ok(DeviceId::from_node_id(public_key))
152}
153
154#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
155#[non_exhaustive]
156pub struct RelayConfig {
157    #[serde(default)]
158    pub relay_urls: Vec<String>,
159    #[serde(with = "duration_seconds")]
160    pub timeout: Duration,
161    #[serde(default = "default_auto_fallback")]
162    pub auto_fallback: bool,
163}
164
165impl Default for RelayConfig {
166    fn default() -> Self {
167        Self {
168            relay_urls: Vec::new(),
169            timeout: Duration::from_secs(10),
170            auto_fallback: true,
171        }
172    }
173}
174
175const fn default_auto_fallback() -> bool {
176    true
177}
178
179mod duration_seconds {
180    use std::time::Duration;
181
182    use serde::{Deserialize, Deserializer, Serializer};
183
184    pub fn serialize<S>(duration: &Duration, serializer: S) -> Result<S::Ok, S::Error>
185    where
186        S: Serializer,
187    {
188        serializer.serialize_u64(duration.as_secs())
189    }
190
191    pub fn deserialize<'de, D>(deserializer: D) -> Result<Duration, D::Error>
192    where
193        D: Deserializer<'de>,
194    {
195        let seconds = u64::deserialize(deserializer)?;
196        Ok(Duration::from_secs(seconds))
197    }
198}
199
200/// A length-delimited byte tunnel negotiated using Syncthing relay v1 messages.
201///
202/// The TLS/BEP authentication layer is intentionally left to the caller that owns
203/// Syncthing-compatible credentials; this type only handles the relay's typed
204/// control messages and the subsequent byte tunnel.
205pub struct SyncthingRelayTransport {
206    stream: Mutex<TcpStream>,
207    pub relay_url: String,
208    pub session_key: [u8; 32],
209}
210
211impl SyncthingRelayTransport {
212    /// # Errors
213    ///
214    /// Returns an error if the connection to the relay fails.
215    pub async fn connect(
216        relay_url_in: impl Into<String>,
217        device_id: DeviceId,
218        timeout_duration: Duration,
219    ) -> Result<Self> {
220        let relay_url = relay_url_in.into();
221        let address = relay_address(&relay_url)?;
222        let stream = timeout(timeout_duration, TcpStream::connect(&address))
223            .await
224            .map_err(|error| SyncwebError::operation("relay connection timed out", error))?
225            .map_err(|error| SyncwebError::operation("failed to connect to relay", error))?;
226        let transport = Self {
227            stream: Mutex::new(stream),
228            relay_url,
229            session_key: [0; 32],
230        };
231        transport
232            .send_message(&RelayMessage::JoinRelayRequest(JoinRelayRequest { device_id }))
233            .await?;
234        Ok(transport)
235    }
236
237    /// # Errors
238    ///
239    /// Returns an error if the relay message fails to send.
240    pub async fn send_message(&self, message: &RelayMessage) -> Result<()> {
241        self.write_frame(&message.encode()).await
242    }
243
244    /// # Errors
245    ///
246    /// Returns an error if receiving a relay message fails.
247    pub async fn recv_message(&self) -> Result<RelayMessage> {
248        RelayMessage::decode(&self.read_frame().await?)
249    }
250
251    /// # Errors
252    ///
253    /// Returns an error if the packet fails to send.
254    pub async fn send_packet(&self, packet: &[u8]) -> Result<()> {
255        self.write_frame(packet).await
256    }
257
258    /// # Errors
259    ///
260    /// Returns an error if receiving the packet fails.
261    pub async fn recv_packet(&self) -> Result<Vec<u8>> {
262        self.read_frame().await
263    }
264
265    async fn write_frame(&self, payload: &[u8]) -> Result<()> {
266        if payload.len() > MAX_FRAME_SIZE {
267            return Err(SyncwebError::RelayFrameTooLarge { max: MAX_FRAME_SIZE });
268        }
269        let len = u32::try_from(payload.len())
270            .map_err(|error| SyncwebError::operation("payload length exceeds u32::MAX", error))?;
271        let mut stream = self.stream.lock().await;
272        stream.write_u32(len).await?;
273        stream.write_all(payload).await?;
274        stream.flush().await?;
275        drop(stream);
276        Ok(())
277    }
278
279    async fn read_frame(&self) -> Result<Vec<u8>> {
280        let mut stream = self.stream.lock().await;
281        let length = usize::try_from(stream.read_u32().await?)
282            .map_err(|error| SyncwebError::operation("frame length exceeds usize::MAX", error))?;
283        if length > MAX_FRAME_SIZE {
284            return Err(SyncwebError::RelayFrameTooLarge { max: MAX_FRAME_SIZE });
285        }
286        let mut payload = vec![0; length];
287        stream.read_exact(&mut payload).await?;
288        drop(stream);
289        Ok(payload)
290    }
291}
292
293#[derive(Clone, Debug, Default)]
294pub struct TransportFallback {
295    config: RelayConfig,
296}
297
298impl TransportFallback {
299    #[must_use]
300    pub const fn new(config: RelayConfig) -> Self {
301        Self { config }
302    }
303
304    #[must_use]
305    pub const fn config(&self) -> &RelayConfig {
306        &self.config
307    }
308
309    /// # Errors
310    ///
311    /// Returns an error if connecting to the device via relay fails.
312    pub async fn connect_relay(&self, device_id: DeviceId) -> Result<SyncthingRelayTransport> {
313        if !self.config.auto_fallback {
314            return Err(SyncwebError::RelayDisabled);
315        }
316        let mut failures = Vec::new();
317        for relay_url in &self.config.relay_urls {
318            match SyncthingRelayTransport::connect(relay_url, device_id, self.config.timeout).await {
319                Ok(transport) => return Ok(transport),
320                Err(error) => failures.push(format!("{relay_url}: {error}")),
321            }
322        }
323        Err(SyncwebError::RelayUnreachable {
324            reasons: if failures.is_empty() {
325                "no relay URLs configured".to_owned()
326            } else {
327                failures.join("; ")
328            },
329        })
330    }
331}
332
333fn relay_address(relay_url: &str) -> Result<String> {
334    let address = relay_url.strip_prefix("tcp://").ok_or(SyncwebError::RelayBadScheme)?;
335    if address.is_empty() || address.contains('/') {
336        return Err(SyncwebError::RelayBadAddress(relay_url.to_owned()));
337    }
338    if address.parse::<std::net::SocketAddr>().is_err() && !address.contains(':') {
339        return Err(SyncwebError::RelayBadAddress(relay_url.to_owned()));
340    }
341    Ok(address.to_owned())
342}
syncweb_core/net/relay.rs

syncweb_core/net/
relay.rs
