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, ClusterControlActionV2,
14    ClusterControlCredentialKindV1, ClusterControlRequestV2, ClusterControlResponseDispositionV2,
15    ClusterControlResponseOutcomeV2, ClusterControlSessionV2, MAX_CLUSTER_CONTROL_FRAME_BYTES,
16    decode_cluster_control_response_v2, encode_cluster_control_request_v2,
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-v2";
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(ClusterControlActionV2::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(ClusterControlActionV2::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(ClusterControlActionV2::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(ClusterControlActionV2::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(ClusterControlActionV2::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            ClusterControlActionV2::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(ClusterControlActionV2::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            ClusterControlActionV2::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            ClusterControlActionV2::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(ClusterControlActionV2::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            ClusterControlActionV2::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            ClusterControlActionV2::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(ClusterControlActionV2::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(ClusterControlActionV2::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(ClusterControlActionV2::ResolveClusterBackup, request_id, body)
258            .await
259    }
260
261    pub async fn delete_backup(
262        &self,
263        request_id: &str,
264        body: impl Serialize,
265    ) -> Result<serde_json::Value, ClusterControlClientError> {
266        self.call(ClusterControlActionV2::DeleteClusterBackup, request_id, body)
267            .await
268    }
269
270    pub async fn resolve_backup_deletion(
271        &self,
272        request_id: &str,
273        body: impl Serialize,
274    ) -> Result<serde_json::Value, ClusterControlClientError> {
275        self.call(ClusterControlActionV2::ResolveClusterBackupDeletion, request_id, body)
276            .await
277    }
278
279    pub async fn restore_readiness(&self, request_id: &str) -> Result<serde_json::Value, ClusterControlClientError> {
280        self.call(
281            ClusterControlActionV2::GetClusterRestoreReadiness,
282            request_id,
283            serde_json::json!({}),
284        )
285        .await
286    }
287
288    pub async fn prepare_offline_recovery(
289        &self,
290        request_id: &str,
291        body: impl Serialize,
292    ) -> Result<serde_json::Value, ClusterControlClientError> {
293        self.call(ClusterControlActionV2::PrepareOfflineRecovery, request_id, body)
294            .await
295    }
296
297    pub async fn call(
298        &self,
299        action: ClusterControlActionV2,
300        request_id: &str,
301        body: impl Serialize,
302    ) -> Result<serde_json::Value, ClusterControlClientError> {
303        if !self.config.credential.kind.admits(action) {
304            return Err(ClusterControlClientError::ActionDenied);
305        }
306        let body = serde_json::to_value(body).map_err(|_| ClusterControlClientError::Codec)?;
307        if !body.is_object() {
308            return Err(ClusterControlClientError::Codec);
309        }
310        let mut correlation = [0_u8; 16];
311        getrandom::fill(&mut correlation).map_err(|_| ClusterControlClientError::RandomUnavailable)?;
312        let mut request = ClusterControlRequestV2 {
313            format_version: CLUSTER_CONTROL_REQUEST_FORMAT_VERSION,
314            session: ClusterControlSessionV2 {
315                format_version: CLUSTER_CONTROL_SESSION_FORMAT_VERSION,
316                credential_kind: self.config.credential.kind,
317                credential: self.config.credential.bytes.clone(),
318                request_id: request_id.to_owned(),
319                action,
320                request_correlation_id: correlation,
321            },
322            body,
323        };
324        let encoded = encode_cluster_control_request_v2(&request);
325        request.session.credential.fill(0);
326        std::hint::black_box(&mut request.session.credential);
327        let mut payload = encoded.map_err(|_| ClusterControlClientError::Codec)?;
328        let result = tokio::time::timeout(self.config.timeout, self.call_once(&payload, request_id, correlation)).await;
329        payload.fill(0);
330        std::hint::black_box(&mut payload);
331        result.map_err(|_| ClusterControlClientError::DeadlineExceeded)?
332    }
333
334    async fn call_once(
335        &self,
336        payload: &[u8],
337        request_id: &str,
338        correlation: [u8; 16],
339    ) -> Result<serde_json::Value, ClusterControlClientError> {
340        let tcp = TcpStream::connect(&self.config.endpoint)
341            .await
342            .map_err(|_| ClusterControlClientError::Transport)?;
343        let server_name = ServerName::try_from(self.config.server_name.clone())
344            .map_err(|_| ClusterControlClientError::InvalidConfiguration)?;
345        let connector = TlsConnector::from(Arc::clone(&self.tls));
346        let mut tls = connector
347            .connect(server_name, tcp)
348            .await
349            .map_err(|_| ClusterControlClientError::Transport)?;
350        let length = u32::try_from(payload.len()).map_err(|_| ClusterControlClientError::Codec)?;
351        tls.write_u32(length)
352            .await
353            .map_err(|_| ClusterControlClientError::Transport)?;
354        tls.write_all(payload)
355            .await
356            .map_err(|_| ClusterControlClientError::Transport)?;
357        let length = usize::try_from(tls.read_u32().await.map_err(|_| ClusterControlClientError::Transport)?)
358            .map_err(|_| ClusterControlClientError::Codec)?;
359        if length == 0 || length > MAX_CLUSTER_CONTROL_FRAME_BYTES {
360            return Err(ClusterControlClientError::Codec);
361        }
362        let mut response = vec![0; length];
363        tls.read_exact(&mut response)
364            .await
365            .map_err(|_| ClusterControlClientError::Transport)?;
366        let response = decode_cluster_control_response_v2(&response).map_err(|_| ClusterControlClientError::Codec)?;
367        if response.request_id != request_id || response.request_correlation_id != correlation {
368            return Err(ClusterControlClientError::ResponseIdentityMismatch);
369        }
370        match response.outcome {
371            ClusterControlResponseOutcomeV2::Succeeded { body } => Ok(body),
372            ClusterControlResponseOutcomeV2::Refused { code, disposition } => {
373                Err(ClusterControlClientError::Refused(ClusterControlRefusal {
374                    code,
375                    disposition,
376                }))
377            }
378        }
379    }
380}
381
382fn build_tls(config: &ClusterControlClientConfig) -> Result<rustls::ClientConfig, ClusterControlClientError> {
383    let mut roots = rustls::RootCertStore::empty();
384    for certificate in &config.trust_certificates_der {
385        roots
386            .add(CertificateDer::from(certificate.clone()))
387            .map_err(|_| ClusterControlClientError::InvalidConfiguration)?;
388    }
389    let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider());
390    let mut tls = rustls::ClientConfig::builder_with_provider(provider)
391        .with_protocol_versions(&[&rustls::version::TLS13])
392        .map_err(|_| ClusterControlClientError::InvalidConfiguration)?
393        .with_root_certificates(roots)
394        .with_no_client_auth();
395    tls.alpn_protocols = vec![CLUSTER_CONTROL_ALPN.to_vec()];
396    Ok(tls)
397}
398
399#[derive(Clone, Debug, Eq, PartialEq)]
400pub struct ClusterControlRefusal {
401    pub code: String,
402    pub disposition: ClusterControlResponseDispositionV2,
403}
404
405#[derive(Debug)]
406pub enum ClusterControlClientError {
407    InvalidConfiguration,
408    ActionDenied,
409    RandomUnavailable,
410    Codec,
411    Transport,
412    DeadlineExceeded,
413    ResponseIdentityMismatch,
414    Refused(ClusterControlRefusal),
415}
416
417impl Display for ClusterControlClientError {
418    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
419        formatter.write_str(match self {
420            Self::InvalidConfiguration => "cluster-control client configuration is invalid",
421            Self::ActionDenied => "cluster-control credential kind cannot invoke this action",
422            Self::RandomUnavailable => "cluster-control request identity entropy is unavailable",
423            Self::Codec => "cluster-control frame is invalid",
424            Self::Transport => "cluster-control transport failed",
425            Self::DeadlineExceeded => "cluster-control call deadline elapsed",
426            Self::ResponseIdentityMismatch => "cluster-control response identity differs from the request",
427            Self::Refused(_) => "cluster-control request was refused",
428        })
429    }
430}
431
432impl Error for ClusterControlClientError {}
433
434#[cfg(test)]
435mod tests {
436    use super::*;
437
438    #[test]
439    fn debug_never_exposes_cluster_credential() -> Result<(), ClusterControlClientError> {
440        let credential = ClusterControlCredential::new(
441            ClusterControlCredentialKindV1::CapacityObserver,
442            b"lcc1_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".to_vec(),
443        )?;
444        let debug = format!("{credential:?}");
445        assert!(!debug.contains("lcc1_AAA"));
446        assert!(debug.contains("REDACTED"));
447        Ok(())
448    }
449}
lodestone_client/cluster_control.rs

lodestone_client/
cluster_control.rs
