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 use lodestone_types::{CollectionId, CollectionSecurityGeneration, LogicalMillis, RecoveryEra};
14
15pub const CLUSTER_CONTROL_SESSION_FORMAT_VERSION: u16 = 3;
16pub const MAX_CLUSTER_CONTROL_SESSION_BYTES: usize = 1_024;
17pub const MAX_CLUSTER_CONTROL_CREDENTIAL_BYTES: usize = 512;
18pub const CLUSTER_CONTROL_CREDENTIAL_PREFIX: &[u8] = b"lcc1_";
19pub const CLUSTER_CONTROL_CREDENTIAL_ENCODED_BYTES: usize = 48;
20pub const CLUSTER_CONTROL_REQUEST_FORMAT_VERSION: u16 = 3;
21pub const CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION: u16 = 3;
22pub const MAX_CLUSTER_CONTROL_FRAME_BYTES: usize = 2 * 1024 * 1024;
23pub const MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES: usize = 128;
24pub const MAX_CLUSTER_CONTROL_ERROR_CODE_BYTES: usize = 96;
25pub const COLLECTION_RETIREMENT_REQUEST_FORMAT_VERSION: u16 = 1;
26pub const COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION: u16 = 1;
27pub const COLLECTION_RETIREMENT_RECEIPT_FORMAT_VERSION: u16 = 1;
28pub const MAX_COLLECTION_RETIREMENT_CURSOR_BYTES: usize = 8 * 1024;
29pub const MAX_COLLECTION_RETIREMENT_BLOCKERS: usize = 32;
30pub const COLLECTION_AUTHORITY_OBSERVATION_REQUEST_FORMAT_VERSION: u16 = 1;
31pub const COLLECTION_AUTHORITY_OBSERVATION_FORMAT_VERSION: u16 = 1;
32
33const COLLECTION_RETIREMENT_REQUEST_DIGEST_DOMAIN: &[u8] =
34    b"lodestone/cluster-control/collection-retirement-request/v1\0";
35const COLLECTION_RETIREMENT_CLOSURE_DIGEST_DOMAIN: &[u8] =
36    b"lodestone/cluster-control/collection-retirement-closure/v1\0";
37const COLLECTION_AUTHORITY_OBSERVATION_DIGEST_DOMAIN: &[u8] =
38    b"lodestone/cluster-control/collection-authority-observation/v1\0";
39
40/// Strictly read-only selection of one allocator-backed Collection identity.
41#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
42#[serde(deny_unknown_fields)]
43pub struct CollectionAuthorityObservationRequestV1 {
44    pub format_version: u16,
45    pub collection_id: CollectionId,
46}
47
48impl CollectionAuthorityObservationRequestV1 {
49    pub fn validate(&self) -> Result<(), CollectionAuthorityObservationContractError> {
50        if self.format_version != COLLECTION_AUTHORITY_OBSERVATION_REQUEST_FORMAT_VERSION {
51            return Err(CollectionAuthorityObservationContractError::InvalidRequest);
52        }
53        Ok(())
54    }
55}
56
57/// Authority visible at one exact recovery era and authenticated catalog root.
58/// `Absent` means that root contains neither live authority nor retained
59/// retirement evidence; it does not claim that the allocator never issued the
60/// Collection identity.
61#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
62#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
63pub enum CollectionAuthorityObservationStateV1 {
64    Absent,
65    Inactive {
66        security_generation: CollectionSecurityGeneration,
67    },
68    Active {
69        security_generation: CollectionSecurityGeneration,
70    },
71    Retiring {
72        security_generation: CollectionSecurityGeneration,
73    },
74    Retired {
75        security_generation: CollectionSecurityGeneration,
76        retired_at: LogicalMillis,
77    },
78}
79
80impl CollectionAuthorityObservationStateV1 {
81    #[must_use]
82    pub const fn security_generation(self) -> Option<CollectionSecurityGeneration> {
83        match self {
84            Self::Absent => None,
85            Self::Inactive { security_generation }
86            | Self::Active { security_generation }
87            | Self::Retiring { security_generation }
88            | Self::Retired {
89                security_generation, ..
90            } => Some(security_generation),
91        }
92    }
93
94    fn validate(self) -> Result<(), CollectionAuthorityObservationContractError> {
95        match self {
96            Self::Absent => Ok(()),
97            Self::Inactive { security_generation }
98            | Self::Active { security_generation }
99            | Self::Retiring { security_generation }
100                if security_generation.get() > 0 =>
101            {
102                Ok(())
103            }
104            Self::Retired {
105                security_generation,
106                retired_at,
107            } if security_generation.get() > 0 && retired_at.get() > 0 => Ok(()),
108            Self::Inactive { .. } | Self::Active { .. } | Self::Retiring { .. } | Self::Retired { .. } => {
109                Err(CollectionAuthorityObservationContractError::InvalidObservation)
110            }
111        }
112    }
113}
114
115#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
116#[serde(deny_unknown_fields)]
117pub struct CollectionAuthorityObservationV1 {
118    pub format_version: u16,
119    pub collection_id: CollectionId,
120    pub recovery_era: RecoveryEra,
121    pub visible_root_digest: [u8; 32],
122    pub observed_at: LogicalMillis,
123    pub authority: CollectionAuthorityObservationStateV1,
124    pub semantic_digest: [u8; 32],
125}
126
127impl CollectionAuthorityObservationV1 {
128    pub fn validate(&self) -> Result<(), CollectionAuthorityObservationContractError> {
129        if self.format_version != COLLECTION_AUTHORITY_OBSERVATION_FORMAT_VERSION
130            || self.visible_root_digest == [0; 32]
131            || self.semantic_digest == [0; 32]
132        {
133            return Err(CollectionAuthorityObservationContractError::InvalidObservation);
134        }
135        self.authority.validate()?;
136        if self.canonical_semantic_digest()? != self.semantic_digest {
137            return Err(CollectionAuthorityObservationContractError::InvalidObservation);
138        }
139        Ok(())
140    }
141
142    pub fn canonical_semantic_digest(&self) -> Result<[u8; 32], CollectionAuthorityObservationContractError> {
143        let encoded = serde_json::to_vec(&(
144            self.format_version,
145            self.collection_id,
146            self.recovery_era,
147            self.visible_root_digest,
148            self.observed_at,
149            self.authority,
150        ))
151        .map_err(|_| CollectionAuthorityObservationContractError::Codec)?;
152        let mut hasher = blake3::Hasher::new();
153        hasher.update(COLLECTION_AUTHORITY_OBSERVATION_DIGEST_DOMAIN);
154        hasher.update(&encoded);
155        Ok(*hasher.finalize().as_bytes())
156    }
157}
158
159#[derive(Clone, Copy, Debug, Eq, PartialEq)]
160pub enum CollectionAuthorityObservationContractError {
161    InvalidRequest,
162    InvalidObservation,
163    Codec,
164}
165
166impl Display for CollectionAuthorityObservationContractError {
167    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
168        formatter.write_str(match self {
169            Self::InvalidRequest => "Collection-authority observation request is invalid",
170            Self::InvalidObservation => "Collection-authority observation is invalid",
171            Self::Codec => "Collection-authority observation canonical encoding failed",
172        })
173    }
174}
175
176impl Error for CollectionAuthorityObservationContractError {}
177
178/// Policy admitted by the Collection provisioner. Protection and retention
179/// authorities remain blockers; this policy never grants permission to
180/// override them.
181#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
182#[serde(rename_all = "snake_case")]
183pub enum CollectionRetirementClosurePolicyV1 {
184    RetireOwnedChildren,
185}
186
187#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
188#[serde(deny_unknown_fields)]
189pub struct CollectionRetirementRequestV1 {
190    pub format_version: u16,
191    pub operation_id: [u8; 32],
192    pub collection_id: CollectionId,
193    pub expected_security_generation: CollectionSecurityGeneration,
194    pub closure_policy: CollectionRetirementClosurePolicyV1,
195}
196
197impl CollectionRetirementRequestV1 {
198    pub fn validate(&self) -> Result<(), CollectionRetirementContractError> {
199        if self.format_version != COLLECTION_RETIREMENT_REQUEST_FORMAT_VERSION
200            || self.operation_id == [0; 32]
201            || self.expected_security_generation.get() == 0
202        {
203            return Err(CollectionRetirementContractError::InvalidRequest);
204        }
205        Ok(())
206    }
207
208    pub fn canonical_request_digest(&self) -> Result<[u8; 32], CollectionRetirementContractError> {
209        self.validate()?;
210        canonical_collection_retirement_digest(COLLECTION_RETIREMENT_REQUEST_DIGEST_DOMAIN, self)
211    }
212}
213
214#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
215#[serde(rename_all = "snake_case")]
216pub enum CollectionRetirementOwnerV1 {
217    Admission,
218    Administration,
219    Index,
220    Security,
221    Operation,
222    Portability,
223    Retention,
224    Protection,
225    Usage,
226    Task,
227    ResidentRoute,
228}
229
230#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
231#[serde(rename_all = "snake_case")]
232pub enum CollectionRetirementBlockerKindV1 {
233    OwnerBusy,
234    ActiveAuthority,
235    PolicyRefusal,
236    RetainedEvidence,
237    PendingOperation,
238}
239
240#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
241#[serde(deny_unknown_fields)]
242pub struct CollectionRetirementBlockerV1 {
243    pub owner: CollectionRetirementOwnerV1,
244    pub kind: CollectionRetirementBlockerKindV1,
245}
246
247#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
248#[serde(deny_unknown_fields)]
249pub struct CollectionRetirementCursorV1 {
250    pub owner: CollectionRetirementOwnerV1,
251    #[serde(default, skip_serializing_if = "Option::is_none")]
252    pub after: Option<Vec<u8>>,
253}
254
255impl CollectionRetirementCursorV1 {
256    fn validate(&self) -> Result<(), CollectionRetirementContractError> {
257        if self
258            .after
259            .as_ref()
260            .is_some_and(|after| after.is_empty() || after.len() > MAX_COLLECTION_RETIREMENT_CURSOR_BYTES)
261        {
262            return Err(CollectionRetirementContractError::InvalidStatus);
263        }
264        Ok(())
265    }
266}
267
268#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
269#[serde(deny_unknown_fields)]
270pub struct CollectionRetirementChildSummaryV1 {
271    pub administration_resources: u64,
272    pub indexes: u64,
273    pub security_resources: u64,
274    pub operations: u64,
275}
276
277#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
278#[serde(deny_unknown_fields)]
279pub struct CollectionRetirementReceiptV1 {
280    pub format_version: u16,
281    pub operation_id: [u8; 32],
282    pub request_digest: [u8; 32],
283    pub collection_id: CollectionId,
284    pub admitted_security_generation: CollectionSecurityGeneration,
285    pub retired_security_generation: CollectionSecurityGeneration,
286    pub admitted_at: LogicalMillis,
287    pub retired_at: LogicalMillis,
288    pub final_root_digest: [u8; 32],
289    pub closure_digest: [u8; 32],
290    pub child_summary: CollectionRetirementChildSummaryV1,
291}
292
293impl CollectionRetirementReceiptV1 {
294    pub fn validate(&self) -> Result<(), CollectionRetirementContractError> {
295        if self.format_version != COLLECTION_RETIREMENT_RECEIPT_FORMAT_VERSION
296            || self.operation_id == [0; 32]
297            || self.request_digest == [0; 32]
298            || self.admitted_security_generation.get() == 0
299            || self.retired_security_generation.get() <= self.admitted_security_generation.get()
300            || self.admitted_at.get() == 0
301            || self.retired_at < self.admitted_at
302            || self.final_root_digest == [0; 32]
303            || self.closure_digest == [0; 32]
304        {
305            return Err(CollectionRetirementContractError::InvalidReceipt);
306        }
307        if self.canonical_closure_digest()? != self.closure_digest {
308            return Err(CollectionRetirementContractError::InvalidReceipt);
309        }
310        Ok(())
311    }
312
313    pub fn canonical_closure_digest(&self) -> Result<[u8; 32], CollectionRetirementContractError> {
314        canonical_collection_retirement_digest(
315            COLLECTION_RETIREMENT_CLOSURE_DIGEST_DOMAIN,
316            &(
317                self.operation_id,
318                self.request_digest,
319                self.collection_id,
320                self.admitted_security_generation,
321                self.retired_security_generation,
322                self.admitted_at,
323                self.retired_at,
324                self.final_root_digest,
325                self.child_summary,
326            ),
327        )
328    }
329}
330
331#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
332#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
333pub enum CollectionRetirementOutcomeV1 {
334    NotSubmitted {
335        format_version: u16,
336        request_digest: [u8; 32],
337    },
338    Blocked {
339        format_version: u16,
340        request_digest: [u8; 32],
341        admitted_security_generation: CollectionSecurityGeneration,
342        cursor: CollectionRetirementCursorV1,
343        child_summary: CollectionRetirementChildSummaryV1,
344        blockers: Vec<CollectionRetirementBlockerV1>,
345    },
346    InProgress {
347        format_version: u16,
348        request_digest: [u8; 32],
349        admitted_security_generation: CollectionSecurityGeneration,
350        cursor: CollectionRetirementCursorV1,
351        child_summary: CollectionRetirementChildSummaryV1,
352    },
353    /// Client-only projection used when a call may have submitted but no
354    /// authoritative response was received. The exact request is retained for
355    /// resolver replay and is never safe to replace with a fresh identity.
356    Indeterminate {
357        format_version: u16,
358        request: Box<CollectionRetirementRequestV1>,
359    },
360    Resolved {
361        format_version: u16,
362        receipt: Box<CollectionRetirementReceiptV1>,
363    },
364    KnownRetired {
365        format_version: u16,
366        request_digest: [u8; 32],
367        collection_id: CollectionId,
368        retired_security_generation: CollectionSecurityGeneration,
369        retired_at: LogicalMillis,
370    },
371    ExpiredUnknown {
372        format_version: u16,
373        request_digest: [u8; 32],
374        collection_id: CollectionId,
375        operation_id: [u8; 32],
376    },
377}
378
379impl CollectionRetirementOutcomeV1 {
380    pub fn validate(&self) -> Result<(), CollectionRetirementContractError> {
381        match self {
382            Self::NotSubmitted {
383                format_version,
384                request_digest,
385            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION && *request_digest != [0; 32] => Ok(()),
386            Self::Blocked {
387                format_version,
388                request_digest,
389                admitted_security_generation,
390                cursor,
391                blockers,
392                ..
393            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION
394                && *request_digest != [0; 32]
395                && admitted_security_generation.get() > 0
396                && !blockers.is_empty()
397                && blockers.len() <= MAX_COLLECTION_RETIREMENT_BLOCKERS
398                && blockers.windows(2).all(|pair| pair[0] < pair[1]) =>
399            {
400                cursor.validate()
401            }
402            Self::InProgress {
403                format_version,
404                request_digest,
405                admitted_security_generation,
406                cursor,
407                ..
408            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION
409                && *request_digest != [0; 32]
410                && admitted_security_generation.get() > 0 =>
411            {
412                cursor.validate()
413            }
414            Self::Indeterminate {
415                format_version,
416                request,
417            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION => request.validate(),
418            Self::Resolved {
419                format_version,
420                receipt,
421            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION => receipt.validate(),
422            Self::KnownRetired {
423                format_version,
424                request_digest,
425                retired_security_generation,
426                retired_at,
427                ..
428            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION
429                && *request_digest != [0; 32]
430                && retired_security_generation.get() > 0
431                && retired_at.get() > 0 =>
432            {
433                Ok(())
434            }
435            Self::ExpiredUnknown {
436                format_version,
437                request_digest,
438                operation_id,
439                ..
440            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION
441                && *request_digest != [0; 32]
442                && *operation_id != [0; 32] =>
443            {
444                Ok(())
445            }
446            Self::NotSubmitted { .. }
447            | Self::Blocked { .. }
448            | Self::InProgress { .. }
449            | Self::Indeterminate { .. }
450            | Self::Resolved { .. }
451            | Self::KnownRetired { .. }
452            | Self::ExpiredUnknown { .. } => Err(CollectionRetirementContractError::InvalidStatus),
453        }
454    }
455}
456
457fn canonical_collection_retirement_digest(
458    domain: &[u8],
459    value: &impl Serialize,
460) -> Result<[u8; 32], CollectionRetirementContractError> {
461    let encoded = serde_json::to_vec(value).map_err(|_| CollectionRetirementContractError::Codec)?;
462    let mut hasher = blake3::Hasher::new();
463    hasher.update(domain);
464    hasher.update(&encoded);
465    Ok(*hasher.finalize().as_bytes())
466}
467
468#[derive(Clone, Copy, Debug, Eq, PartialEq)]
469pub enum CollectionRetirementContractError {
470    InvalidRequest,
471    InvalidStatus,
472    InvalidReceipt,
473    Codec,
474}
475
476impl Display for CollectionRetirementContractError {
477    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
478        formatter.write_str(match self {
479            Self::InvalidRequest => "Collection-retirement request is invalid",
480            Self::InvalidStatus => "Collection-retirement status is invalid",
481            Self::InvalidReceipt => "Collection-retirement receipt is invalid",
482            Self::Codec => "Collection-retirement canonical encoding failed",
483        })
484    }
485}
486
487impl Error for CollectionRetirementContractError {}
488
489#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
490#[serde(rename_all = "snake_case")]
491pub enum ClusterControlCredentialKindV1 {
492    CollectionProvisioner,
493    CapacityObserver,
494    RecoveryOperator,
495}
496
497#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
498#[serde(rename_all = "snake_case")]
499pub enum ClusterControlActionV3 {
500    InitializeCollectionAuthority,
501    ActivateCollectionAuthority,
502    ResolveCollectionAuthority,
503    ObserveCollectionAuthority,
504    InspectCollectionRetirement,
505    RetireCollectionAuthority,
506    ObserveClusterCapacity,
507    ObserveResourceDiagnostics,
508    ReserveClusterControlCredentialRotation,
509    ActivateClusterControlCredentialRotation,
510    RevokeClusterControlCredential,
511    ResolveClusterControlCredential,
512    ListClusterBackups,
513    GetClusterBackup,
514    CreateClusterBackup,
515    ResolveClusterBackup,
516    DeleteClusterBackup,
517    ResolveClusterBackupDeletion,
518    GetClusterRestoreReadiness,
519    PrepareOfflineRecovery,
520}
521
522impl ClusterControlCredentialKindV1 {
523    #[must_use]
524    pub const fn admits(self, action: ClusterControlActionV3) -> bool {
525        match self {
526            Self::CollectionProvisioner => matches!(
527                action,
528                ClusterControlActionV3::InitializeCollectionAuthority
529                    | ClusterControlActionV3::ActivateCollectionAuthority
530                    | ClusterControlActionV3::ResolveCollectionAuthority
531                    | ClusterControlActionV3::ObserveCollectionAuthority
532                    | ClusterControlActionV3::InspectCollectionRetirement
533                    | ClusterControlActionV3::RetireCollectionAuthority
534                    | ClusterControlActionV3::ReserveClusterControlCredentialRotation
535                    | ClusterControlActionV3::ActivateClusterControlCredentialRotation
536                    | ClusterControlActionV3::RevokeClusterControlCredential
537                    | ClusterControlActionV3::ResolveClusterControlCredential
538            ),
539            Self::CapacityObserver => matches!(
540                action,
541                ClusterControlActionV3::ObserveClusterCapacity | ClusterControlActionV3::ObserveResourceDiagnostics
542            ),
543            Self::RecoveryOperator => matches!(
544                action,
545                ClusterControlActionV3::ReserveClusterControlCredentialRotation
546                    | ClusterControlActionV3::ActivateClusterControlCredentialRotation
547                    | ClusterControlActionV3::RevokeClusterControlCredential
548                    | ClusterControlActionV3::ResolveClusterControlCredential
549                    | ClusterControlActionV3::ListClusterBackups
550                    | ClusterControlActionV3::GetClusterBackup
551                    | ClusterControlActionV3::CreateClusterBackup
552                    | ClusterControlActionV3::ResolveClusterBackup
553                    | ClusterControlActionV3::DeleteClusterBackup
554                    | ClusterControlActionV3::ResolveClusterBackupDeletion
555                    | ClusterControlActionV3::GetClusterRestoreReadiness
556                    | ClusterControlActionV3::PrepareOfflineRecovery
557            ),
558        }
559    }
560}
561
562#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
563#[serde(deny_unknown_fields)]
564pub struct ClusterControlSessionV3 {
565    pub format_version: u16,
566    pub credential_kind: ClusterControlCredentialKindV1,
567    pub credential: Vec<u8>,
568    pub request_id: String,
569    pub action: ClusterControlActionV3,
570    pub request_correlation_id: [u8; 16],
571}
572
573impl fmt::Debug for ClusterControlSessionV3 {
574    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
575        formatter
576            .debug_struct("ClusterControlSessionV3")
577            .field("format_version", &self.format_version)
578            .field("credential_kind", &self.credential_kind)
579            .field("credential", &"[REDACTED]")
580            .field("request_id", &self.request_id)
581            .field("action", &self.action)
582            .field("request_correlation_id", &self.request_correlation_id)
583            .finish()
584    }
585}
586
587impl ClusterControlSessionV3 {
588    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
589        if self.format_version != CLUSTER_CONTROL_SESSION_FORMAT_VERSION {
590            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
591        }
592        validate_cluster_control_credential_v1(&self.credential)?;
593        if self.request_id.is_empty()
594            || self.request_id.len() > MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES
595            || self.request_id.bytes().any(|byte| byte.is_ascii_control())
596        {
597            return Err(ClusterControlWireError::InvalidRequestIdentity);
598        }
599        if !self.credential_kind.admits(self.action) {
600            return Err(ClusterControlWireError::ActionDenied);
601        }
602        Ok(())
603    }
604}
605
606pub fn validate_cluster_control_credential_v1(credential: &[u8]) -> Result<(), ClusterControlWireError> {
607    if credential.len() != CLUSTER_CONTROL_CREDENTIAL_ENCODED_BYTES
608        || !credential.starts_with(CLUSTER_CONTROL_CREDENTIAL_PREFIX)
609        || !credential[CLUSTER_CONTROL_CREDENTIAL_PREFIX.len()..]
610            .iter()
611            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
612    {
613        return Err(ClusterControlWireError::InvalidCredentialShape);
614    }
615    Ok(())
616}
617
618#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
619#[serde(deny_unknown_fields)]
620pub struct ClusterControlRequestV3 {
621    pub format_version: u16,
622    pub session: ClusterControlSessionV3,
623    pub body: serde_json::Value,
624}
625
626impl ClusterControlRequestV3 {
627    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
628        if self.format_version != CLUSTER_CONTROL_REQUEST_FORMAT_VERSION {
629            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
630        }
631        self.session.validate()?;
632        if !self.body.is_object() {
633            return Err(ClusterControlWireError::InvalidBody);
634        }
635        Ok(())
636    }
637}
638
639#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
640#[serde(rename_all = "snake_case")]
641pub enum ClusterControlResponseDispositionV3 {
642    RetrySameRequest,
643    ResolveOperation,
644    CorrectRequest,
645    Reauthenticate,
646    Terminal,
647}
648
649#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
650#[serde(tag = "outcome", rename_all = "snake_case", deny_unknown_fields)]
651pub enum ClusterControlResponseOutcomeV3 {
652    Succeeded {
653        body: serde_json::Value,
654    },
655    Refused {
656        code: String,
657        disposition: ClusterControlResponseDispositionV3,
658    },
659}
660
661#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
662#[serde(deny_unknown_fields)]
663pub struct ClusterControlResponseV3 {
664    pub format_version: u16,
665    pub request_id: String,
666    pub request_correlation_id: [u8; 16],
667    pub outcome: ClusterControlResponseOutcomeV3,
668}
669
670impl ClusterControlResponseV3 {
671    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
672        if self.format_version != CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION {
673            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
674        }
675        validate_request_id(&self.request_id)?;
676        match &self.outcome {
677            ClusterControlResponseOutcomeV3::Succeeded { body } if body.is_object() => Ok(()),
678            ClusterControlResponseOutcomeV3::Succeeded { .. } => Err(ClusterControlWireError::InvalidBody),
679            ClusterControlResponseOutcomeV3::Refused { code, .. }
680                if !code.is_empty()
681                    && code.len() <= MAX_CLUSTER_CONTROL_ERROR_CODE_BYTES
682                    && code
683                        .bytes()
684                        .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_') =>
685            {
686                Ok(())
687            }
688            ClusterControlResponseOutcomeV3::Refused { .. } => Err(ClusterControlWireError::InvalidErrorCode),
689        }
690    }
691}
692
693pub fn encode_cluster_control_request_v3(
694    request: &ClusterControlRequestV3,
695) -> Result<Vec<u8>, ClusterControlWireError> {
696    request.validate()?;
697    encode_frame(request)
698}
699
700pub fn decode_cluster_control_request_v3(bytes: &[u8]) -> Result<ClusterControlRequestV3, ClusterControlWireError> {
701    let request: ClusterControlRequestV3 = decode_frame(bytes)?;
702    request.validate()?;
703    Ok(request)
704}
705
706pub fn encode_cluster_control_response_v3(
707    response: &ClusterControlResponseV3,
708) -> Result<Vec<u8>, ClusterControlWireError> {
709    response.validate()?;
710    encode_frame(response)
711}
712
713pub fn decode_cluster_control_response_v3(bytes: &[u8]) -> Result<ClusterControlResponseV3, ClusterControlWireError> {
714    let response: ClusterControlResponseV3 = decode_frame(bytes)?;
715    response.validate()?;
716    Ok(response)
717}
718
719fn encode_frame(value: &impl Serialize) -> Result<Vec<u8>, ClusterControlWireError> {
720    let bytes = serde_json::to_vec(value).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
721    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_FRAME_BYTES {
722        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
723    }
724    Ok(bytes)
725}
726
727fn decode_frame<T: for<'de> Deserialize<'de> + Serialize>(bytes: &[u8]) -> Result<T, ClusterControlWireError> {
728    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_FRAME_BYTES {
729        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
730    }
731    let value = serde_json::from_slice(bytes).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
732    if serde_json::to_vec(&value).map_err(|error| ClusterControlWireError::Codec(error.to_string()))? != bytes {
733        return Err(ClusterControlWireError::NonCanonical);
734    }
735    Ok(value)
736}
737
738fn validate_request_id(request_id: &str) -> Result<(), ClusterControlWireError> {
739    if request_id.is_empty()
740        || request_id.len() > MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES
741        || request_id.bytes().any(|byte| byte.is_ascii_control())
742    {
743        return Err(ClusterControlWireError::InvalidRequestIdentity);
744    }
745    Ok(())
746}
747
748pub fn encode_cluster_control_session_v3(
749    session: &ClusterControlSessionV3,
750) -> Result<Vec<u8>, ClusterControlWireError> {
751    session.validate()?;
752    let bytes = serde_json::to_vec(session).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
753    if bytes.len() > MAX_CLUSTER_CONTROL_SESSION_BYTES {
754        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
755    }
756    Ok(bytes)
757}
758
759pub fn decode_cluster_control_session_v3(bytes: &[u8]) -> Result<ClusterControlSessionV3, ClusterControlWireError> {
760    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_SESSION_BYTES {
761        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
762    }
763    let session: ClusterControlSessionV3 =
764        serde_json::from_slice(bytes).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
765    session.validate()?;
766    if serde_json::to_vec(&session).map_err(|error| ClusterControlWireError::Codec(error.to_string()))? != bytes {
767        return Err(ClusterControlWireError::NonCanonical);
768    }
769    Ok(session)
770}
771
772#[derive(Clone, Debug, Eq, PartialEq)]
773pub enum ClusterControlWireError {
774    UnsupportedVersion(u16),
775    InvalidCredentialShape,
776    InvalidRequestIdentity,
777    InvalidBody,
778    InvalidErrorCode,
779    ActionDenied,
780    FrameTooLarge(usize),
781    NonCanonical,
782    Codec(String),
783}
784
785impl Display for ClusterControlWireError {
786    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
787        match self {
788            Self::UnsupportedVersion(version) => {
789                write!(formatter, "unsupported cluster-control wire version {version}")
790            }
791            Self::InvalidCredentialShape => formatter.write_str("invalid cluster-control credential shape"),
792            Self::InvalidRequestIdentity => formatter.write_str("invalid cluster-control request identity"),
793            Self::InvalidBody => formatter.write_str("invalid cluster-control request or response body"),
794            Self::InvalidErrorCode => formatter.write_str("invalid cluster-control error code"),
795            Self::ActionDenied => formatter.write_str("cluster-control credential kind does not admit the action"),
796            Self::FrameTooLarge(size) => write!(formatter, "cluster-control frame size {size} exceeds the bound"),
797            Self::NonCanonical => formatter.write_str("cluster-control frame is not canonical"),
798            Self::Codec(error) => write!(formatter, "cluster-control frame codec failed: {error}"),
799        }
800    }
801}
802
803impl Error for ClusterControlWireError {}
804
805#[cfg(test)]
806mod tests {
807    use super::*;
808
809    fn retirement_request() -> Result<CollectionRetirementRequestV1, Box<dyn Error>> {
810        Ok(CollectionRetirementRequestV1 {
811            format_version: COLLECTION_RETIREMENT_REQUEST_FORMAT_VERSION,
812            operation_id: [11; 32],
813            collection_id: CollectionId::new(7)?,
814            expected_security_generation: CollectionSecurityGeneration::new(4),
815            closure_policy: CollectionRetirementClosurePolicyV1::RetireOwnedChildren,
816        })
817    }
818
819    fn session(
820        credential_kind: ClusterControlCredentialKindV1,
821        action: ClusterControlActionV3,
822    ) -> ClusterControlSessionV3 {
823        ClusterControlSessionV3 {
824            format_version: CLUSTER_CONTROL_SESSION_FORMAT_VERSION,
825            credential_kind,
826            credential: b"lcc1_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".to_vec(),
827            request_id: "capacity-sample-1".to_owned(),
828            action,
829            request_correlation_id: [7; 16],
830        }
831    }
832
833    #[test]
834    fn credential_namespaces_are_disjoint_and_canonical() -> Result<(), Box<dyn Error>> {
835        let observer = session(
836            ClusterControlCredentialKindV1::CapacityObserver,
837            ClusterControlActionV3::ObserveClusterCapacity,
838        );
839        let bytes = encode_cluster_control_session_v3(&observer)?;
840        assert_eq!(decode_cluster_control_session_v3(&bytes)?, observer);
841
842        let provisioner_observation = session(
843            ClusterControlCredentialKindV1::CollectionProvisioner,
844            ClusterControlActionV3::ObserveCollectionAuthority,
845        );
846        assert!(provisioner_observation.validate().is_ok());
847        for credential_kind in [
848            ClusterControlCredentialKindV1::CapacityObserver,
849            ClusterControlCredentialKindV1::RecoveryOperator,
850        ] {
851            assert_eq!(
852                session(credential_kind, ClusterControlActionV3::ObserveCollectionAuthority).validate(),
853                Err(ClusterControlWireError::ActionDenied)
854            );
855        }
856
857        let provisioner_backup = session(
858            ClusterControlCredentialKindV1::CollectionProvisioner,
859            ClusterControlActionV3::GetClusterBackup,
860        );
861        assert_eq!(
862            provisioner_backup.validate(),
863            Err(ClusterControlWireError::ActionDenied)
864        );
865
866        let observer_mutation = session(
867            ClusterControlCredentialKindV1::CapacityObserver,
868            ClusterControlActionV3::InitializeCollectionAuthority,
869        );
870        assert_eq!(observer_mutation.validate(), Err(ClusterControlWireError::ActionDenied));
871
872        let recovery_capacity = session(
873            ClusterControlCredentialKindV1::RecoveryOperator,
874            ClusterControlActionV3::ObserveClusterCapacity,
875        );
876        assert_eq!(recovery_capacity.validate(), Err(ClusterControlWireError::ActionDenied));
877
878        let recovery_delete = session(
879            ClusterControlCredentialKindV1::RecoveryOperator,
880            ClusterControlActionV3::DeleteClusterBackup,
881        );
882        assert!(recovery_delete.validate().is_ok());
883        let provisioner_delete = session(
884            ClusterControlCredentialKindV1::CollectionProvisioner,
885            ClusterControlActionV3::DeleteClusterBackup,
886        );
887        assert_eq!(
888            provisioner_delete.validate(),
889            Err(ClusterControlWireError::ActionDenied)
890        );
891
892        let mut public_shape = observer;
893        public_shape.credential = b"1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".to_vec();
894        assert_eq!(
895            public_shape.validate(),
896            Err(ClusterControlWireError::InvalidCredentialShape)
897        );
898        Ok(())
899    }
900
901    #[test]
902    fn decoder_refuses_predecessor_newer_unknown_and_malformed_frames() -> Result<(), Box<dyn Error>> {
903        let observer = session(
904            ClusterControlCredentialKindV1::CapacityObserver,
905            ClusterControlActionV3::ObserveClusterCapacity,
906        );
907        let predecessor = String::from_utf8(encode_cluster_control_session_v3(&observer)?)?.replacen(
908            &format!("\"format_version\":{CLUSTER_CONTROL_SESSION_FORMAT_VERSION}"),
909            &format!("\"format_version\":{}", CLUSTER_CONTROL_SESSION_FORMAT_VERSION - 1),
910            1,
911        );
912        assert_eq!(
913            decode_cluster_control_session_v3(predecessor.as_bytes()),
914            Err(ClusterControlWireError::UnsupportedVersion(2))
915        );
916        let mut newer = serde_json::to_value(&observer)?;
917        newer["format_version"] = serde_json::json!(CLUSTER_CONTROL_SESSION_FORMAT_VERSION + 1);
918        assert!(matches!(
919            decode_cluster_control_session_v3(&serde_json::to_vec(&newer)?),
920            Err(ClusterControlWireError::UnsupportedVersion(_))
921        ));
922
923        let mut unknown = serde_json::to_value(observer)?;
924        unknown["collection_id"] = serde_json::json!(1);
925        assert!(decode_cluster_control_session_v3(&serde_json::to_vec(&unknown)?).is_err());
926
927        for malformed in [
928            Vec::new(),
929            vec![0xff],
930            vec![b'x'; MAX_CLUSTER_CONTROL_SESSION_BYTES + 1],
931        ] {
932            assert!(decode_cluster_control_session_v3(&malformed).is_err());
933        }
934        Ok(())
935    }
936
937    #[test]
938    fn request_and_response_envelopes_are_strict_bounded_and_correlated() -> Result<(), Box<dyn Error>> {
939        let request = ClusterControlRequestV3 {
940            format_version: CLUSTER_CONTROL_REQUEST_FORMAT_VERSION,
941            session: session(
942                ClusterControlCredentialKindV1::CapacityObserver,
943                ClusterControlActionV3::ObserveClusterCapacity,
944            ),
945            body: serde_json::json!({}),
946        };
947        let encoded = encode_cluster_control_request_v3(&request)?;
948        assert_eq!(decode_cluster_control_request_v3(&encoded)?, request);
949
950        let predecessor_request = String::from_utf8(encoded.clone())?.replacen(
951            &format!("\"format_version\":{CLUSTER_CONTROL_REQUEST_FORMAT_VERSION}"),
952            &format!("\"format_version\":{}", CLUSTER_CONTROL_REQUEST_FORMAT_VERSION - 1),
953            1,
954        );
955        assert_eq!(
956            decode_cluster_control_request_v3(predecessor_request.as_bytes()),
957            Err(ClusterControlWireError::UnsupportedVersion(2))
958        );
959
960        let response = ClusterControlResponseV3 {
961            format_version: CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION,
962            request_id: request.session.request_id.clone(),
963            request_correlation_id: request.session.request_correlation_id,
964            outcome: ClusterControlResponseOutcomeV3::Succeeded {
965                body: serde_json::json!({"observation_id": "sample-1"}),
966            },
967        };
968        let encoded = encode_cluster_control_response_v3(&response)?;
969        assert_eq!(decode_cluster_control_response_v3(&encoded)?, response);
970
971        let predecessor_response = String::from_utf8(encoded)?.replacen(
972            &format!("\"format_version\":{CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION}"),
973            &format!("\"format_version\":{}", CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION - 1),
974            1,
975        );
976        assert_eq!(
977            decode_cluster_control_response_v3(predecessor_response.as_bytes()),
978            Err(ClusterControlWireError::UnsupportedVersion(2))
979        );
980
981        let mut unknown = serde_json::to_value(&request)?;
982        unknown["unexpected"] = serde_json::json!(true);
983        assert!(decode_cluster_control_request_v3(&serde_json::to_vec(&unknown)?).is_err());
984        assert!(matches!(
985            decode_cluster_control_request_v3(&vec![b'x'; MAX_CLUSTER_CONTROL_FRAME_BYTES + 1]),
986            Err(ClusterControlWireError::FrameTooLarge(_))
987        ));
988        Ok(())
989    }
990
991    #[test]
992    fn collection_retirement_identity_binds_every_admitted_semantic() -> Result<(), Box<dyn Error>> {
993        let request = retirement_request()?;
994        let digest = request.canonical_request_digest()?;
995        let mut changed = request.clone();
996        changed.expected_security_generation = CollectionSecurityGeneration::new(5);
997        assert_ne!(changed.canonical_request_digest()?, digest);
998
999        let mut invalid = request;
1000        invalid.operation_id = [0; 32];
1001        assert_eq!(
1002            invalid.validate(),
1003            Err(CollectionRetirementContractError::InvalidRequest)
1004        );
1005        Ok(())
1006    }
1007
1008    #[test]
1009    fn collection_authority_observation_digest_binds_identity_root_era_and_state() -> Result<(), Box<dyn Error>> {
1010        let request = CollectionAuthorityObservationRequestV1 {
1011            format_version: COLLECTION_AUTHORITY_OBSERVATION_REQUEST_FORMAT_VERSION,
1012            collection_id: CollectionId::new(7)?,
1013        };
1014        request.validate()?;
1015        let mut observation = CollectionAuthorityObservationV1 {
1016            format_version: COLLECTION_AUTHORITY_OBSERVATION_FORMAT_VERSION,
1017            collection_id: request.collection_id,
1018            recovery_era: RecoveryEra::new(3)?,
1019            visible_root_digest: [17; 32],
1020            observed_at: LogicalMillis::new(19),
1021            authority: CollectionAuthorityObservationStateV1::Active {
1022                security_generation: CollectionSecurityGeneration::new(4),
1023            },
1024            semantic_digest: [0; 32],
1025        };
1026        observation.semantic_digest = observation.canonical_semantic_digest()?;
1027        observation.validate()?;
1028
1029        let digest = observation.semantic_digest;
1030        observation.collection_id = CollectionId::new(8)?;
1031        assert_eq!(
1032            observation.validate(),
1033            Err(CollectionAuthorityObservationContractError::InvalidObservation)
1034        );
1035        observation.collection_id = request.collection_id;
1036        observation.authority = CollectionAuthorityObservationStateV1::Retiring {
1037            security_generation: CollectionSecurityGeneration::new(4),
1038        };
1039        assert_ne!(observation.canonical_semantic_digest()?, digest);
1040
1041        let malformed = serde_json::json!({
1042            "format_version": COLLECTION_AUTHORITY_OBSERVATION_FORMAT_VERSION,
1043            "collection_id": request.collection_id,
1044            "recovery_era": RecoveryEra::new(3)?,
1045            "visible_root_digest": vec![17; 32],
1046            "observed_at": LogicalMillis::new(19),
1047            "authority": {"state": "active", "security_generation": 4},
1048            "semantic_digest": vec![0; 32],
1049            "unexpected": true
1050        });
1051        assert!(serde_json::from_value::<CollectionAuthorityObservationV1>(malformed).is_err());
1052        Ok(())
1053    }
1054
1055    #[test]
1056    fn collection_retirement_receipt_and_recovery_statuses_fail_closed() -> Result<(), Box<dyn Error>> {
1057        let request = retirement_request()?;
1058        let mut receipt = CollectionRetirementReceiptV1 {
1059            format_version: COLLECTION_RETIREMENT_RECEIPT_FORMAT_VERSION,
1060            operation_id: request.operation_id,
1061            request_digest: request.canonical_request_digest()?,
1062            collection_id: request.collection_id,
1063            admitted_security_generation: CollectionSecurityGeneration::new(5),
1064            retired_security_generation: CollectionSecurityGeneration::new(6),
1065            admitted_at: LogicalMillis::new(12),
1066            retired_at: LogicalMillis::new(18),
1067            final_root_digest: [21; 32],
1068            closure_digest: [0; 32],
1069            child_summary: CollectionRetirementChildSummaryV1 {
1070                administration_resources: 2,
1071                indexes: 3,
1072                security_resources: 4,
1073                operations: 0,
1074            },
1075        };
1076        receipt.closure_digest = receipt.canonical_closure_digest()?;
1077        receipt.validate()?;
1078        let closure_digest = receipt.closure_digest;
1079        receipt.child_summary.indexes += 1;
1080        assert_eq!(
1081            receipt.validate(),
1082            Err(CollectionRetirementContractError::InvalidReceipt)
1083        );
1084        assert_ne!(receipt.canonical_closure_digest()?, closure_digest);
1085
1086        let blocked = CollectionRetirementOutcomeV1::Blocked {
1087            format_version: COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION,
1088            request_digest: request.canonical_request_digest()?,
1089            admitted_security_generation: CollectionSecurityGeneration::new(5),
1090            cursor: CollectionRetirementCursorV1 {
1091                owner: CollectionRetirementOwnerV1::Protection,
1092                after: None,
1093            },
1094            child_summary: CollectionRetirementChildSummaryV1::default(),
1095            blockers: vec![CollectionRetirementBlockerV1 {
1096                owner: CollectionRetirementOwnerV1::Protection,
1097                kind: CollectionRetirementBlockerKindV1::PolicyRefusal,
1098            }],
1099        };
1100        blocked.validate()?;
1101        let indeterminate = CollectionRetirementOutcomeV1::Indeterminate {
1102            format_version: COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION,
1103            request: Box::new(request),
1104        };
1105        indeterminate.validate()?;
1106        Ok(())
1107    }
1108}
lodestone_proto/cluster_control_protocol.rs

lodestone_proto/
cluster_control_protocol.rs
