lodestone_proto/
cluster_control_protocol.rs

1//! Strict cluster-control session wire values shared by servers and external clients.
2//!
3//! Cluster control is deliberately disjoint from [`crate::PublicSessionV1`].
4//! A cluster-control credential has no Collection grant, and the action enum
5//! makes each credential namespace's closed capability set explicit before a
6//! listener or durable credential authority is allowed to accept it.
7
8use std::error::Error;
9use std::fmt::{self, Display, Formatter};
10
11use serde::{Deserialize, Serialize};
12
13pub const CLUSTER_CONTROL_SESSION_FORMAT_VERSION: u16 = 2;
14pub const MAX_CLUSTER_CONTROL_SESSION_BYTES: usize = 1_024;
15pub const MAX_CLUSTER_CONTROL_CREDENTIAL_BYTES: usize = 512;
16pub const CLUSTER_CONTROL_CREDENTIAL_PREFIX: &[u8] = b"lcc1_";
17pub const CLUSTER_CONTROL_CREDENTIAL_ENCODED_BYTES: usize = 48;
18pub const CLUSTER_CONTROL_REQUEST_FORMAT_VERSION: u16 = 2;
19pub const CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION: u16 = 2;
20pub const MAX_CLUSTER_CONTROL_FRAME_BYTES: usize = 2 * 1024 * 1024;
21pub const MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES: usize = 128;
22pub const MAX_CLUSTER_CONTROL_ERROR_CODE_BYTES: usize = 96;
23
24#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
25#[serde(rename_all = "snake_case")]
26pub enum ClusterControlCredentialKindV1 {
27    CollectionProvisioner,
28    CapacityObserver,
29    RecoveryOperator,
30}
31
32#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
33#[serde(rename_all = "snake_case")]
34pub enum ClusterControlActionV2 {
35    InitializeCollectionAuthority,
36    ActivateCollectionAuthority,
37    ResolveCollectionAuthority,
38    InspectCollectionRetirement,
39    RetireCollectionAuthority,
40    ObserveClusterCapacity,
41    ObserveResourceDiagnostics,
42    ReserveClusterControlCredentialRotation,
43    ActivateClusterControlCredentialRotation,
44    RevokeClusterControlCredential,
45    ResolveClusterControlCredential,
46    ListClusterBackups,
47    GetClusterBackup,
48    CreateClusterBackup,
49    ResolveClusterBackup,
50    DeleteClusterBackup,
51    ResolveClusterBackupDeletion,
52    GetClusterRestoreReadiness,
53    PrepareOfflineRecovery,
54}
55
56impl ClusterControlCredentialKindV1 {
57    #[must_use]
58    pub const fn admits(self, action: ClusterControlActionV2) -> bool {
59        match self {
60            Self::CollectionProvisioner => matches!(
61                action,
62                ClusterControlActionV2::InitializeCollectionAuthority
63                    | ClusterControlActionV2::ActivateCollectionAuthority
64                    | ClusterControlActionV2::ResolveCollectionAuthority
65                    | ClusterControlActionV2::InspectCollectionRetirement
66                    | ClusterControlActionV2::RetireCollectionAuthority
67                    | ClusterControlActionV2::ReserveClusterControlCredentialRotation
68                    | ClusterControlActionV2::ActivateClusterControlCredentialRotation
69                    | ClusterControlActionV2::RevokeClusterControlCredential
70                    | ClusterControlActionV2::ResolveClusterControlCredential
71            ),
72            Self::CapacityObserver => matches!(
73                action,
74                ClusterControlActionV2::ObserveClusterCapacity | ClusterControlActionV2::ObserveResourceDiagnostics
75            ),
76            Self::RecoveryOperator => matches!(
77                action,
78                ClusterControlActionV2::ReserveClusterControlCredentialRotation
79                    | ClusterControlActionV2::ActivateClusterControlCredentialRotation
80                    | ClusterControlActionV2::RevokeClusterControlCredential
81                    | ClusterControlActionV2::ResolveClusterControlCredential
82                    | ClusterControlActionV2::ListClusterBackups
83                    | ClusterControlActionV2::GetClusterBackup
84                    | ClusterControlActionV2::CreateClusterBackup
85                    | ClusterControlActionV2::ResolveClusterBackup
86                    | ClusterControlActionV2::DeleteClusterBackup
87                    | ClusterControlActionV2::ResolveClusterBackupDeletion
88                    | ClusterControlActionV2::GetClusterRestoreReadiness
89                    | ClusterControlActionV2::PrepareOfflineRecovery
90            ),
91        }
92    }
93}
94
95#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
96#[serde(deny_unknown_fields)]
97pub struct ClusterControlSessionV2 {
98    pub format_version: u16,
99    pub credential_kind: ClusterControlCredentialKindV1,
100    pub credential: Vec<u8>,
101    pub request_id: String,
102    pub action: ClusterControlActionV2,
103    pub request_correlation_id: [u8; 16],
104}
105
106impl fmt::Debug for ClusterControlSessionV2 {
107    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
108        formatter
109            .debug_struct("ClusterControlSessionV2")
110            .field("format_version", &self.format_version)
111            .field("credential_kind", &self.credential_kind)
112            .field("credential", &"[REDACTED]")
113            .field("request_id", &self.request_id)
114            .field("action", &self.action)
115            .field("request_correlation_id", &self.request_correlation_id)
116            .finish()
117    }
118}
119
120impl ClusterControlSessionV2 {
121    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
122        if self.format_version != CLUSTER_CONTROL_SESSION_FORMAT_VERSION {
123            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
124        }
125        validate_cluster_control_credential_v1(&self.credential)?;
126        if self.request_id.is_empty()
127            || self.request_id.len() > MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES
128            || self.request_id.bytes().any(|byte| byte.is_ascii_control())
129        {
130            return Err(ClusterControlWireError::InvalidRequestIdentity);
131        }
132        if !self.credential_kind.admits(self.action) {
133            return Err(ClusterControlWireError::ActionDenied);
134        }
135        Ok(())
136    }
137}
138
139pub fn validate_cluster_control_credential_v1(credential: &[u8]) -> Result<(), ClusterControlWireError> {
140    if credential.len() != CLUSTER_CONTROL_CREDENTIAL_ENCODED_BYTES
141        || !credential.starts_with(CLUSTER_CONTROL_CREDENTIAL_PREFIX)
142        || !credential[CLUSTER_CONTROL_CREDENTIAL_PREFIX.len()..]
143            .iter()
144            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
145    {
146        return Err(ClusterControlWireError::InvalidCredentialShape);
147    }
148    Ok(())
149}
150
151#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
152#[serde(deny_unknown_fields)]
153pub struct ClusterControlRequestV2 {
154    pub format_version: u16,
155    pub session: ClusterControlSessionV2,
156    pub body: serde_json::Value,
157}
158
159impl ClusterControlRequestV2 {
160    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
161        if self.format_version != CLUSTER_CONTROL_REQUEST_FORMAT_VERSION {
162            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
163        }
164        self.session.validate()?;
165        if !self.body.is_object() {
166            return Err(ClusterControlWireError::InvalidBody);
167        }
168        Ok(())
169    }
170}
171
172#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
173#[serde(rename_all = "snake_case")]
174pub enum ClusterControlResponseDispositionV2 {
175    RetrySameRequest,
176    ResolveOperation,
177    CorrectRequest,
178    Reauthenticate,
179    Terminal,
180}
181
182#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
183#[serde(tag = "outcome", rename_all = "snake_case", deny_unknown_fields)]
184pub enum ClusterControlResponseOutcomeV2 {
185    Succeeded {
186        body: serde_json::Value,
187    },
188    Refused {
189        code: String,
190        disposition: ClusterControlResponseDispositionV2,
191    },
192}
193
194#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
195#[serde(deny_unknown_fields)]
196pub struct ClusterControlResponseV2 {
197    pub format_version: u16,
198    pub request_id: String,
199    pub request_correlation_id: [u8; 16],
200    pub outcome: ClusterControlResponseOutcomeV2,
201}
202
203impl ClusterControlResponseV2 {
204    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
205        if self.format_version != CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION {
206            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
207        }
208        validate_request_id(&self.request_id)?;
209        match &self.outcome {
210            ClusterControlResponseOutcomeV2::Succeeded { body } if body.is_object() => Ok(()),
211            ClusterControlResponseOutcomeV2::Succeeded { .. } => Err(ClusterControlWireError::InvalidBody),
212            ClusterControlResponseOutcomeV2::Refused { code, .. }
213                if !code.is_empty()
214                    && code.len() <= MAX_CLUSTER_CONTROL_ERROR_CODE_BYTES
215                    && code
216                        .bytes()
217                        .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_') =>
218            {
219                Ok(())
220            }
221            ClusterControlResponseOutcomeV2::Refused { .. } => Err(ClusterControlWireError::InvalidErrorCode),
222        }
223    }
224}
225
226pub fn encode_cluster_control_request_v2(
227    request: &ClusterControlRequestV2,
228) -> Result<Vec<u8>, ClusterControlWireError> {
229    request.validate()?;
230    encode_frame(request)
231}
232
233pub fn decode_cluster_control_request_v2(bytes: &[u8]) -> Result<ClusterControlRequestV2, ClusterControlWireError> {
234    let request: ClusterControlRequestV2 = decode_frame(bytes)?;
235    request.validate()?;
236    Ok(request)
237}
238
239pub fn encode_cluster_control_response_v2(
240    response: &ClusterControlResponseV2,
241) -> Result<Vec<u8>, ClusterControlWireError> {
242    response.validate()?;
243    encode_frame(response)
244}
245
246pub fn decode_cluster_control_response_v2(bytes: &[u8]) -> Result<ClusterControlResponseV2, ClusterControlWireError> {
247    let response: ClusterControlResponseV2 = decode_frame(bytes)?;
248    response.validate()?;
249    Ok(response)
250}
251
252fn encode_frame(value: &impl Serialize) -> Result<Vec<u8>, ClusterControlWireError> {
253    let bytes = serde_json::to_vec(value).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
254    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_FRAME_BYTES {
255        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
256    }
257    Ok(bytes)
258}
259
260fn decode_frame<T: for<'de> Deserialize<'de> + Serialize>(bytes: &[u8]) -> Result<T, ClusterControlWireError> {
261    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_FRAME_BYTES {
262        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
263    }
264    let value = serde_json::from_slice(bytes).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
265    if serde_json::to_vec(&value).map_err(|error| ClusterControlWireError::Codec(error.to_string()))? != bytes {
266        return Err(ClusterControlWireError::NonCanonical);
267    }
268    Ok(value)
269}
270
271fn validate_request_id(request_id: &str) -> Result<(), ClusterControlWireError> {
272    if request_id.is_empty()
273        || request_id.len() > MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES
274        || request_id.bytes().any(|byte| byte.is_ascii_control())
275    {
276        return Err(ClusterControlWireError::InvalidRequestIdentity);
277    }
278    Ok(())
279}
280
281pub fn encode_cluster_control_session_v2(
282    session: &ClusterControlSessionV2,
283) -> Result<Vec<u8>, ClusterControlWireError> {
284    session.validate()?;
285    let bytes = serde_json::to_vec(session).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
286    if bytes.len() > MAX_CLUSTER_CONTROL_SESSION_BYTES {
287        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
288    }
289    Ok(bytes)
290}
291
292pub fn decode_cluster_control_session_v2(bytes: &[u8]) -> Result<ClusterControlSessionV2, ClusterControlWireError> {
293    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_SESSION_BYTES {
294        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
295    }
296    let session: ClusterControlSessionV2 =
297        serde_json::from_slice(bytes).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
298    session.validate()?;
299    if serde_json::to_vec(&session).map_err(|error| ClusterControlWireError::Codec(error.to_string()))? != bytes {
300        return Err(ClusterControlWireError::NonCanonical);
301    }
302    Ok(session)
303}
304
305#[derive(Clone, Debug, Eq, PartialEq)]
306pub enum ClusterControlWireError {
307    UnsupportedVersion(u16),
308    InvalidCredentialShape,
309    InvalidRequestIdentity,
310    InvalidBody,
311    InvalidErrorCode,
312    ActionDenied,
313    FrameTooLarge(usize),
314    NonCanonical,
315    Codec(String),
316}
317
318impl Display for ClusterControlWireError {
319    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
320        match self {
321            Self::UnsupportedVersion(version) => {
322                write!(formatter, "unsupported cluster-control wire version {version}")
323            }
324            Self::InvalidCredentialShape => formatter.write_str("invalid cluster-control credential shape"),
325            Self::InvalidRequestIdentity => formatter.write_str("invalid cluster-control request identity"),
326            Self::InvalidBody => formatter.write_str("invalid cluster-control request or response body"),
327            Self::InvalidErrorCode => formatter.write_str("invalid cluster-control error code"),
328            Self::ActionDenied => formatter.write_str("cluster-control credential kind does not admit the action"),
329            Self::FrameTooLarge(size) => write!(formatter, "cluster-control frame size {size} exceeds the bound"),
330            Self::NonCanonical => formatter.write_str("cluster-control frame is not canonical"),
331            Self::Codec(error) => write!(formatter, "cluster-control frame codec failed: {error}"),
332        }
333    }
334}
335
336impl Error for ClusterControlWireError {}
337
338#[cfg(test)]
339mod tests {
340    use super::*;
341
342    fn session(
343        credential_kind: ClusterControlCredentialKindV1,
344        action: ClusterControlActionV2,
345    ) -> ClusterControlSessionV2 {
346        ClusterControlSessionV2 {
347            format_version: CLUSTER_CONTROL_SESSION_FORMAT_VERSION,
348            credential_kind,
349            credential: b"lcc1_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".to_vec(),
350            request_id: "capacity-sample-1".to_owned(),
351            action,
352            request_correlation_id: [7; 16],
353        }
354    }
355
356    #[test]
357    fn credential_namespaces_are_disjoint_and_canonical() -> Result<(), Box<dyn Error>> {
358        let observer = session(
359            ClusterControlCredentialKindV1::CapacityObserver,
360            ClusterControlActionV2::ObserveClusterCapacity,
361        );
362        let bytes = encode_cluster_control_session_v2(&observer)?;
363        assert_eq!(decode_cluster_control_session_v2(&bytes)?, observer);
364
365        let provisioner_backup = session(
366            ClusterControlCredentialKindV1::CollectionProvisioner,
367            ClusterControlActionV2::GetClusterBackup,
368        );
369        assert_eq!(
370            provisioner_backup.validate(),
371            Err(ClusterControlWireError::ActionDenied)
372        );
373
374        let observer_mutation = session(
375            ClusterControlCredentialKindV1::CapacityObserver,
376            ClusterControlActionV2::InitializeCollectionAuthority,
377        );
378        assert_eq!(observer_mutation.validate(), Err(ClusterControlWireError::ActionDenied));
379
380        let recovery_capacity = session(
381            ClusterControlCredentialKindV1::RecoveryOperator,
382            ClusterControlActionV2::ObserveClusterCapacity,
383        );
384        assert_eq!(recovery_capacity.validate(), Err(ClusterControlWireError::ActionDenied));
385
386        let recovery_delete = session(
387            ClusterControlCredentialKindV1::RecoveryOperator,
388            ClusterControlActionV2::DeleteClusterBackup,
389        );
390        assert!(recovery_delete.validate().is_ok());
391        let provisioner_delete = session(
392            ClusterControlCredentialKindV1::CollectionProvisioner,
393            ClusterControlActionV2::DeleteClusterBackup,
394        );
395        assert_eq!(
396            provisioner_delete.validate(),
397            Err(ClusterControlWireError::ActionDenied)
398        );
399
400        let mut public_shape = observer;
401        public_shape.credential = b"1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".to_vec();
402        assert_eq!(
403            public_shape.validate(),
404            Err(ClusterControlWireError::InvalidCredentialShape)
405        );
406        Ok(())
407    }
408
409    #[test]
410    fn decoder_refuses_predecessor_newer_unknown_and_malformed_frames() -> Result<(), Box<dyn Error>> {
411        let observer = session(
412            ClusterControlCredentialKindV1::CapacityObserver,
413            ClusterControlActionV2::ObserveClusterCapacity,
414        );
415        let predecessor = String::from_utf8(encode_cluster_control_session_v2(&observer)?)?.replacen(
416            &format!("\"format_version\":{CLUSTER_CONTROL_SESSION_FORMAT_VERSION}"),
417            &format!("\"format_version\":{}", CLUSTER_CONTROL_SESSION_FORMAT_VERSION - 1),
418            1,
419        );
420        assert_eq!(
421            decode_cluster_control_session_v2(predecessor.as_bytes()),
422            Err(ClusterControlWireError::UnsupportedVersion(1))
423        );
424        let mut newer = serde_json::to_value(&observer)?;
425        newer["format_version"] = serde_json::json!(CLUSTER_CONTROL_SESSION_FORMAT_VERSION + 1);
426        assert!(matches!(
427            decode_cluster_control_session_v2(&serde_json::to_vec(&newer)?),
428            Err(ClusterControlWireError::UnsupportedVersion(_))
429        ));
430
431        let mut unknown = serde_json::to_value(observer)?;
432        unknown["collection_id"] = serde_json::json!(1);
433        assert!(decode_cluster_control_session_v2(&serde_json::to_vec(&unknown)?).is_err());
434
435        for malformed in [
436            Vec::new(),
437            vec![0xff],
438            vec![b'x'; MAX_CLUSTER_CONTROL_SESSION_BYTES + 1],
439        ] {
440            assert!(decode_cluster_control_session_v2(&malformed).is_err());
441        }
442        Ok(())
443    }
444
445    #[test]
446    fn request_and_response_envelopes_are_strict_bounded_and_correlated() -> Result<(), Box<dyn Error>> {
447        let request = ClusterControlRequestV2 {
448            format_version: CLUSTER_CONTROL_REQUEST_FORMAT_VERSION,
449            session: session(
450                ClusterControlCredentialKindV1::CapacityObserver,
451                ClusterControlActionV2::ObserveClusterCapacity,
452            ),
453            body: serde_json::json!({}),
454        };
455        let encoded = encode_cluster_control_request_v2(&request)?;
456        assert_eq!(decode_cluster_control_request_v2(&encoded)?, request);
457
458        let predecessor_request = String::from_utf8(encoded.clone())?.replacen(
459            &format!("\"format_version\":{CLUSTER_CONTROL_REQUEST_FORMAT_VERSION}"),
460            &format!("\"format_version\":{}", CLUSTER_CONTROL_REQUEST_FORMAT_VERSION - 1),
461            1,
462        );
463        assert_eq!(
464            decode_cluster_control_request_v2(predecessor_request.as_bytes()),
465            Err(ClusterControlWireError::UnsupportedVersion(1))
466        );
467
468        let response = ClusterControlResponseV2 {
469            format_version: CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION,
470            request_id: request.session.request_id.clone(),
471            request_correlation_id: request.session.request_correlation_id,
472            outcome: ClusterControlResponseOutcomeV2::Succeeded {
473                body: serde_json::json!({"observation_id": "sample-1"}),
474            },
475        };
476        let encoded = encode_cluster_control_response_v2(&response)?;
477        assert_eq!(decode_cluster_control_response_v2(&encoded)?, response);
478
479        let predecessor_response = String::from_utf8(encoded)?.replacen(
480            &format!("\"format_version\":{CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION}"),
481            &format!("\"format_version\":{}", CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION - 1),
482            1,
483        );
484        assert_eq!(
485            decode_cluster_control_response_v2(predecessor_response.as_bytes()),
486            Err(ClusterControlWireError::UnsupportedVersion(1))
487        );
488
489        let mut unknown = serde_json::to_value(&request)?;
490        unknown["unexpected"] = serde_json::json!(true);
491        assert!(decode_cluster_control_request_v2(&serde_json::to_vec(&unknown)?).is_err());
492        assert!(matches!(
493            decode_cluster_control_request_v2(&vec![b'x'; MAX_CLUSTER_CONTROL_FRAME_BYTES + 1]),
494            Err(ClusterControlWireError::FrameTooLarge(_))
495        ));
496        Ok(())
497    }
498}
lodestone_proto/cluster_control_protocol.rs

lodestone_proto/
cluster_control_protocol.rs
