libcfd_rpc/
quic.rs

1//! Plain-Rust types for the QUIC per-stream metadata protocol.
2//!
3//! The wire format is Cap'n Proto (see `quic_metadata_protocol.capnp`);
4//! encoding and decoding happen here so the `libcfd` crate never touches
5//! `capnp` directly.
6
7use crate::error::Result;
8use crate::io::{AsyncStream, read_message};
9use crate::quic_metadata_protocol_capnp as mpc;
10
11/// Magic bytes that identify a data (request) stream, from cloudflared's
12/// `tunnelrpc/quic/protocol.go`.
13pub const DATA_STREAM_PROTOCOL_SIGNATURE: [u8; 6] = [0x0A, 0x36, 0xCD, 0x12, 0xA1, 0x3E];
14
15/// Magic bytes that identify an edge-initiated RPC stream.
16pub const RPC_STREAM_PROTOCOL_SIGNATURE: [u8; 6] = [0x52, 0xBB, 0x82, 0x5C, 0xDB, 0x65];
17
18/// The per-stream protocol version ("01").
19pub const PROTOCOL_V1: &[u8] = b"01";
20
21/// Metadata key carrying the HTTP request method on a data stream.
22pub const HTTP_METHOD_KEY: &str = "HttpMethod";
23/// Metadata key carrying the HTTP request host on a data stream.
24pub const HTTP_HOST_KEY: &str = "HttpHost";
25/// Prefix for per-header metadata entries (e.g. `HttpHeader:content-type`).
26pub const HTTP_HEADER_KEY: &str = "HttpHeader";
27/// Metadata key carrying the HTTP response status on a data stream.
28pub const HTTP_STATUS_KEY: &str = "HttpStatus";
29
30/// The kind of connection the edge requests on a data stream.
31#[derive(Debug, Clone, Copy, PartialEq, Eq)]
32pub enum ConnectionType {
33    /// A regular HTTP request.
34    Http,
35    /// A websocket upgrade.
36    Websocket,
37    /// A raw TCP stream.
38    Tcp,
39}
40
41impl ConnectionType {
42    fn from_capnp(t: mpc::ConnectionType) -> Result<Self> {
43        Ok(match t {
44            mpc::ConnectionType::Http => Self::Http,
45            mpc::ConnectionType::Websocket => Self::Websocket,
46            mpc::ConnectionType::Tcp => Self::Tcp,
47        })
48    }
49
50    fn to_capnp(self) -> mpc::ConnectionType {
51        match self {
52            Self::Http => mpc::ConnectionType::Http,
53            Self::Websocket => mpc::ConnectionType::Websocket,
54            Self::Tcp => mpc::ConnectionType::Tcp,
55        }
56    }
57}
58
59/// The connection request sent by the edge on a data stream.
60#[derive(Debug, Clone)]
61pub struct ConnectRequest {
62    /// The destination host and path (e.g. `http://example.com/path`).
63    pub destination: String,
64    /// Whether the stream carries HTTP, websocket, or raw TCP traffic.
65    pub connection_type: ConnectionType,
66    /// Method, host, and per-header metadata entries.
67    pub metadata: Vec<(String, String)>,
68}
69
70/// The connection response sent back to the edge.
71#[derive(Debug, Clone, Default)]
72pub struct ConnectResponse {
73    /// An error message; empty on success.
74    pub error: String,
75    /// Status and response-header metadata entries.
76    pub metadata: Vec<(String, String)>,
77}
78
79/// Reads a `ConnectRequest` message from a stream.
80pub async fn read_connect_request<S: AsyncStream + Unpin>(
81    stream: &mut S,
82) -> Result<ConnectRequest> {
83    let reader = read_message(stream).await?;
84    decode_connect_request_message(&reader)
85}
86
87/// Writes a `ConnectResponse` message to a stream.
88pub async fn write_connect_response<S: AsyncStream + Unpin>(
89    stream: &mut S,
90    response: &ConnectResponse,
91) -> Result<()> {
92    let message = encode_connect_response(response)?;
93    let bytes = crate::io::serialize_message(&message);
94    crate::io::write_raw(stream, &bytes).await
95}
96
97/// Encodes a `ConnectResponse` into a Cap'n Proto message.
98///
99/// Matches cloudflared's `ConnectResponse.ToPogs`: the error field stays a
100/// null pointer when empty (capnp-go leaves it unset), keeping the wire
101/// bytes identical to the reference implementation.
102pub fn encode_connect_response(
103    response: &ConnectResponse,
104) -> Result<capnp::message::Builder<capnp::message::HeapAllocator>> {
105    let mut message = capnp::message::Builder::new_default();
106    let mut root = message.init_root::<mpc::connect_response::Builder>();
107    if !response.error.is_empty() {
108        root.set_error(&response.error);
109    }
110    let mut md = root
111        .reborrow()
112        .init_metadata(response.metadata.len() as u32);
113    for (i, (key, val)) in response.metadata.iter().enumerate() {
114        let mut entry = md.reborrow().get(i as u32);
115        entry.set_key(key);
116        entry.set_val(val);
117    }
118    Ok(message)
119}
120
121/// Encodes a `ConnectRequest` into a Cap'n Proto message (used by tests and
122/// the mock edge).
123pub fn encode_connect_request(
124    request: &ConnectRequest,
125) -> Result<capnp::message::Builder<capnp::message::HeapAllocator>> {
126    let mut message = capnp::message::Builder::new_default();
127    let mut root = message.init_root::<mpc::connect_request::Builder>();
128    root.set_dest(&request.destination);
129    root.set_type(request.connection_type.to_capnp());
130    let mut md = root.reborrow().init_metadata(request.metadata.len() as u32);
131    for (i, (key, val)) in request.metadata.iter().enumerate() {
132        let mut entry = md.reborrow().get(i as u32);
133        entry.set_key(key);
134        entry.set_val(val);
135    }
136    Ok(message)
137}
138
139/// Writes a `ConnectRequest` message to a stream (used by the mock edge).
140pub async fn write_connect_request<S: AsyncStream + Unpin>(
141    stream: &mut S,
142    request: &ConnectRequest,
143) -> Result<()> {
144    let message = encode_connect_request(request)?;
145    let bytes = crate::io::serialize_message(&message);
146    crate::io::write_raw(stream, &bytes).await
147}
148
149fn decode_connect_request_message<R: capnp::message::ReaderSegments>(
150    reader: &capnp::message::Reader<R>,
151) -> Result<ConnectRequest> {
152    let root = reader.get_root::<mpc::connect_request::Reader>()?;
153    let destination = root.get_dest()?.to_str()?.to_string();
154    let connection_type = ConnectionType::from_capnp(root.get_type()?)?;
155    let mut metadata = Vec::new();
156    for entry in root.get_metadata()? {
157        metadata.push((
158            entry.get_key()?.to_str()?.to_string(),
159            entry.get_val()?.to_str()?.to_string(),
160        ));
161    }
162    Ok(ConnectRequest {
163        destination,
164        connection_type,
165        metadata,
166    })
167}
168
169/// Decodes a `ConnectRequest` from serialized message bytes.
170pub fn decode_connect_request_bytes(bytes: &[u8]) -> Result<ConnectRequest> {
171    let reader = capnp::serialize::read_message_from_flat_slice(
172        &mut &bytes[..],
173        capnp::message::ReaderOptions::new(),
174    )?;
175    decode_connect_request_message(&reader)
176}
177
178/// Decodes a `ConnectResponse` from serialized message bytes.
179pub fn decode_connect_response_bytes(bytes: &[u8]) -> Result<ConnectResponse> {
180    let reader = capnp::serialize::read_message_from_flat_slice(
181        &mut &bytes[..],
182        capnp::message::ReaderOptions::new(),
183    )?;
184    let root = reader.get_root::<mpc::connect_response::Reader>()?;
185    let error = root.get_error()?.to_str()?.to_string();
186    let mut metadata = Vec::new();
187    for entry in root.get_metadata()? {
188        metadata.push((
189            entry.get_key()?.to_str()?.to_string(),
190            entry.get_val()?.to_str()?.to_string(),
191        ));
192    }
193    Ok(ConnectResponse { error, metadata })
194}
195
196/// Reads a `ConnectResponse` message from a stream (used by the mock edge).
197pub async fn read_connect_response<S: AsyncStream + Unpin>(
198    stream: &mut S,
199) -> Result<ConnectResponse> {
200    let reader = read_message(stream).await?;
201    let root = reader.get_root::<mpc::connect_response::Reader>()?;
202    let error = root.get_error()?.to_str()?.to_string();
203    let mut metadata = Vec::new();
204    for entry in root.get_metadata()? {
205        metadata.push((
206            entry.get_key()?.to_str()?.to_string(),
207            entry.get_val()?.to_str()?.to_string(),
208        ));
209    }
210    Ok(ConnectResponse { error, metadata })
211}
212
213#[cfg(test)]
214mod tests {
215    use super::*;
216
217    #[test]
218    fn connect_request_round_trip() {
219        let request = ConnectRequest {
220            destination: "http://example.com/path".into(),
221            connection_type: ConnectionType::Http,
222            metadata: vec![
223                ("HttpMethod".into(), "GET".into()),
224                ("HttpHost".into(), "example.com".into()),
225            ],
226        };
227        let message = encode_connect_request(&request).unwrap();
228        let bytes = capnp::serialize::write_message_to_words(&message);
229        let decoded = decode_connect_request_bytes(&bytes).unwrap();
230        assert_eq!(decoded.destination, request.destination);
231        assert_eq!(decoded.connection_type, request.connection_type);
232        assert_eq!(decoded.metadata, request.metadata);
233    }
234
235    #[test]
236    fn connect_response_round_trip() {
237        let response = ConnectResponse {
238            error: String::new(),
239            metadata: vec![
240                ("HttpStatus".into(), "200".into()),
241                ("HttpHeader:content-type".into(), "text/plain".into()),
242            ],
243        };
244        let message = encode_connect_response(&response).unwrap();
245        let bytes = capnp::serialize::write_message_to_words(&message);
246        let decoded = decode_connect_response_bytes(&bytes).unwrap();
247        assert_eq!(decoded.error, "");
248        assert_eq!(decoded.metadata, response.metadata);
249    }
250}
libcfd_rpc/quic.rs

libcfd_rpc/
quic.rs
