lodestone-proto 0.0.26

//! Strict cluster-control session wire values shared by servers and external clients.
//!
//! Cluster control is deliberately disjoint from [`crate::PublicSessionV1`].
//! A cluster-control credential has no Collection grant, and the action enum
//! makes each credential namespace's closed capability set explicit before a
//! listener or durable credential authority is allowed to accept it.

use std::error::Error;
use std::fmt::{self, Display, Formatter};

use serde::{Deserialize, Serialize};

pub const CLUSTER_CONTROL_SESSION_FORMAT_VERSION: u16 = 2;
pub const MAX_CLUSTER_CONTROL_SESSION_BYTES: usize = 1_024;
pub const MAX_CLUSTER_CONTROL_CREDENTIAL_BYTES: usize = 512;
pub const CLUSTER_CONTROL_CREDENTIAL_PREFIX: &[u8] = b"lcc1_";
pub const CLUSTER_CONTROL_CREDENTIAL_ENCODED_BYTES: usize = 48;
pub const CLUSTER_CONTROL_REQUEST_FORMAT_VERSION: u16 = 2;
pub const CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION: u16 = 2;
pub const MAX_CLUSTER_CONTROL_FRAME_BYTES: usize = 2 * 1024 * 1024;
pub const MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES: usize = 128;
pub const MAX_CLUSTER_CONTROL_ERROR_CODE_BYTES: usize = 96;

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ClusterControlCredentialKindV1 {
    CollectionProvisioner,
    CapacityObserver,
    RecoveryOperator,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ClusterControlActionV2 {
    InitializeCollectionAuthority,
    ActivateCollectionAuthority,
    ResolveCollectionAuthority,
    InspectCollectionRetirement,
    RetireCollectionAuthority,
    ObserveClusterCapacity,
    ObserveResourceDiagnostics,
    ReserveClusterControlCredentialRotation,
    ActivateClusterControlCredentialRotation,
    RevokeClusterControlCredential,
    ResolveClusterControlCredential,
    ListClusterBackups,
    GetClusterBackup,
    CreateClusterBackup,
    ResolveClusterBackup,
    DeleteClusterBackup,
    ResolveClusterBackupDeletion,
    GetClusterRestoreReadiness,
    PrepareOfflineRecovery,
}

impl ClusterControlCredentialKindV1 {
    #[must_use]
    pub const fn admits(self, action: ClusterControlActionV2) -> bool {
        match self {
            Self::CollectionProvisioner => matches!(
                action,
                ClusterControlActionV2::InitializeCollectionAuthority
                    | ClusterControlActionV2::ActivateCollectionAuthority
                    | ClusterControlActionV2::ResolveCollectionAuthority
                    | ClusterControlActionV2::InspectCollectionRetirement
                    | ClusterControlActionV2::RetireCollectionAuthority
                    | ClusterControlActionV2::ReserveClusterControlCredentialRotation
                    | ClusterControlActionV2::ActivateClusterControlCredentialRotation
                    | ClusterControlActionV2::RevokeClusterControlCredential
                    | ClusterControlActionV2::ResolveClusterControlCredential
            ),
            Self::CapacityObserver => matches!(
                action,
                ClusterControlActionV2::ObserveClusterCapacity | ClusterControlActionV2::ObserveResourceDiagnostics
            ),
            Self::RecoveryOperator => matches!(
                action,
                ClusterControlActionV2::ReserveClusterControlCredentialRotation
                    | ClusterControlActionV2::ActivateClusterControlCredentialRotation
                    | ClusterControlActionV2::RevokeClusterControlCredential
                    | ClusterControlActionV2::ResolveClusterControlCredential
                    | ClusterControlActionV2::ListClusterBackups
                    | ClusterControlActionV2::GetClusterBackup
                    | ClusterControlActionV2::CreateClusterBackup
                    | ClusterControlActionV2::ResolveClusterBackup
                    | ClusterControlActionV2::DeleteClusterBackup
                    | ClusterControlActionV2::ResolveClusterBackupDeletion
                    | ClusterControlActionV2::GetClusterRestoreReadiness
                    | ClusterControlActionV2::PrepareOfflineRecovery
            ),
        }
    }
}

#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ClusterControlSessionV2 {
    pub format_version: u16,
    pub credential_kind: ClusterControlCredentialKindV1,
    pub credential: Vec<u8>,
    pub request_id: String,
    pub action: ClusterControlActionV2,
    pub request_correlation_id: [u8; 16],
}

impl fmt::Debug for ClusterControlSessionV2 {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ClusterControlSessionV2")
            .field("format_version", &self.format_version)
            .field("credential_kind", &self.credential_kind)
            .field("credential", &"[REDACTED]")
            .field("request_id", &self.request_id)
            .field("action", &self.action)
            .field("request_correlation_id", &self.request_correlation_id)
            .finish()
    }
}

impl ClusterControlSessionV2 {
    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
        if self.format_version != CLUSTER_CONTROL_SESSION_FORMAT_VERSION {
            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
        }
        validate_cluster_control_credential_v1(&self.credential)?;
        if self.request_id.is_empty()
            || self.request_id.len() > MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES
            || self.request_id.bytes().any(|byte| byte.is_ascii_control())
        {
            return Err(ClusterControlWireError::InvalidRequestIdentity);
        }
        if !self.credential_kind.admits(self.action) {
            return Err(ClusterControlWireError::ActionDenied);
        }
        Ok(())
    }
}

pub fn validate_cluster_control_credential_v1(credential: &[u8]) -> Result<(), ClusterControlWireError> {
    if credential.len() != CLUSTER_CONTROL_CREDENTIAL_ENCODED_BYTES
        || !credential.starts_with(CLUSTER_CONTROL_CREDENTIAL_PREFIX)
        || !credential[CLUSTER_CONTROL_CREDENTIAL_PREFIX.len()..]
            .iter()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
    {
        return Err(ClusterControlWireError::InvalidCredentialShape);
    }
    Ok(())
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ClusterControlRequestV2 {
    pub format_version: u16,
    pub session: ClusterControlSessionV2,
    pub body: serde_json::Value,
}

impl ClusterControlRequestV2 {
    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
        if self.format_version != CLUSTER_CONTROL_REQUEST_FORMAT_VERSION {
            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
        }
        self.session.validate()?;
        if !self.body.is_object() {
            return Err(ClusterControlWireError::InvalidBody);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ClusterControlResponseDispositionV2 {
    RetrySameRequest,
    ResolveOperation,
    CorrectRequest,
    Reauthenticate,
    Terminal,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "outcome", rename_all = "snake_case", deny_unknown_fields)]
pub enum ClusterControlResponseOutcomeV2 {
    Succeeded {
        body: serde_json::Value,
    },
    Refused {
        code: String,
        disposition: ClusterControlResponseDispositionV2,
    },
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ClusterControlResponseV2 {
    pub format_version: u16,
    pub request_id: String,
    pub request_correlation_id: [u8; 16],
    pub outcome: ClusterControlResponseOutcomeV2,
}

impl ClusterControlResponseV2 {
    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
        if self.format_version != CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION {
            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
        }
        validate_request_id(&self.request_id)?;
        match &self.outcome {
            ClusterControlResponseOutcomeV2::Succeeded { body } if body.is_object() => Ok(()),
            ClusterControlResponseOutcomeV2::Succeeded { .. } => Err(ClusterControlWireError::InvalidBody),
            ClusterControlResponseOutcomeV2::Refused { code, .. }
                if !code.is_empty()
                    && code.len() <= MAX_CLUSTER_CONTROL_ERROR_CODE_BYTES
                    && code
                        .bytes()
                        .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_') =>
            {
                Ok(())
            }
            ClusterControlResponseOutcomeV2::Refused { .. } => Err(ClusterControlWireError::InvalidErrorCode),
        }
    }
}

pub fn encode_cluster_control_request_v2(
    request: &ClusterControlRequestV2,
) -> Result<Vec<u8>, ClusterControlWireError> {
    request.validate()?;
    encode_frame(request)
}

pub fn decode_cluster_control_request_v2(bytes: &[u8]) -> Result<ClusterControlRequestV2, ClusterControlWireError> {
    let request: ClusterControlRequestV2 = decode_frame(bytes)?;
    request.validate()?;
    Ok(request)
}

pub fn encode_cluster_control_response_v2(
    response: &ClusterControlResponseV2,
) -> Result<Vec<u8>, ClusterControlWireError> {
    response.validate()?;
    encode_frame(response)
}

pub fn decode_cluster_control_response_v2(bytes: &[u8]) -> Result<ClusterControlResponseV2, ClusterControlWireError> {
    let response: ClusterControlResponseV2 = decode_frame(bytes)?;
    response.validate()?;
    Ok(response)
}

fn encode_frame(value: &impl Serialize) -> Result<Vec<u8>, ClusterControlWireError> {
    let bytes = serde_json::to_vec(value).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_FRAME_BYTES {
        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
    }
    Ok(bytes)
}

fn decode_frame<T: for<'de> Deserialize<'de> + Serialize>(bytes: &[u8]) -> Result<T, ClusterControlWireError> {
    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_FRAME_BYTES {
        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
    }
    let value = serde_json::from_slice(bytes).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
    if serde_json::to_vec(&value).map_err(|error| ClusterControlWireError::Codec(error.to_string()))? != bytes {
        return Err(ClusterControlWireError::NonCanonical);
    }
    Ok(value)
}

fn validate_request_id(request_id: &str) -> Result<(), ClusterControlWireError> {
    if request_id.is_empty()
        || request_id.len() > MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES
        || request_id.bytes().any(|byte| byte.is_ascii_control())
    {
        return Err(ClusterControlWireError::InvalidRequestIdentity);
    }
    Ok(())
}

pub fn encode_cluster_control_session_v2(
    session: &ClusterControlSessionV2,
) -> Result<Vec<u8>, ClusterControlWireError> {
    session.validate()?;
    let bytes = serde_json::to_vec(session).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
    if bytes.len() > MAX_CLUSTER_CONTROL_SESSION_BYTES {
        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
    }
    Ok(bytes)
}

pub fn decode_cluster_control_session_v2(bytes: &[u8]) -> Result<ClusterControlSessionV2, ClusterControlWireError> {
    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_SESSION_BYTES {
        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
    }
    let session: ClusterControlSessionV2 =
        serde_json::from_slice(bytes).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
    session.validate()?;
    if serde_json::to_vec(&session).map_err(|error| ClusterControlWireError::Codec(error.to_string()))? != bytes {
        return Err(ClusterControlWireError::NonCanonical);
    }
    Ok(session)
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ClusterControlWireError {
    UnsupportedVersion(u16),
    InvalidCredentialShape,
    InvalidRequestIdentity,
    InvalidBody,
    InvalidErrorCode,
    ActionDenied,
    FrameTooLarge(usize),
    NonCanonical,
    Codec(String),
}

impl Display for ClusterControlWireError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnsupportedVersion(version) => {
                write!(formatter, "unsupported cluster-control wire version {version}")
            }
            Self::InvalidCredentialShape => formatter.write_str("invalid cluster-control credential shape"),
            Self::InvalidRequestIdentity => formatter.write_str("invalid cluster-control request identity"),
            Self::InvalidBody => formatter.write_str("invalid cluster-control request or response body"),
            Self::InvalidErrorCode => formatter.write_str("invalid cluster-control error code"),
            Self::ActionDenied => formatter.write_str("cluster-control credential kind does not admit the action"),
            Self::FrameTooLarge(size) => write!(formatter, "cluster-control frame size {size} exceeds the bound"),
            Self::NonCanonical => formatter.write_str("cluster-control frame is not canonical"),
            Self::Codec(error) => write!(formatter, "cluster-control frame codec failed: {error}"),
        }
    }
}

impl Error for ClusterControlWireError {}

#[cfg(test)]
mod tests {
    use super::*;

    fn session(
        credential_kind: ClusterControlCredentialKindV1,
        action: ClusterControlActionV2,
    ) -> ClusterControlSessionV2 {
        ClusterControlSessionV2 {
            format_version: CLUSTER_CONTROL_SESSION_FORMAT_VERSION,
            credential_kind,
            credential: b"lcc1_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".to_vec(),
            request_id: "capacity-sample-1".to_owned(),
            action,
            request_correlation_id: [7; 16],
        }
    }

    #[test]
    fn credential_namespaces_are_disjoint_and_canonical() -> Result<(), Box<dyn Error>> {
        let observer = session(
            ClusterControlCredentialKindV1::CapacityObserver,
            ClusterControlActionV2::ObserveClusterCapacity,
        );
        let bytes = encode_cluster_control_session_v2(&observer)?;
        assert_eq!(decode_cluster_control_session_v2(&bytes)?, observer);

        let provisioner_backup = session(
            ClusterControlCredentialKindV1::CollectionProvisioner,
            ClusterControlActionV2::GetClusterBackup,
        );
        assert_eq!(
            provisioner_backup.validate(),
            Err(ClusterControlWireError::ActionDenied)
        );

        let observer_mutation = session(
            ClusterControlCredentialKindV1::CapacityObserver,
            ClusterControlActionV2::InitializeCollectionAuthority,
        );
        assert_eq!(observer_mutation.validate(), Err(ClusterControlWireError::ActionDenied));

        let recovery_capacity = session(
            ClusterControlCredentialKindV1::RecoveryOperator,
            ClusterControlActionV2::ObserveClusterCapacity,
        );
        assert_eq!(recovery_capacity.validate(), Err(ClusterControlWireError::ActionDenied));

        let recovery_delete = session(
            ClusterControlCredentialKindV1::RecoveryOperator,
            ClusterControlActionV2::DeleteClusterBackup,
        );
        assert!(recovery_delete.validate().is_ok());
        let provisioner_delete = session(
            ClusterControlCredentialKindV1::CollectionProvisioner,
            ClusterControlActionV2::DeleteClusterBackup,
        );
        assert_eq!(
            provisioner_delete.validate(),
            Err(ClusterControlWireError::ActionDenied)
        );

        let mut public_shape = observer;
        public_shape.credential = b"1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".to_vec();
        assert_eq!(
            public_shape.validate(),
            Err(ClusterControlWireError::InvalidCredentialShape)
        );
        Ok(())
    }

    #[test]
    fn decoder_refuses_predecessor_newer_unknown_and_malformed_frames() -> Result<(), Box<dyn Error>> {
        let observer = session(
            ClusterControlCredentialKindV1::CapacityObserver,
            ClusterControlActionV2::ObserveClusterCapacity,
        );
        let predecessor = String::from_utf8(encode_cluster_control_session_v2(&observer)?)?.replacen(
            &format!("\"format_version\":{CLUSTER_CONTROL_SESSION_FORMAT_VERSION}"),
            &format!("\"format_version\":{}", CLUSTER_CONTROL_SESSION_FORMAT_VERSION - 1),
            1,
        );
        assert_eq!(
            decode_cluster_control_session_v2(predecessor.as_bytes()),
            Err(ClusterControlWireError::UnsupportedVersion(1))
        );
        let mut newer = serde_json::to_value(&observer)?;
        newer["format_version"] = serde_json::json!(CLUSTER_CONTROL_SESSION_FORMAT_VERSION + 1);
        assert!(matches!(
            decode_cluster_control_session_v2(&serde_json::to_vec(&newer)?),
            Err(ClusterControlWireError::UnsupportedVersion(_))
        ));

        let mut unknown = serde_json::to_value(observer)?;
        unknown["collection_id"] = serde_json::json!(1);
        assert!(decode_cluster_control_session_v2(&serde_json::to_vec(&unknown)?).is_err());

        for malformed in [
            Vec::new(),
            vec![0xff],
            vec![b'x'; MAX_CLUSTER_CONTROL_SESSION_BYTES + 1],
        ] {
            assert!(decode_cluster_control_session_v2(&malformed).is_err());
        }
        Ok(())
    }

    #[test]
    fn request_and_response_envelopes_are_strict_bounded_and_correlated() -> Result<(), Box<dyn Error>> {
        let request = ClusterControlRequestV2 {
            format_version: CLUSTER_CONTROL_REQUEST_FORMAT_VERSION,
            session: session(
                ClusterControlCredentialKindV1::CapacityObserver,
                ClusterControlActionV2::ObserveClusterCapacity,
            ),
            body: serde_json::json!({}),
        };
        let encoded = encode_cluster_control_request_v2(&request)?;
        assert_eq!(decode_cluster_control_request_v2(&encoded)?, request);

        let predecessor_request = String::from_utf8(encoded.clone())?.replacen(
            &format!("\"format_version\":{CLUSTER_CONTROL_REQUEST_FORMAT_VERSION}"),
            &format!("\"format_version\":{}", CLUSTER_CONTROL_REQUEST_FORMAT_VERSION - 1),
            1,
        );
        assert_eq!(
            decode_cluster_control_request_v2(predecessor_request.as_bytes()),
            Err(ClusterControlWireError::UnsupportedVersion(1))
        );

        let response = ClusterControlResponseV2 {
            format_version: CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION,
            request_id: request.session.request_id.clone(),
            request_correlation_id: request.session.request_correlation_id,
            outcome: ClusterControlResponseOutcomeV2::Succeeded {
                body: serde_json::json!({"observation_id": "sample-1"}),
            },
        };
        let encoded = encode_cluster_control_response_v2(&response)?;
        assert_eq!(decode_cluster_control_response_v2(&encoded)?, response);

        let predecessor_response = String::from_utf8(encoded)?.replacen(
            &format!("\"format_version\":{CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION}"),
            &format!("\"format_version\":{}", CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION - 1),
            1,
        );
        assert_eq!(
            decode_cluster_control_response_v2(predecessor_response.as_bytes()),
            Err(ClusterControlWireError::UnsupportedVersion(1))
        );

        let mut unknown = serde_json::to_value(&request)?;
        unknown["unexpected"] = serde_json::json!(true);
        assert!(decode_cluster_control_request_v2(&serde_json::to_vec(&unknown)?).is_err());
        assert!(matches!(
            decode_cluster_control_request_v2(&vec![b'x'; MAX_CLUSTER_CONTROL_FRAME_BYTES + 1]),
            Err(ClusterControlWireError::FrameTooLarge(_))
        ));
        Ok(())
    }
}