libcfd_rpc/
tunnel.rs

1use crate::error::Result;
2use crate::rpc::RpcClient;
3use crate::tunnelrpc_capnp;
4
5/// The 64-bit interface id of the edge's `RegistrationServer`.
6pub const REGISTRATION_SERVER_INTERFACE_ID: u64 = 0xf716_95ec_7fe8_5497;
7/// Method id of `registerConnection`.
8pub const METHOD_REGISTER_CONNECTION: u16 = 0;
9/// Method id of `unregisterConnection`.
10pub const METHOD_UNREGISTER_CONNECTION: u16 = 1;
11/// Method id of `updateLocalConfiguration`.
12pub const METHOD_UPDATE_LOCAL_CONFIGURATION: u16 = 2;
13
14/// The client's connector identity, sent as `ConnectionOptions.client`.
15#[derive(Debug, Clone, Default)]
16pub struct ClientInformation {
17    /// 16-byte connector UUID.
18    pub client_identifier: Vec<u8>,
19    /// Feature flags the connector advertises (e.g. `serialized_headers`).
20    pub features: Vec<String>,
21    /// The client version string.
22    pub version: String,
23    /// The client OS/architecture string.
24    pub arch: String,
25}
26
27/// Parameters sent with `registerConnection`.
28#[derive(Debug, Clone, Default)]
29pub struct ConnectionOptions {
30    /// The connector identity advertised to the edge.
31    pub client: ClientInformation,
32    /// Raw IP bytes of the local edge-facing address.
33    pub origin_local_ip: Vec<u8>,
34    /// Whether to replace an existing connection for the same tunnel.
35    pub replace_existing: bool,
36    /// The compression quality to use (0 disables it).
37    pub compression_quality: u8,
38    /// How many previous connection attempts this process made.
39    pub number_previous_attempts: u8,
40}
41
42/// Credentials proving ownership of the tunnel.
43#[derive(Debug, Clone, Default)]
44pub struct TunnelAuth {
45    /// The account tag that owns the tunnel.
46    pub account_tag: String,
47    /// The tunnel secret (opaque bytes; never logged).
48    pub tunnel_secret: Vec<u8>,
49}
50
51/// A rejected registration.
52#[derive(Debug, Clone)]
53pub struct ConnectionError {
54    /// The edge's error cause string.
55    pub cause: String,
56    /// Nanoseconds to wait before retrying.
57    pub retry_after: i64,
58    /// Whether the edge considers the failure retryable.
59    pub should_retry: bool,
60}
61
62/// A successful registration.
63#[derive(Debug, Clone)]
64pub struct ConnectionDetails {
65    /// Per-connection UUID (16 bytes).
66    pub uuid: Vec<u8>,
67    /// Airport code of the edge colo.
68    pub location_name: String,
69    /// Whether the tunnel is configured remotely by the edge.
70    pub tunnel_is_remotely_managed: bool,
71}
72
73/// The `ConnectionResponse` union.
74#[derive(Debug, Clone)]
75pub enum ConnectionResponse {
76    /// The edge rejected the registration.
77    Error(ConnectionError),
78    /// The registration succeeded.
79    Details(ConnectionDetails),
80}
81
82/// A typed client for the tunnel registration interface.
83///
84/// Wraps an [`RpcClient`] and exposes only plain Rust types so the caller
85/// never touches Cap'n Proto directly.
86pub struct TunnelClient<S> {
87    rpc: RpcClient<S>,
88}
89
90impl<S: crate::io::AsyncStream + Unpin> TunnelClient<S> {
91    /// Wraps an [`RpcClient`] as a typed registration client.
92    pub fn new(rpc: RpcClient<S>) -> Self {
93        Self { rpc }
94    }
95
96    /// Bootstraps the edge's registration interface.
97    pub async fn bootstrap(&mut self) -> Result<()> {
98        self.rpc.bootstrap().await.map(|_| ())
99    }
100
101    /// Calls `registerConnection` with the tunnel credentials and options.
102    pub async fn register_connection(
103        &mut self,
104        auth: TunnelAuth,
105        tunnel_identifier: &[u8],
106        connection_index: u8,
107        options: &ConnectionOptions,
108    ) -> Result<ConnectionResponse> {
109        let auth_ = auth;
110        let tunnel_identifier = tunnel_identifier.to_vec();
111        let options = options.clone();
112        self.rpc
113            .call(
114                0,
115                REGISTRATION_SERVER_INTERFACE_ID,
116                METHOD_REGISTER_CONNECTION,
117                |payload| {
118                    let mut parameters = payload
119                        .reborrow()
120                        .init_content()
121                        .init_as::<tunnelrpc_capnp::registration_server::register_connection_params::Builder>();
122                    {
123                        let mut a = parameters.reborrow().init_auth();
124                        a.set_account_tag(&auth_.account_tag);
125                        a.set_tunnel_secret(&auth_.tunnel_secret);
126                    }
127                    parameters.set_tunnel_id(&tunnel_identifier);
128                    parameters.set_conn_index(connection_index);
129                    {
130                        let mut o = parameters.reborrow().init_options();
131                        let mut c = o.reborrow().init_client();
132                        c.set_client_id(&options.client.client_identifier);
133                        let mut features = c
134                            .reborrow()
135                            .init_features(options.client.features.len() as u32);
136                        for (i, f) in options.client.features.iter().enumerate() {
137                            features.set(i as u32, f);
138                        }
139                        c.set_version(&options.client.version);
140                        c.set_arch(&options.client.arch);
141                        o.set_origin_local_ip(&options.origin_local_ip);
142                        o.set_replace_existing(options.replace_existing);
143                        o.set_compression_quality(options.compression_quality);
144                        o.set_num_previous_attempts(options.number_previous_attempts);
145                    }
146                    payload.reborrow().init_cap_table(0);
147                    Ok(())
148                },
149                |results| {
150                    let results_reader = results
151                        .reborrow()
152                        .get_content()
153                        .get_as::<tunnelrpc_capnp::registration_server::register_connection_results::Reader<'_>>()?;
154                    let connection_response = results_reader.reborrow().get_result()?;
155                    match connection_response.reborrow().get_result().which()? {                        tunnelrpc_capnp::connection_response::result::Error(e) => {
156                            let e = e?;
157                            Ok(ConnectionResponse::Error(ConnectionError {
158                                cause: e.get_cause()?.to_str()?.to_string(),
159                                retry_after: e.get_retry_after(),
160                                should_retry: e.get_should_retry(),
161                            }))
162                        }
163                        tunnelrpc_capnp::connection_response::result::ConnectionDetails(d) => {
164                            let d = d?;
165                            Ok(ConnectionResponse::Details(ConnectionDetails {
166                                uuid: d.get_uuid()?.to_vec(),
167                                location_name: d.get_location_name()?.to_str()?.to_string(),
168                                tunnel_is_remotely_managed: d
169                                    .get_tunnel_is_remotely_managed(),
170                            }))
171                        }
172                    }
173                },
174            )
175            .await
176    }
177
178    /// Calls `unregisterConnection` for the current connection.
179    pub async fn unregister_connection(&mut self) -> Result<()> {
180        self.rpc
181            .call(
182                0,
183                REGISTRATION_SERVER_INTERFACE_ID,
184                METHOD_UNREGISTER_CONNECTION,
185                |payload| {
186                    payload
187                        .reborrow()
188                        .init_content()
189                        .init_as::<tunnelrpc_capnp::registration_server::unregister_connection_params::Builder>();
190                    payload.reborrow().init_cap_table(0);
191                    Ok(())
192                },
193                |_results| Ok(()),
194            )
195            .await
196    }
197
198    /// Pushes the local configuration to the edge via
199    /// `updateLocalConfiguration` (for locally-managed tunnels).
200    pub async fn update_local_configuration(&mut self, configuration: &[u8]) -> Result<()> {
201        let configuration = configuration.to_vec();
202        self.rpc
203            .call(
204                0,
205                REGISTRATION_SERVER_INTERFACE_ID,
206                METHOD_UPDATE_LOCAL_CONFIGURATION,
207                |payload| {
208                    let mut parameters = payload
209                        .reborrow()
210                        .init_content()
211                        .init_as::<tunnelrpc_capnp::registration_server::update_local_configuration_params::Builder>();
212                    parameters.set_config(&configuration);
213                    payload.reborrow().init_cap_table(0);
214                    Ok(())
215                },
216                |_results| Ok(()),
217            )
218            .await
219    }
220
221    /// Returns the underlying [`RpcClient`] without releasing the
222    /// registration capability.
223    pub fn into_inner(self) -> RpcClient<S> {
224        self.rpc
225    }
226
227    /// Releases the registration capability and returns the underlying
228    /// stream, mirroring capnp-go's client `Close()`.
229    pub async fn close(self) -> Result<S> {
230        self.rpc.close().await
231    }
232}
233
234/// Convenience: `ConnectionResponse` with `should_retry` mapped to a typed
235/// result so callers can distinguish retryable failures without parsing
236/// strings.
237impl ConnectionResponse {
238    /// Converts the response into a typed registration result.
239    pub fn into_result(self) -> std::result::Result<ConnectionDetails, RegistrationFailure> {
240        match self {
241            Self::Details(d) => Ok(d),
242            Self::Error(e) => Err(RegistrationFailure::from(e)),
243        }
244    }
245}
246
247/// A typed registration failure, split by retryability.
248#[derive(Debug, Clone)]
249pub enum RegistrationFailure {
250    /// The edge asked us to retry after a delay.
251    Retryable {
252        /// The edge's cause string.
253        cause: String,
254        /// Nanoseconds to wait before retrying.
255        retry_after: i64,
256    },
257    /// The edge will keep rejecting this tunnel; retrying is pointless.
258    Permanent(String),
259}
260
261impl From<ConnectionError> for RegistrationFailure {
262    fn from(e: ConnectionError) -> Self {
263        if e.should_retry {
264            Self::Retryable {
265                cause: e.cause,
266                retry_after: e.retry_after,
267            }
268        } else {
269            Self::Permanent(e.cause)
270        }
271    }
272}
273
274impl std::fmt::Display for RegistrationFailure {
275    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
276        match self {
277            Self::Retryable { cause, retry_after } => {
278                write!(
279                    f,
280                    "retryable registration failure ({cause}, retry after {retry_after}ns)"
281                )
282            }
283            Self::Permanent(cause) => write!(f, "permanent registration failure: {cause}"),
284        }
285    }
286}
287
288impl std::error::Error for RegistrationFailure {}
libcfd_rpc/tunnel.rs

libcfd_rpc/
tunnel.rs
