lodestone_client/
cluster_control.rs

1//! Dedicated current-only cluster-control client.
2//!
3//! The client has no Collection profile, API-key provider, public session, or
4//! connection-pool reuse. Each call opens one TLS connection with the isolated
5//! ALPN and sends one bounded canonical cluster-control frame.
6
7use std::error::Error;
8use std::fmt::{self, Display, Formatter};
9use std::sync::Arc;
10use std::time::Duration;
11
12use lodestone_proto::{
13    CLUSTER_CONTROL_REQUEST_FORMAT_VERSION, CLUSTER_CONTROL_SESSION_FORMAT_VERSION, ClusterControlActionV1,
14    ClusterControlCredentialKindV1, ClusterControlRequestV1, ClusterControlResponseDispositionV1,
15    ClusterControlResponseOutcomeV1, ClusterControlSessionV1, MAX_CLUSTER_CONTROL_FRAME_BYTES,
16    decode_cluster_control_response_v1, encode_cluster_control_request_v1,
17};
18use rustls::pki_types::{CertificateDer, ServerName};
19use serde::Serialize;
20use tokio::io::{AsyncReadExt, AsyncWriteExt};
21use tokio::net::TcpStream;
22use tokio_rustls::TlsConnector;
23
24pub const CLUSTER_CONTROL_ALPN: &[u8] = b"lodestone-cluster-control-v1";
25const MAX_CLUSTER_CONTROL_CALL_MILLIS: u64 = 120_000;
26const MAX_CLUSTER_CONTROL_ENDPOINT_BYTES: usize = 512;
27
28pub struct ClusterControlCredential {
29    kind: ClusterControlCredentialKindV1,
30    bytes: Vec<u8>,
31}
32
33impl ClusterControlCredential {
34    pub fn new(kind: ClusterControlCredentialKindV1, bytes: Vec<u8>) -> Result<Self, ClusterControlClientError> {
35        if lodestone_proto::validate_cluster_control_credential_v1(&bytes).is_err() {
36            return Err(ClusterControlClientError::InvalidConfiguration);
37        }
38        Ok(Self { kind, bytes })
39    }
40
41    #[must_use]
42    pub const fn kind(&self) -> ClusterControlCredentialKindV1 {
43        self.kind
44    }
45}
46
47impl fmt::Debug for ClusterControlCredential {
48    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
49        formatter
50            .debug_struct("ClusterControlCredential")
51            .field("kind", &self.kind)
52            .field("bytes", &"[REDACTED]")
53            .finish()
54    }
55}
56
57impl Drop for ClusterControlCredential {
58    fn drop(&mut self) {
59        self.bytes.fill(0);
60        std::hint::black_box(&mut self.bytes);
61    }
62}
63
64pub struct ClusterControlClientConfig {
65    endpoint: String,
66    server_name: String,
67    trust_certificates_der: Vec<Vec<u8>>,
68    credential: ClusterControlCredential,
69    timeout: Duration,
70}
71
72impl ClusterControlClientConfig {
73    pub fn new(
74        endpoint: String,
75        server_name: String,
76        trust_certificates_der: Vec<Vec<u8>>,
77        credential: ClusterControlCredential,
78        timeout: Duration,
79    ) -> Result<Self, ClusterControlClientError> {
80        if endpoint.is_empty()
81            || endpoint.len() > MAX_CLUSTER_CONTROL_ENDPOINT_BYTES
82            || endpoint.parse::<std::net::SocketAddr>().is_err()
83            || !matches!(ServerName::try_from(server_name.clone()), Ok(ServerName::DnsName(_)))
84            || trust_certificates_der.is_empty()
85            || timeout.is_zero()
86            || timeout > Duration::from_millis(MAX_CLUSTER_CONTROL_CALL_MILLIS)
87        {
88            return Err(ClusterControlClientError::InvalidConfiguration);
89        }
90        Ok(Self {
91            endpoint,
92            server_name,
93            trust_certificates_der,
94            credential,
95            timeout,
96        })
97    }
98}
99
100pub struct ClusterControlClient {
101    config: ClusterControlClientConfig,
102    tls: Arc<rustls::ClientConfig>,
103}
104
105impl ClusterControlClient {
106    pub fn new(config: ClusterControlClientConfig) -> Result<Self, ClusterControlClientError> {
107        let tls = build_tls(&config)?;
108        Ok(Self {
109            config,
110            tls: Arc::new(tls),
111        })
112    }
113
114    pub async fn initialize_collection(
115        &self,
116        request_id: &str,
117        body: impl Serialize,
118    ) -> Result<serde_json::Value, ClusterControlClientError> {
119        self.call(ClusterControlActionV1::InitializeCollectionAuthority, request_id, body)
120            .await
121    }
122
123    pub async fn activate_collection(
124        &self,
125        request_id: &str,
126        body: impl Serialize,
127    ) -> Result<serde_json::Value, ClusterControlClientError> {
128        self.call(ClusterControlActionV1::ActivateCollectionAuthority, request_id, body)
129            .await
130    }
131
132    pub async fn resolve_collection(
133        &self,
134        request_id: &str,
135        body: impl Serialize,
136    ) -> Result<serde_json::Value, ClusterControlClientError> {
137        self.call(ClusterControlActionV1::ResolveCollectionAuthority, request_id, body)
138            .await
139    }
140
141    pub async fn inspect_collection_retirement(
142        &self,
143        request_id: &str,
144        body: impl Serialize,
145    ) -> Result<serde_json::Value, ClusterControlClientError> {
146        self.call(ClusterControlActionV1::InspectCollectionRetirement, request_id, body)
147            .await
148    }
149
150    pub async fn retire_collection(
151        &self,
152        request_id: &str,
153        body: impl Serialize,
154    ) -> Result<serde_json::Value, ClusterControlClientError> {
155        self.call(ClusterControlActionV1::RetireCollectionAuthority, request_id, body)
156            .await
157    }
158
159    pub async fn observe_capacity(&self, request_id: &str) -> Result<serde_json::Value, ClusterControlClientError> {
160        self.call(
161            ClusterControlActionV1::ObserveClusterCapacity,
162            request_id,
163            serde_json::json!({}),
164        )
165        .await
166    }
167
168    pub async fn observe_resource_diagnostics(
169        &self,
170        request_id: &str,
171        body: impl Serialize,
172    ) -> Result<serde_json::Value, ClusterControlClientError> {
173        self.call(ClusterControlActionV1::ObserveResourceDiagnostics, request_id, body)
174            .await
175    }
176
177    pub async fn reserve_credential_rotation(
178        &self,
179        request_id: &str,
180        body: impl Serialize,
181    ) -> Result<serde_json::Value, ClusterControlClientError> {
182        self.call(
183            ClusterControlActionV1::ReserveClusterControlCredentialRotation,
184            request_id,
185            body,
186        )
187        .await
188    }
189
190    pub async fn activate_credential_rotation(
191        &self,
192        request_id: &str,
193        body: impl Serialize,
194    ) -> Result<serde_json::Value, ClusterControlClientError> {
195        self.call(
196            ClusterControlActionV1::ActivateClusterControlCredentialRotation,
197            request_id,
198            body,
199        )
200        .await
201    }
202
203    pub async fn revoke_credential(
204        &self,
205        request_id: &str,
206        body: impl Serialize,
207    ) -> Result<serde_json::Value, ClusterControlClientError> {
208        self.call(ClusterControlActionV1::RevokeClusterControlCredential, request_id, body)
209            .await
210    }
211
212    pub async fn resolve_credential(
213        &self,
214        request_id: &str,
215        body: impl Serialize,
216    ) -> Result<serde_json::Value, ClusterControlClientError> {
217        self.call(
218            ClusterControlActionV1::ResolveClusterControlCredential,
219            request_id,
220            body,
221        )
222        .await
223    }
224
225    pub async fn list_backups(&self, request_id: &str) -> Result<serde_json::Value, ClusterControlClientError> {
226        self.call(
227            ClusterControlActionV1::ListClusterBackups,
228            request_id,
229            serde_json::json!({}),
230        )
231        .await
232    }
233
234    pub async fn get_backup(
235        &self,
236        request_id: &str,
237        body: impl Serialize,
238    ) -> Result<serde_json::Value, ClusterControlClientError> {
239        self.call(ClusterControlActionV1::GetClusterBackup, request_id, body)
240            .await
241    }
242
243    pub async fn create_backup(
244        &self,
245        request_id: &str,
246        body: impl Serialize,
247    ) -> Result<serde_json::Value, ClusterControlClientError> {
248        self.call(ClusterControlActionV1::CreateClusterBackup, request_id, body)
249            .await
250    }
251
252    pub async fn resolve_backup(
253        &self,
254        request_id: &str,
255        body: impl Serialize,
256    ) -> Result<serde_json::Value, ClusterControlClientError> {
257        self.call(ClusterControlActionV1::ResolveClusterBackup, request_id, body)
258            .await
259    }
260
261    pub async fn restore_readiness(&self, request_id: &str) -> Result<serde_json::Value, ClusterControlClientError> {
262        self.call(
263            ClusterControlActionV1::GetClusterRestoreReadiness,
264            request_id,
265            serde_json::json!({}),
266        )
267        .await
268    }
269
270    pub async fn prepare_offline_recovery(
271        &self,
272        request_id: &str,
273        body: impl Serialize,
274    ) -> Result<serde_json::Value, ClusterControlClientError> {
275        self.call(ClusterControlActionV1::PrepareOfflineRecovery, request_id, body)
276            .await
277    }
278
279    pub async fn call(
280        &self,
281        action: ClusterControlActionV1,
282        request_id: &str,
283        body: impl Serialize,
284    ) -> Result<serde_json::Value, ClusterControlClientError> {
285        if !self.config.credential.kind.admits(action) {
286            return Err(ClusterControlClientError::ActionDenied);
287        }
288        let body = serde_json::to_value(body).map_err(|_| ClusterControlClientError::Codec)?;
289        if !body.is_object() {
290            return Err(ClusterControlClientError::Codec);
291        }
292        let mut correlation = [0_u8; 16];
293        getrandom::fill(&mut correlation).map_err(|_| ClusterControlClientError::RandomUnavailable)?;
294        let mut request = ClusterControlRequestV1 {
295            format_version: CLUSTER_CONTROL_REQUEST_FORMAT_VERSION,
296            session: ClusterControlSessionV1 {
297                format_version: CLUSTER_CONTROL_SESSION_FORMAT_VERSION,
298                credential_kind: self.config.credential.kind,
299                credential: self.config.credential.bytes.clone(),
300                request_id: request_id.to_owned(),
301                action,
302                request_correlation_id: correlation,
303            },
304            body,
305        };
306        let encoded = encode_cluster_control_request_v1(&request);
307        request.session.credential.fill(0);
308        std::hint::black_box(&mut request.session.credential);
309        let mut payload = encoded.map_err(|_| ClusterControlClientError::Codec)?;
310        let result = tokio::time::timeout(self.config.timeout, self.call_once(&payload, request_id, correlation)).await;
311        payload.fill(0);
312        std::hint::black_box(&mut payload);
313        result.map_err(|_| ClusterControlClientError::DeadlineExceeded)?
314    }
315
316    async fn call_once(
317        &self,
318        payload: &[u8],
319        request_id: &str,
320        correlation: [u8; 16],
321    ) -> Result<serde_json::Value, ClusterControlClientError> {
322        let tcp = TcpStream::connect(&self.config.endpoint)
323            .await
324            .map_err(|_| ClusterControlClientError::Transport)?;
325        let server_name = ServerName::try_from(self.config.server_name.clone())
326            .map_err(|_| ClusterControlClientError::InvalidConfiguration)?;
327        let connector = TlsConnector::from(Arc::clone(&self.tls));
328        let mut tls = connector
329            .connect(server_name, tcp)
330            .await
331            .map_err(|_| ClusterControlClientError::Transport)?;
332        let length = u32::try_from(payload.len()).map_err(|_| ClusterControlClientError::Codec)?;
333        tls.write_u32(length)
334            .await
335            .map_err(|_| ClusterControlClientError::Transport)?;
336        tls.write_all(payload)
337            .await
338            .map_err(|_| ClusterControlClientError::Transport)?;
339        let length = usize::try_from(tls.read_u32().await.map_err(|_| ClusterControlClientError::Transport)?)
340            .map_err(|_| ClusterControlClientError::Codec)?;
341        if length == 0 || length > MAX_CLUSTER_CONTROL_FRAME_BYTES {
342            return Err(ClusterControlClientError::Codec);
343        }
344        let mut response = vec![0; length];
345        tls.read_exact(&mut response)
346            .await
347            .map_err(|_| ClusterControlClientError::Transport)?;
348        let response = decode_cluster_control_response_v1(&response).map_err(|_| ClusterControlClientError::Codec)?;
349        if response.request_id != request_id || response.request_correlation_id != correlation {
350            return Err(ClusterControlClientError::ResponseIdentityMismatch);
351        }
352        match response.outcome {
353            ClusterControlResponseOutcomeV1::Succeeded { body } => Ok(body),
354            ClusterControlResponseOutcomeV1::Refused { code, disposition } => {
355                Err(ClusterControlClientError::Refused(ClusterControlRefusal {
356                    code,
357                    disposition,
358                }))
359            }
360        }
361    }
362}
363
364fn build_tls(config: &ClusterControlClientConfig) -> Result<rustls::ClientConfig, ClusterControlClientError> {
365    let mut roots = rustls::RootCertStore::empty();
366    for certificate in &config.trust_certificates_der {
367        roots
368            .add(CertificateDer::from(certificate.clone()))
369            .map_err(|_| ClusterControlClientError::InvalidConfiguration)?;
370    }
371    let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider());
372    let mut tls = rustls::ClientConfig::builder_with_provider(provider)
373        .with_protocol_versions(&[&rustls::version::TLS13])
374        .map_err(|_| ClusterControlClientError::InvalidConfiguration)?
375        .with_root_certificates(roots)
376        .with_no_client_auth();
377    tls.alpn_protocols = vec![CLUSTER_CONTROL_ALPN.to_vec()];
378    Ok(tls)
379}
380
381#[derive(Clone, Debug, Eq, PartialEq)]
382pub struct ClusterControlRefusal {
383    pub code: String,
384    pub disposition: ClusterControlResponseDispositionV1,
385}
386
387#[derive(Debug)]
388pub enum ClusterControlClientError {
389    InvalidConfiguration,
390    ActionDenied,
391    RandomUnavailable,
392    Codec,
393    Transport,
394    DeadlineExceeded,
395    ResponseIdentityMismatch,
396    Refused(ClusterControlRefusal),
397}
398
399impl Display for ClusterControlClientError {
400    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
401        formatter.write_str(match self {
402            Self::InvalidConfiguration => "cluster-control client configuration is invalid",
403            Self::ActionDenied => "cluster-control credential kind cannot invoke this action",
404            Self::RandomUnavailable => "cluster-control request identity entropy is unavailable",
405            Self::Codec => "cluster-control frame is invalid",
406            Self::Transport => "cluster-control transport failed",
407            Self::DeadlineExceeded => "cluster-control call deadline elapsed",
408            Self::ResponseIdentityMismatch => "cluster-control response identity differs from the request",
409            Self::Refused(_) => "cluster-control request was refused",
410        })
411    }
412}
413
414impl Error for ClusterControlClientError {}
415
416#[cfg(test)]
417mod tests {
418    use super::*;
419
420    #[test]
421    fn debug_never_exposes_cluster_credential() -> Result<(), ClusterControlClientError> {
422        let credential = ClusterControlCredential::new(
423            ClusterControlCredentialKindV1::CapacityObserver,
424            b"lcc1_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".to_vec(),
425        )?;
426        let debug = format!("{credential:?}");
427        assert!(!debug.contains("lcc1_AAA"));
428        assert!(debug.contains("REDACTED"));
429        Ok(())
430    }
431}
lodestone_client/cluster_control.rs

lodestone_client/
cluster_control.rs
