libcfd_rpc/
cloudflared.rs

1//! Server side of the connector's `CloudflaredServer` interface.
2//!
3//! The edge opens a dedicated RPC stream (prefixed with the RPC protocol
4//! signature) and calls the connector's main interface: `updateConfiguration`
5//! pushes the remotely-managed tunnel configuration (its ingress rules and
6//! therefore its public hostnames), while the UDP session methods manage
7//! QUIC datagram sessions, which libcfd does not support.
8
9use capnp::private::capability::{ClientHook, ParamsHook, ResultsHook};
10use capnp::traits::ImbueMut;
11
12use crate::error::{Result, RpcError};
13use crate::io::{AsyncStream, read_message, write_raw};
14use crate::rpc_capnp;
15use crate::tunnelrpc_capnp;
16
17/// The 64-bit interface id of the connector's `CloudflaredServer`.
18pub const CLOUDFLARED_SERVER_INTERFACE_ID: u64 = 0xf548_cef9_dea2_a4a1;
19/// The 64-bit interface id of `SessionManager` (extended by
20/// `CloudflaredServer`); capnp-go addresses its methods by this id.
21pub const SESSION_MANAGER_INTERFACE_ID: u64 = 0x8394_45a5_9fb0_1686;
22/// The 64-bit interface id of `ConfigurationManager`; the edge addresses
23/// `updateConfiguration` by this id.
24pub const CONFIGURATION_MANAGER_INTERFACE_ID: u64 = 0xb48e_dfbd_aa25_db04;
25
26/// The connector's reply to `updateConfiguration`.
27#[derive(Debug, Clone, Default)]
28pub struct UpdateConfigurationResponse {
29    /// The configuration version the connector applied.
30    pub latest_applied_version: i32,
31    /// An empty string on success.
32    pub error: String,
33}
34
35/// The connector's reply to `registerUdpSession`.
36#[derive(Debug, Clone, Default)]
37pub struct RegisterUdpSessionResponse {
38    /// An empty string on success.
39    pub error: String,
40    /// Session spans (unused by libcfd).
41    pub spans: Vec<u8>,
42}
43
44/// Handlers for the RPC calls the edge makes on the connector.
45pub trait CloudflaredHandler: Send + Sync {
46    /// Applies a remotely-managed configuration push from the edge.
47    fn update_configuration(
48        &self,
49        version: i32,
50        configuration: &[u8],
51    ) -> UpdateConfigurationResponse;
52    /// Registers a UDP session for QUIC datagrams. libcfd does not support
53    /// this; the default replies with an error.
54    fn register_udp_session(
55        &self,
56        _session_identifier: &[u8; 16],
57        _destination_ip: &[u8],
58        _destination_port: u16,
59    ) -> RegisterUdpSessionResponse {
60        RegisterUdpSessionResponse {
61            error: "UDP sessions are not supported".into(),
62            spans: Vec::new(),
63        }
64    }
65    /// Unregisters a UDP session. No-op by default.
66    fn unregister_udp_session(&self, _session_identifier: &[u8; 16], _message: &str) {}
67}
68
69/// Serves the edge's calls on an RPC stream until the stream ends.
70///
71/// Answers the bootstrap with a `senderHosted` capability (mirroring
72/// capnp-go's main-interface handshake), dispatches `updateConfiguration`
73/// to [`CloudflaredHandler::update_configuration`], and answers the UDP
74/// session methods from the handler. The edge sends `finish` and `release`
75/// messages between calls; those need no reply.
76pub async fn serve_cloudflared<S, H>(stream: &mut S, handler: &H) -> Result<()>
77where
78    S: AsyncStream + Unpin,
79    H: CloudflaredHandler,
80{
81    loop {
82        // Capnp reads and reply building stay in this scope so no non-Send capnp state crosses an await (mirroring the RPC client's rule).
83        let reply: Option<Vec<u8>> = {
84            let reader = match read_message(stream).await {
85                Ok(reader) => reader,
86                Err(RpcError::Eof) => return Ok(()),
87                Err(e) => return Err(e),
88            };
89            let root = reader.get_root::<rpc_capnp::message::Reader>()?;
90            match root.reborrow().which()? {
91                rpc_capnp::message::Bootstrap(b) => {
92                    let question = b?.get_question_id();
93                    tracing::debug!(question, "edge bootstrapped the RPC stream");
94                    Some(build_bootstrap_return(question)?)
95                }
96                rpc_capnp::message::Call(c) => {
97                    let call = c?;
98                    let question = call.get_question_id();
99                    tracing::debug!(
100                        question,
101                        interface_id = call.get_interface_id(),
102                        method_id = call.get_method_id(),
103                        "edge called an RPC method"
104                    );
105                    let reply = match classify(call.get_interface_id(), call.get_method_id()) {
106                        Method::UpdateConfiguration => {
107                            let parameters = call.reborrow().get_params()?.get_content().get_as::<
108                                tunnelrpc_capnp::configuration_manager::update_configuration_params::Reader<'_>,
109                            >()?;
110                            let response = handler.update_configuration(
111                                parameters.get_version(),
112                                parameters.get_config()?,
113                            );
114                            build_update_configuration_return(question, &response)?
115                        }
116                        Method::RegisterUdpSession => {
117                            let parameters = call.reborrow().get_params()?.get_content().get_as::<
118                                tunnelrpc_capnp::session_manager::register_udp_session_params::Reader<'_>,
119                            >()?;
120                            let response = handler.register_udp_session(
121                                &session_identifier_bytes(parameters.get_session_id()?),
122                                parameters.get_dst_ip()?,
123                                parameters.get_dst_port(),
124                            );
125                            build_register_udp_session_return(question, &response)?
126                        }
127                        Method::UnregisterUdpSession => {
128                            build_unregister_udp_session_return(question)?
129                        }
130                        Method::Unknown => {
131                            tracing::debug!(
132                                interface_id = call.get_interface_id(),
133                                method_id = call.get_method_id(),
134                                "edge called an unknown RPC method"
135                            );
136                            crate::rpc::build_exception(question, "unimplemented")?
137                        }
138                    };
139                    Some(reply)
140                }
141                rpc_capnp::message::Finish(_) | rpc_capnp::message::Release(_) => None,
142                _ => None,
143            }
144        };
145        if let Some(reply) = reply {
146            write_raw(stream, &reply).await?;
147        }
148    }
149}
150
151/// Which `CloudflaredServer` method a call addresses.
152///
153/// Method ordinals are per-interface: `updateConfiguration` is method 0 on
154/// `ConfigurationManager` but method 2 on the combined `CloudflaredServer`
155/// (after `SessionManager`'s two methods); the edge uses the sub-interface
156/// ids and ordinals.
157#[derive(Debug, Clone, Copy, PartialEq, Eq)]
158enum Method {
159    UpdateConfiguration,
160    RegisterUdpSession,
161    UnregisterUdpSession,
162    Unknown,
163}
164
165fn classify(interface_identifier: u64, method_identifier: u16) -> Method {
166    match (interface_identifier, method_identifier) {
167        (CONFIGURATION_MANAGER_INTERFACE_ID, 0) | (CLOUDFLARED_SERVER_INTERFACE_ID, 2) => {
168            Method::UpdateConfiguration
169        }
170        (SESSION_MANAGER_INTERFACE_ID, 0) | (CLOUDFLARED_SERVER_INTERFACE_ID, 0) => {
171            Method::RegisterUdpSession
172        }
173        (SESSION_MANAGER_INTERFACE_ID, 1) | (CLOUDFLARED_SERVER_INTERFACE_ID, 1) => {
174            Method::UnregisterUdpSession
175        }
176        _ => Method::Unknown,
177    }
178}
179
180/// Copies a data blob into a 16-byte session id, zero-padding short inputs.
181fn session_identifier_bytes(data: &[u8]) -> [u8; 16] {
182    let mut identifier = [0u8; 16];
183    let n = data.len().min(16);
184    identifier[..n].copy_from_slice(&data[..n]);
185    identifier
186}
187
188/// Builds the bootstrap answer: a `return` whose results carry the
189/// connector's main interface as a `senderHosted` capability, exactly as
190/// capnp-go answers a bootstrap question.
191fn build_bootstrap_return(question: u32) -> Result<Vec<u8>> {
192    // The capability table must outlive the message so the interface pointer below can reference it.
193    let mut caps: capnp::private::layout::CapTable = Vec::new();
194    let mut message = capnp::message::Builder::new_default();
195    let root = message.init_root::<rpc_capnp::message::Builder>();
196    let mut ret = root.init_return();
197    ret.set_answer_id(question);
198    let mut results = ret.init_results();
199    let mut payload = results.reborrow();
200    let mut content = payload.reborrow().init_content();
201    content.imbue_mut(&mut caps);
202    content.set_as_capability(Box::new(DummyHook));
203    let mut ctab = payload.reborrow().init_cap_table(1);
204    ctab.reborrow().get(0).set_sender_hosted(0);
205    Ok(crate::io::serialize_message(&message))
206}
207
208/// Builds the `return` for `updateConfiguration` with the typed results
209/// payload.
210fn build_update_configuration_return(
211    question: u32,
212    response: &UpdateConfigurationResponse,
213) -> Result<Vec<u8>> {
214    let mut message = capnp::message::Builder::new_default();
215    let root = message.init_root::<rpc_capnp::message::Builder>();
216    let mut ret = root.init_return();
217    ret.set_answer_id(question);
218    let mut results = ret.init_results();
219    let mut payload = results.reborrow();
220    {
221        let content = payload.reborrow().init_content();
222        let mut results_reader =
223            content.init_as::<tunnelrpc_capnp::configuration_manager::update_configuration_results::Builder>();
224        let mut response_builder = results_reader.reborrow().init_result();
225        response_builder.set_latest_applied_version(response.latest_applied_version);
226        response_builder.set_err(&response.error);
227    }
228    payload.reborrow().init_cap_table(0);
229    Ok(crate::io::serialize_message(&message))
230}
231
232/// Builds the `return` for `registerUdpSession`.
233fn build_register_udp_session_return(
234    question: u32,
235    response: &RegisterUdpSessionResponse,
236) -> Result<Vec<u8>> {
237    let mut message = capnp::message::Builder::new_default();
238    let root = message.init_root::<rpc_capnp::message::Builder>();
239    let mut ret = root.init_return();
240    ret.set_answer_id(question);
241    let mut results = ret.init_results();
242    let mut payload = results.reborrow();
243    {
244        let content = payload.reborrow().init_content();
245        let mut results_reader =
246            content
247                .init_as::<tunnelrpc_capnp::session_manager::register_udp_session_results::Builder>(
248                );
249        let mut response_builder = results_reader.reborrow().init_result();
250        response_builder.set_err(&response.error);
251        response_builder.set_spans(&response.spans);
252    }
253    payload.reborrow().init_cap_table(0);
254    Ok(crate::io::serialize_message(&message))
255}
256
257/// Builds the `return` for `unregisterUdpSession` (an empty results struct).
258fn build_unregister_udp_session_return(question: u32) -> Result<Vec<u8>> {
259    let mut message = capnp::message::Builder::new_default();
260    let root = message.init_root::<rpc_capnp::message::Builder>();
261    let mut ret = root.init_return();
262    ret.set_answer_id(question);
263    let mut results = ret.init_results();
264    let mut payload = results.reborrow();
265    payload
266        .reborrow()
267        .init_content()
268        .init_as::<tunnelrpc_capnp::session_manager::unregister_udp_session_results::Builder>();
269    payload.reborrow().init_cap_table(0);
270    Ok(crate::io::serialize_message(&message))
271}
272
273/// A placeholder capability hook for the bootstrap answer's cap table.
274///
275/// libcfd dispatches incoming calls by interface and method id directly, so
276/// this hook is never invoked; it only exists so the interface pointer can
277/// be written with capnp-rust's pointer API.
278struct DummyHook;
279
280impl ClientHook for DummyHook {
281    fn add_ref(&self) -> Box<dyn ClientHook> {
282        Box::new(DummyHook)
283    }
284    fn new_call(
285        &self,
286        _interface_identifier: u64,
287        _method_identifier: u16,
288        _size_hint: Option<capnp::MessageSize>,
289    ) -> capnp::capability::Request<capnp::any_pointer::Owned, capnp::any_pointer::Owned> {
290        unreachable!("dummy hook is never called")
291    }
292    fn call(
293        &self,
294        _interface_identifier: u64,
295        _method_identifier: u16,
296        _params: Box<dyn ParamsHook>,
297        _results: Box<dyn ResultsHook>,
298    ) -> capnp::capability::Promise<(), capnp::Error> {
299        unreachable!("dummy hook is never called")
300    }
301    fn get_brand(&self) -> usize {
302        0
303    }
304    fn get_ptr(&self) -> usize {
305        0
306    }
307    fn get_resolved(&self) -> Option<Box<dyn ClientHook>> {
308        None
309    }
310    fn when_more_resolved(
311        &self,
312    ) -> Option<capnp::capability::Promise<Box<dyn ClientHook>, capnp::Error>> {
313        None
314    }
315    fn when_resolved(&self) -> capnp::capability::Promise<(), capnp::Error> {
316        capnp::capability::Promise::ok(())
317    }
318}
libcfd_rpc/cloudflared.rs

libcfd_rpc/
cloudflared.rs
