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 = 1;
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 = 1;
19pub const CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION: u16 = 1;
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 ClusterControlActionV1 {
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    GetClusterRestoreReadiness,
51    PrepareOfflineRecovery,
52}
53
54impl ClusterControlCredentialKindV1 {
55    #[must_use]
56    pub const fn admits(self, action: ClusterControlActionV1) -> bool {
57        match self {
58            Self::CollectionProvisioner => matches!(
59                action,
60                ClusterControlActionV1::InitializeCollectionAuthority
61                    | ClusterControlActionV1::ActivateCollectionAuthority
62                    | ClusterControlActionV1::ResolveCollectionAuthority
63                    | ClusterControlActionV1::InspectCollectionRetirement
64                    | ClusterControlActionV1::RetireCollectionAuthority
65                    | ClusterControlActionV1::ReserveClusterControlCredentialRotation
66                    | ClusterControlActionV1::ActivateClusterControlCredentialRotation
67                    | ClusterControlActionV1::RevokeClusterControlCredential
68                    | ClusterControlActionV1::ResolveClusterControlCredential
69            ),
70            Self::CapacityObserver => matches!(
71                action,
72                ClusterControlActionV1::ObserveClusterCapacity | ClusterControlActionV1::ObserveResourceDiagnostics
73            ),
74            Self::RecoveryOperator => matches!(
75                action,
76                ClusterControlActionV1::ReserveClusterControlCredentialRotation
77                    | ClusterControlActionV1::ActivateClusterControlCredentialRotation
78                    | ClusterControlActionV1::RevokeClusterControlCredential
79                    | ClusterControlActionV1::ResolveClusterControlCredential
80                    | ClusterControlActionV1::ListClusterBackups
81                    | ClusterControlActionV1::GetClusterBackup
82                    | ClusterControlActionV1::CreateClusterBackup
83                    | ClusterControlActionV1::ResolveClusterBackup
84                    | ClusterControlActionV1::GetClusterRestoreReadiness
85                    | ClusterControlActionV1::PrepareOfflineRecovery
86            ),
87        }
88    }
89}
90
91#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
92#[serde(deny_unknown_fields)]
93pub struct ClusterControlSessionV1 {
94    pub format_version: u16,
95    pub credential_kind: ClusterControlCredentialKindV1,
96    pub credential: Vec<u8>,
97    pub request_id: String,
98    pub action: ClusterControlActionV1,
99    pub request_correlation_id: [u8; 16],
100}
101
102impl fmt::Debug for ClusterControlSessionV1 {
103    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
104        formatter
105            .debug_struct("ClusterControlSessionV1")
106            .field("format_version", &self.format_version)
107            .field("credential_kind", &self.credential_kind)
108            .field("credential", &"[REDACTED]")
109            .field("request_id", &self.request_id)
110            .field("action", &self.action)
111            .field("request_correlation_id", &self.request_correlation_id)
112            .finish()
113    }
114}
115
116impl ClusterControlSessionV1 {
117    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
118        if self.format_version != CLUSTER_CONTROL_SESSION_FORMAT_VERSION {
119            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
120        }
121        validate_cluster_control_credential_v1(&self.credential)?;
122        if self.request_id.is_empty()
123            || self.request_id.len() > MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES
124            || self.request_id.bytes().any(|byte| byte.is_ascii_control())
125        {
126            return Err(ClusterControlWireError::InvalidRequestIdentity);
127        }
128        if !self.credential_kind.admits(self.action) {
129            return Err(ClusterControlWireError::ActionDenied);
130        }
131        Ok(())
132    }
133}
134
135pub fn validate_cluster_control_credential_v1(credential: &[u8]) -> Result<(), ClusterControlWireError> {
136    if credential.len() != CLUSTER_CONTROL_CREDENTIAL_ENCODED_BYTES
137        || !credential.starts_with(CLUSTER_CONTROL_CREDENTIAL_PREFIX)
138        || !credential[CLUSTER_CONTROL_CREDENTIAL_PREFIX.len()..]
139            .iter()
140            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
141    {
142        return Err(ClusterControlWireError::InvalidCredentialShape);
143    }
144    Ok(())
145}
146
147#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
148#[serde(deny_unknown_fields)]
149pub struct ClusterControlRequestV1 {
150    pub format_version: u16,
151    pub session: ClusterControlSessionV1,
152    pub body: serde_json::Value,
153}
154
155impl ClusterControlRequestV1 {
156    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
157        if self.format_version != CLUSTER_CONTROL_REQUEST_FORMAT_VERSION {
158            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
159        }
160        self.session.validate()?;
161        if !self.body.is_object() {
162            return Err(ClusterControlWireError::InvalidBody);
163        }
164        Ok(())
165    }
166}
167
168#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
169#[serde(rename_all = "snake_case")]
170pub enum ClusterControlResponseDispositionV1 {
171    RetrySameRequest,
172    ResolveOperation,
173    CorrectRequest,
174    Reauthenticate,
175    Terminal,
176}
177
178#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
179#[serde(tag = "outcome", rename_all = "snake_case", deny_unknown_fields)]
180pub enum ClusterControlResponseOutcomeV1 {
181    Succeeded {
182        body: serde_json::Value,
183    },
184    Refused {
185        code: String,
186        disposition: ClusterControlResponseDispositionV1,
187    },
188}
189
190#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
191#[serde(deny_unknown_fields)]
192pub struct ClusterControlResponseV1 {
193    pub format_version: u16,
194    pub request_id: String,
195    pub request_correlation_id: [u8; 16],
196    pub outcome: ClusterControlResponseOutcomeV1,
197}
198
199impl ClusterControlResponseV1 {
200    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
201        if self.format_version != CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION {
202            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
203        }
204        validate_request_id(&self.request_id)?;
205        match &self.outcome {
206            ClusterControlResponseOutcomeV1::Succeeded { body } if body.is_object() => Ok(()),
207            ClusterControlResponseOutcomeV1::Succeeded { .. } => Err(ClusterControlWireError::InvalidBody),
208            ClusterControlResponseOutcomeV1::Refused { code, .. }
209                if !code.is_empty()
210                    && code.len() <= MAX_CLUSTER_CONTROL_ERROR_CODE_BYTES
211                    && code
212                        .bytes()
213                        .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_') =>
214            {
215                Ok(())
216            }
217            ClusterControlResponseOutcomeV1::Refused { .. } => Err(ClusterControlWireError::InvalidErrorCode),
218        }
219    }
220}
221
222pub fn encode_cluster_control_request_v1(
223    request: &ClusterControlRequestV1,
224) -> Result<Vec<u8>, ClusterControlWireError> {
225    request.validate()?;
226    encode_frame(request)
227}
228
229pub fn decode_cluster_control_request_v1(bytes: &[u8]) -> Result<ClusterControlRequestV1, ClusterControlWireError> {
230    let request: ClusterControlRequestV1 = decode_frame(bytes)?;
231    request.validate()?;
232    Ok(request)
233}
234
235pub fn encode_cluster_control_response_v1(
236    response: &ClusterControlResponseV1,
237) -> Result<Vec<u8>, ClusterControlWireError> {
238    response.validate()?;
239    encode_frame(response)
240}
241
242pub fn decode_cluster_control_response_v1(bytes: &[u8]) -> Result<ClusterControlResponseV1, ClusterControlWireError> {
243    let response: ClusterControlResponseV1 = decode_frame(bytes)?;
244    response.validate()?;
245    Ok(response)
246}
247
248fn encode_frame(value: &impl Serialize) -> Result<Vec<u8>, ClusterControlWireError> {
249    let bytes = serde_json::to_vec(value).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
250    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_FRAME_BYTES {
251        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
252    }
253    Ok(bytes)
254}
255
256fn decode_frame<T: for<'de> Deserialize<'de> + Serialize>(bytes: &[u8]) -> Result<T, ClusterControlWireError> {
257    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_FRAME_BYTES {
258        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
259    }
260    let value = serde_json::from_slice(bytes).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
261    if serde_json::to_vec(&value).map_err(|error| ClusterControlWireError::Codec(error.to_string()))? != bytes {
262        return Err(ClusterControlWireError::NonCanonical);
263    }
264    Ok(value)
265}
266
267fn validate_request_id(request_id: &str) -> Result<(), ClusterControlWireError> {
268    if request_id.is_empty()
269        || request_id.len() > MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES
270        || request_id.bytes().any(|byte| byte.is_ascii_control())
271    {
272        return Err(ClusterControlWireError::InvalidRequestIdentity);
273    }
274    Ok(())
275}
276
277pub fn encode_cluster_control_session_v1(
278    session: &ClusterControlSessionV1,
279) -> Result<Vec<u8>, ClusterControlWireError> {
280    session.validate()?;
281    let bytes = serde_json::to_vec(session).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
282    if bytes.len() > MAX_CLUSTER_CONTROL_SESSION_BYTES {
283        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
284    }
285    Ok(bytes)
286}
287
288pub fn decode_cluster_control_session_v1(bytes: &[u8]) -> Result<ClusterControlSessionV1, ClusterControlWireError> {
289    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_SESSION_BYTES {
290        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
291    }
292    let session: ClusterControlSessionV1 =
293        serde_json::from_slice(bytes).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
294    session.validate()?;
295    if serde_json::to_vec(&session).map_err(|error| ClusterControlWireError::Codec(error.to_string()))? != bytes {
296        return Err(ClusterControlWireError::NonCanonical);
297    }
298    Ok(session)
299}
300
301#[derive(Clone, Debug, Eq, PartialEq)]
302pub enum ClusterControlWireError {
303    UnsupportedVersion(u16),
304    InvalidCredentialShape,
305    InvalidRequestIdentity,
306    InvalidBody,
307    InvalidErrorCode,
308    ActionDenied,
309    FrameTooLarge(usize),
310    NonCanonical,
311    Codec(String),
312}
313
314impl Display for ClusterControlWireError {
315    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
316        match self {
317            Self::UnsupportedVersion(version) => {
318                write!(formatter, "unsupported cluster-control wire version {version}")
319            }
320            Self::InvalidCredentialShape => formatter.write_str("invalid cluster-control credential shape"),
321            Self::InvalidRequestIdentity => formatter.write_str("invalid cluster-control request identity"),
322            Self::InvalidBody => formatter.write_str("invalid cluster-control request or response body"),
323            Self::InvalidErrorCode => formatter.write_str("invalid cluster-control error code"),
324            Self::ActionDenied => formatter.write_str("cluster-control credential kind does not admit the action"),
325            Self::FrameTooLarge(size) => write!(formatter, "cluster-control frame size {size} exceeds the bound"),
326            Self::NonCanonical => formatter.write_str("cluster-control frame is not canonical"),
327            Self::Codec(error) => write!(formatter, "cluster-control frame codec failed: {error}"),
328        }
329    }
330}
331
332impl Error for ClusterControlWireError {}
333
334#[cfg(test)]
335mod tests {
336    use super::*;
337
338    fn session(
339        credential_kind: ClusterControlCredentialKindV1,
340        action: ClusterControlActionV1,
341    ) -> ClusterControlSessionV1 {
342        ClusterControlSessionV1 {
343            format_version: CLUSTER_CONTROL_SESSION_FORMAT_VERSION,
344            credential_kind,
345            credential: b"lcc1_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".to_vec(),
346            request_id: "capacity-sample-1".to_owned(),
347            action,
348            request_correlation_id: [7; 16],
349        }
350    }
351
352    #[test]
353    fn credential_namespaces_are_disjoint_and_canonical() -> Result<(), Box<dyn Error>> {
354        let observer = session(
355            ClusterControlCredentialKindV1::CapacityObserver,
356            ClusterControlActionV1::ObserveClusterCapacity,
357        );
358        let bytes = encode_cluster_control_session_v1(&observer)?;
359        assert_eq!(decode_cluster_control_session_v1(&bytes)?, observer);
360
361        let provisioner_backup = session(
362            ClusterControlCredentialKindV1::CollectionProvisioner,
363            ClusterControlActionV1::GetClusterBackup,
364        );
365        assert_eq!(
366            provisioner_backup.validate(),
367            Err(ClusterControlWireError::ActionDenied)
368        );
369
370        let observer_mutation = session(
371            ClusterControlCredentialKindV1::CapacityObserver,
372            ClusterControlActionV1::InitializeCollectionAuthority,
373        );
374        assert_eq!(observer_mutation.validate(), Err(ClusterControlWireError::ActionDenied));
375
376        let recovery_capacity = session(
377            ClusterControlCredentialKindV1::RecoveryOperator,
378            ClusterControlActionV1::ObserveClusterCapacity,
379        );
380        assert_eq!(recovery_capacity.validate(), Err(ClusterControlWireError::ActionDenied));
381
382        let mut public_shape = observer;
383        public_shape.credential = b"1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".to_vec();
384        assert_eq!(
385            public_shape.validate(),
386            Err(ClusterControlWireError::InvalidCredentialShape)
387        );
388        Ok(())
389    }
390
391    #[test]
392    fn decoder_refuses_newer_unknown_and_malformed_frames() -> Result<(), Box<dyn Error>> {
393        let observer = session(
394            ClusterControlCredentialKindV1::CapacityObserver,
395            ClusterControlActionV1::ObserveClusterCapacity,
396        );
397        let mut newer = serde_json::to_value(&observer)?;
398        newer["format_version"] = serde_json::json!(CLUSTER_CONTROL_SESSION_FORMAT_VERSION + 1);
399        assert!(matches!(
400            decode_cluster_control_session_v1(&serde_json::to_vec(&newer)?),
401            Err(ClusterControlWireError::UnsupportedVersion(_))
402        ));
403
404        let mut unknown = serde_json::to_value(observer)?;
405        unknown["collection_id"] = serde_json::json!(1);
406        assert!(decode_cluster_control_session_v1(&serde_json::to_vec(&unknown)?).is_err());
407
408        for malformed in [
409            Vec::new(),
410            vec![0xff],
411            vec![b'x'; MAX_CLUSTER_CONTROL_SESSION_BYTES + 1],
412        ] {
413            assert!(decode_cluster_control_session_v1(&malformed).is_err());
414        }
415        Ok(())
416    }
417
418    #[test]
419    fn request_and_response_envelopes_are_strict_bounded_and_correlated() -> Result<(), Box<dyn Error>> {
420        let request = ClusterControlRequestV1 {
421            format_version: CLUSTER_CONTROL_REQUEST_FORMAT_VERSION,
422            session: session(
423                ClusterControlCredentialKindV1::CapacityObserver,
424                ClusterControlActionV1::ObserveClusterCapacity,
425            ),
426            body: serde_json::json!({}),
427        };
428        let encoded = encode_cluster_control_request_v1(&request)?;
429        assert_eq!(decode_cluster_control_request_v1(&encoded)?, request);
430
431        let response = ClusterControlResponseV1 {
432            format_version: CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION,
433            request_id: request.session.request_id.clone(),
434            request_correlation_id: request.session.request_correlation_id,
435            outcome: ClusterControlResponseOutcomeV1::Succeeded {
436                body: serde_json::json!({"observation_id": "sample-1"}),
437            },
438        };
439        let encoded = encode_cluster_control_response_v1(&response)?;
440        assert_eq!(decode_cluster_control_response_v1(&encoded)?, response);
441
442        let mut unknown = serde_json::to_value(&request)?;
443        unknown["unexpected"] = serde_json::json!(true);
444        assert!(decode_cluster_control_request_v1(&serde_json::to_vec(&unknown)?).is_err());
445        assert!(matches!(
446            decode_cluster_control_request_v1(&vec![b'x'; MAX_CLUSTER_CONTROL_FRAME_BYTES + 1]),
447            Err(ClusterControlWireError::FrameTooLarge(_))
448        ));
449        Ok(())
450    }
451}
lodestone_proto/cluster_control_protocol.rs

lodestone_proto/
cluster_control_protocol.rs
