syncweb_core/net/relay/
protocol.rs

1//! bep-relay protocol messages and their framing.
2//!
3//! Every message is a fixed 12-byte header (big-endian `magic`, `messageType`,
4//! `messageLength`) followed by an XDR-encoded payload of at most
5//! [`MAX_MESSAGE_LENGTH`] bytes. Long-lived relay *sessions* carry raw bytes
6//! using the same framing (`u32` length prefix, no padding).
7//!
8//! Upstream reference: <https://github.com/syncthing/syncthing/tree/main/lib/relay/protocol>.
9
10use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
11
12use crate::error::{Result, SyncwebError};
13
14use super::xdr::{Reader, push_bool, push_bytes, push_u16, push_u32};
15
16/// The bep-relay ALPN identifier used for the TLS join connection.
17pub const PROTOCOL_NAME: &[u8] = b"bep-relay";
18/// XDR magic identifying a bep-relay header.
19pub const MAGIC: u32 = 0x9E79_BC40;
20/// Maximum payload size accepted by relay servers.
21pub const MAX_MESSAGE_LENGTH: usize = 1024;
22/// Safety cap for session data frames (QUIC datagrams are well below this).
23pub const MAX_DATA_FRAME: usize = 8192;
24/// Maximum length of the byte fields in connect/invitation messages.
25pub const MAX_BYTE_FIELD: usize = 32;
26
27const TYPE_PING: u8 = 0;
28const TYPE_PONG: u8 = 1;
29const TYPE_JOIN_RELAY_REQUEST: u8 = 2;
30const TYPE_JOIN_SESSION_REQUEST: u8 = 3;
31const TYPE_RESPONSE: u8 = 4;
32const TYPE_CONNECT_REQUEST: u8 = 5;
33const TYPE_SESSION_INVITATION: u8 = 6;
34const TYPE_RELAY_FULL: u8 = 7;
35
36/// One bep-relay control message.
37#[derive(Clone, Debug, Eq, PartialEq)]
38#[non_exhaustive]
39pub enum Message {
40    /// Periodic keepalive sent by either side.
41    Ping,
42    /// Keepalive acknowledgement.
43    Pong,
44    /// Join the relay registration with an optional access token.
45    JoinRelayRequest { token: String },
46    /// Propose to join an existing session with its shared key.
47    JoinSessionRequest { key: Vec<u8> },
48    /// Status reply to a previous request.
49    Response { code: i32, message: String },
50    /// Ask the relay to broker a session towards a device id.
51    ConnectRequest { id: Vec<u8> },
52    /// An invitation to a session, delivered to the dialed device on its
53    /// registered connection and to the dialing device on the fresh
54    /// connection that carried the connection request.
55    SessionInvitation {
56        from: Vec<u8>,
57        key: Vec<u8>,
58        address: Vec<u8>,
59        port: u16,
60        server_socket: bool,
61    },
62    /// The relay has no capacity left.
63    RelayFull,
64}
65
66impl Message {
67    /// Response code for a successful request.
68    pub const RESPONSE_SUCCESS: i32 = 0;
69    /// Response code: the dialed device is unknown to the relay.
70    pub const RESPONSE_NOT_FOUND: i32 = 1;
71    /// Response code: the device is already connected to the relay.
72    pub const RESPONSE_ALREADY_CONNECTED: i32 = 2;
73    /// Response code: the join token was rejected.
74    pub const RESPONSE_WRONG_TOKEN: i32 = 3;
75    /// Response code: an unexpected message arrived.
76    pub const RESPONSE_UNEXPECTED: i32 = 100;
77
78    const fn type_id(&self) -> u8 {
79        match self {
80            Self::Ping => TYPE_PING,
81            Self::Pong => TYPE_PONG,
82            Self::JoinRelayRequest { .. } => TYPE_JOIN_RELAY_REQUEST,
83            Self::JoinSessionRequest { .. } => TYPE_JOIN_SESSION_REQUEST,
84            Self::Response { .. } => TYPE_RESPONSE,
85            Self::ConnectRequest { .. } => TYPE_CONNECT_REQUEST,
86            Self::SessionInvitation { .. } => TYPE_SESSION_INVITATION,
87            Self::RelayFull => TYPE_RELAY_FULL,
88        }
89    }
90
91    /// Encode `self` into a complete wire frame (header + payload).
92    ///
93    /// # Errors
94    ///
95    /// Returns an error if the payload exceeds [`MAX_MESSAGE_LENGTH`].
96    pub fn encode_frame(&self) -> Result<Vec<u8>> {
97        let mut payload = Vec::new();
98        match self {
99            Self::Ping | Self::Pong | Self::RelayFull => {}
100            Self::JoinRelayRequest { token } => {
101                push_bytes(&mut payload, token.as_bytes()).map_err(decode_error)?;
102            }
103            Self::JoinSessionRequest { key } => {
104                push_bytes(&mut payload, key).map_err(decode_error)?;
105            }
106            Self::Response { code, message } => {
107                push_u32(&mut payload, u32::from_le_bytes(code.to_le_bytes()));
108                push_bytes(&mut payload, message.as_bytes()).map_err(decode_error)?;
109            }
110            Self::ConnectRequest { id } => {
111                push_bytes(&mut payload, id).map_err(decode_error)?;
112            }
113            Self::SessionInvitation {
114                from,
115                key,
116                address,
117                port,
118                server_socket,
119            } => {
120                push_bytes(&mut payload, from).map_err(decode_error)?;
121                push_bytes(&mut payload, key).map_err(decode_error)?;
122                push_bytes(&mut payload, address).map_err(decode_error)?;
123                push_u16(&mut payload, *port);
124                push_bool(&mut payload, *server_socket);
125            }
126        }
127        if payload.len() > MAX_MESSAGE_LENGTH {
128            return Err(SyncwebError::RelayFrameTooLarge {
129                max: MAX_MESSAGE_LENGTH,
130            });
131        }
132        let mut frame = Vec::with_capacity(payload.len().saturating_add(12));
133        push_u32(&mut frame, MAGIC);
134        push_u32(&mut frame, u32::from(self.type_id()));
135        push_u32(
136            &mut frame,
137            u32::try_from(payload.len())
138                .map_err(|error| SyncwebError::RelayDecode(format!("payload length exceeds u32::MAX: {error}")))?,
139        );
140        frame.extend_from_slice(&payload);
141        Ok(frame)
142    }
143
144    /// Decode a complete wire frame (header + payload) into a message.
145    ///
146    /// # Errors
147    ///
148    /// Returns an error if the frame is malformed, truncated, or carries an
149    /// unknown message type.
150    pub fn decode_frame(frame: &[u8]) -> Result<Self> {
151        let (rest, payload) = frame
152            .split_at_checked(12)
153            .ok_or_else(|| SyncwebError::RelayDecode("frame is shorter than the header".to_owned()))?;
154        let header: [u8; 12] = rest
155            .try_into()
156            .map_err(|error| SyncwebError::RelayDecode(format!("frame header is not twelve bytes: {error}")))?;
157        let (magic, message_type, length_words) = read_header_words(header);
158        if magic != MAGIC {
159            return Err(SyncwebError::RelayDecode(format!("bad frame magic {magic:#x}")));
160        }
161        let length = usize::try_from(length_words)
162            .map_err(|error| SyncwebError::RelayDecode(format!("frame length exceeds usize: {error}")))?;
163        if length > MAX_MESSAGE_LENGTH {
164            return Err(SyncwebError::RelayFrameTooLarge {
165                max: MAX_MESSAGE_LENGTH,
166            });
167        }
168        if payload.len() != length {
169            return Err(SyncwebError::RelayDecode(format!(
170                "frame payload is {} bytes but the header announces {length}",
171                payload.len()
172            )));
173        }
174        Self::decode_payload(message_type, payload)
175    }
176
177    fn decode_payload(message_type: u8, payload: &[u8]) -> Result<Self> {
178        let mut reader = Reader::new(payload);
179        let message = match message_type {
180            TYPE_PING => Message::Ping,
181            TYPE_PONG => Message::Pong,
182            TYPE_RELAY_FULL => Message::RelayFull,
183            TYPE_JOIN_RELAY_REQUEST => {
184                let token = reader.take_bytes(MAX_MESSAGE_LENGTH).map_err(decode_error)?;
185                Message::JoinRelayRequest {
186                    token: utf8_field(token, "token")?,
187                }
188            }
189            TYPE_JOIN_SESSION_REQUEST => Message::JoinSessionRequest {
190                key: reader.take_bytes(MAX_BYTE_FIELD).map_err(decode_error)?,
191            },
192            TYPE_RESPONSE => {
193                let code = i32::from_ne_bytes(reader.take_u32().map_err(decode_error)?.to_ne_bytes());
194                let message = reader.take_bytes(MAX_MESSAGE_LENGTH).map_err(decode_error)?;
195                Message::Response {
196                    code,
197                    message: utf8_field(message, "response message")?,
198                }
199            }
200            TYPE_CONNECT_REQUEST => Message::ConnectRequest {
201                id: reader.take_bytes(MAX_BYTE_FIELD).map_err(decode_error)?,
202            },
203            TYPE_SESSION_INVITATION => {
204                let from = reader.take_bytes(MAX_BYTE_FIELD).map_err(decode_error)?;
205                let key = reader.take_bytes(MAX_BYTE_FIELD).map_err(decode_error)?;
206                let address = reader.take_bytes(MAX_BYTE_FIELD).map_err(decode_error)?;
207                let port_words = reader.take_u32().map_err(decode_error)?;
208                let port_bytes = port_words.to_be_bytes();
209                let port = u16::from_be_bytes([port_bytes[2], port_bytes[3]]);
210                let server_socket = reader.take_bool().map_err(decode_error)?;
211                Message::SessionInvitation {
212                    from,
213                    key,
214                    address,
215                    port,
216                    server_socket,
217                }
218            }
219            other => {
220                return Err(SyncwebError::RelayDecode(format!("unknown message type {other}")));
221            }
222        };
223        reader.finish().map_err(decode_error)?;
224        Ok(message)
225    }
226}
227
228/// Wrap an encoding or decoding detail message in [`SyncwebError::RelayDecode`].
229fn decode_error(detail: impl Into<String>) -> SyncwebError {
230    SyncwebError::RelayDecode(detail.into())
231}
232
233fn utf8_field(bytes: Vec<u8>, field: &str) -> Result<String> {
234    String::from_utf8(bytes).map_err(|error| SyncwebError::RelayDecode(format!("{field} is not UTF-8: {error}")))
235}
236
237/// Write `message` as a single length-prefixed frame to `writer`.
238///
239/// # Errors
240///
241/// Returns an error if encoding or the underlying write fails.
242pub async fn write_message<W>(writer: &mut W, message: &Message) -> Result<()>
243where
244    W: AsyncWrite + Unpin,
245{
246    let frame = message.encode_frame()?;
247    writer.write_all(&frame).await?;
248    writer.flush().await?;
249    Ok(())
250}
251
252/// Read one length-prefixed message from `reader`.
253///
254/// # Errors
255///
256/// Returns an error on EOF or if the frame is malformed.
257pub async fn read_message<R>(reader: &mut R) -> Result<Message>
258where
259    R: AsyncRead + Unpin,
260{
261    let mut header = [0_u8; 12];
262    reader.read_exact(&mut header).await?;
263    let (magic, message_type, length_words) = read_header_words(header);
264    if magic != MAGIC {
265        return Err(SyncwebError::RelayDecode(format!("bad frame magic {magic:#x}")));
266    }
267    let length = usize::try_from(length_words)
268        .map_err(|error| SyncwebError::RelayDecode(format!("frame length exceeds usize: {error}")))?;
269    if length > MAX_MESSAGE_LENGTH {
270        return Err(SyncwebError::RelayFrameTooLarge {
271            max: MAX_MESSAGE_LENGTH,
272        });
273    }
274    let mut payload = vec![0_u8; length];
275    reader.read_exact(&mut payload).await?;
276    Message::decode_payload(message_type, &payload)
277}
278
279/// Decode the three big-endian words of a 12-byte header.
280fn read_header_words(header: [u8; 12]) -> (u32, u8, u32) {
281    let magic = u32::from_be_bytes([header[0], header[1], header[2], header[3]]);
282    let type_words = u32::from_be_bytes([header[4], header[5], header[6], header[7]]);
283    let length = u32::from_be_bytes([header[8], header[9], header[10], header[11]]);
284    // The type word must fit in a single byte; default to `u8::MAX` for
285    // anything larger so decoding reports an unknown type cleanly.
286    let message_type = u8::try_from(type_words).unwrap_or(u8::MAX);
287    (magic, message_type, length)
288}
289
290/// Write a session data frame (`u32` length prefix, unpadded).
291///
292/// # Errors
293///
294/// Returns an error if `data` is too large or the underlying write fails.
295pub async fn write_data_frame<W>(writer: &mut W, data: &[u8]) -> Result<()>
296where
297    W: AsyncWrite + Unpin,
298{
299    if data.len() > MAX_DATA_FRAME {
300        return Err(SyncwebError::RelayFrameTooLarge { max: MAX_DATA_FRAME });
301    }
302    let length = u32::try_from(data.len())
303        .map_err(|error| SyncwebError::operation("data frame length exceeds u32::MAX", error))?;
304    writer.write_all(&length.to_be_bytes()).await?;
305    writer.write_all(data).await?;
306    Ok(())
307}
308
309/// Read one session data frame.
310///
311/// Returns an [`std::io::Error`] on EOF, propagating the transport closure so
312/// sessions can be re-established.
313///
314/// # Errors
315///
316/// Returns an error on EOF or if the frame is malformed or oversized.
317pub async fn read_data_frame<R>(reader: &mut R) -> Result<Vec<u8>>
318where
319    R: AsyncRead + Unpin,
320{
321    let mut length_bytes = [0_u8; 4];
322    reader.read_exact(&mut length_bytes).await?;
323    let length = usize::try_from(u32::from_be_bytes(length_bytes))
324        .map_err(|error| SyncwebError::operation("data frame length exceeds usize", error))?;
325    if length > MAX_DATA_FRAME {
326        return Err(SyncwebError::RelayFrameTooLarge { max: MAX_DATA_FRAME });
327    }
328    let mut data = vec![0_u8; length];
329    reader.read_exact(&mut data).await?;
330    Ok(data)
331}
332
333#[cfg(test)]
334mod tests {
335    use super::*;
336
337    #[test]
338    fn every_message_round_trips() {
339        let messages = [
340            Message::Ping,
341            Message::Pong,
342            Message::RelayFull,
343            Message::JoinRelayRequest {
344                token: "thunder".to_owned(),
345            },
346            Message::JoinSessionRequest { key: vec![1_u8; 32] },
347            Message::Response {
348                code: 0,
349                message: "success".to_owned(),
350            },
351            Message::ConnectRequest { id: vec![7_u8; 32] },
352            Message::SessionInvitation {
353                from: vec![9_u8; 32],
354                key: vec![3_u8; 32],
355                address: vec![10, 0, 0, 1],
356                port: 22_067,
357                server_socket: true,
358            },
359        ];
360        for message in messages {
361            let frame = message.encode_frame().expect("encode");
362            assert_eq!(Message::decode_frame(&frame).expect("decode"), message);
363        }
364    }
365
366    #[test]
367    fn session_invitation_padding_round_trips_odd_lengths() {
368        let message = Message::SessionInvitation {
369            from: vec![9_u8; 32],
370            key: vec![1_u8; 32],
371            address: vec![10, 0, 0],
372            port: 1,
373            server_socket: false,
374        };
375        let frame = message.encode_frame().expect("encode");
376        assert_eq!(Message::decode_frame(&frame).expect("decode"), message);
377    }
378
379    #[test]
380    fn bad_magic_is_rejected() {
381        let mut frame = Message::Ping.encode_frame().expect("encode");
382        if let Some(first) = frame.first_mut() {
383            *first = 0;
384        }
385        assert!(Message::decode_frame(&frame).is_err());
386    }
387
388    #[test]
389    fn truncated_frame_is_rejected() {
390        let frame = Message::Ping.encode_frame().expect("encode");
391        let truncated = frame.get(..10).unwrap_or(&frame);
392        assert!(Message::decode_frame(truncated).is_err());
393    }
394
395    #[test]
396    fn trailing_bytes_are_rejected() {
397        let mut frame = Message::Ping.encode_frame().expect("encode");
398        frame.push(0);
399        assert!(Message::decode_frame(&frame).is_err());
400    }
401
402    #[tokio::test]
403    async fn tokio_framing_round_trips() {
404        let mut buffer = Vec::new();
405        for message in [
406            Message::JoinRelayRequest { token: "x".to_owned() },
407            Message::ConnectRequest { id: vec![5_u8; 32] },
408        ] {
409            write_message(&mut buffer, &message).await.expect("write");
410            assert_eq!(read_message(&mut &buffer[..]).await.expect("read"), message);
411            buffer.clear();
412        }
413    }
414}
syncweb_core/net/relay/protocol.rs

syncweb_core/net/relay/
protocol.rs
