lodestone-proto 0.0.16

//! Strict current-format Gateway -> Metastore Core API authority wire.
//!
//! The internal transport authenticates and bounds the workload edge before this
//! codec runs. This application envelope carries only public Lodestone API DTOs
//! plus the authenticated collection identity needed for authority-side
//! reauthorization; it never carries a metastore command or Ready capability.

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

use lodestone_types::{ApiKeyId, AuthorizationBindingV1, CollectionId, PrincipalId, UnixMillis, WorkloadClassV1};
use serde::{Deserialize, Serialize};

use crate::{
    Command, ErrorBody, RPC_MAX_CONTROL_PAYLOAD_BYTES, RequestBody, ResultBody, command_for_request, command_for_result,
};

pub const CORE_API_AUTHORITY_FORMAT_VERSION: u16 = 5;
pub const MAX_CORE_API_AUTHORITY_BYTES: usize = RPC_MAX_CONTROL_PAYLOAD_BYTES;

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CoreApiAuthorityContextV2 {
    pub collection_id: CollectionId,
    pub principal_id: PrincipalId,
    pub api_key_id: ApiKeyId,
    pub authorization_binding: AuthorizationBindingV1,
    pub workload_class: WorkloadClassV1,
    pub request_correlation_id: [u8; 16],
    pub response_diagnostic_id: [u8; 16],
    pub absolute_deadline_unix_millis: UnixMillis,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CoreApiAuthorityRequestV2 {
    pub format_version: u16,
    pub context: CoreApiAuthorityContextV2,
    pub request: RequestBody,
}

impl CoreApiAuthorityRequestV2 {
    pub fn current(
        context: CoreApiAuthorityContextV2,
        request: RequestBody,
    ) -> Result<Self, CoreApiAuthorityWireError> {
        let value = Self {
            format_version: CORE_API_AUTHORITY_FORMAT_VERSION,
            context,
            request,
        };
        value.validate()?;
        Ok(value)
    }

    pub fn validate(&self) -> Result<(), CoreApiAuthorityWireError> {
        if self.format_version != CORE_API_AUTHORITY_FORMAT_VERSION {
            return Err(CoreApiAuthorityWireError::UnsupportedVersion(self.format_version));
        }
        if self.context.request_correlation_id == [0; 16]
            || self.context.response_diagnostic_id == [0; 16]
            || self.context.absolute_deadline_unix_millis.get() == 0
            || !self.context.authorization_binding.is_valid()
            || self.context.collection_id != self.context.authorization_binding.collection_id
            || self.context.principal_id != self.context.authorization_binding.principal_id
            || self.context.api_key_id != self.context.authorization_binding.api_key_id
        {
            return Err(CoreApiAuthorityWireError::InvalidContext);
        }
        let command = command_for_request(&self.request);
        if !core_api_authority_command_is_admitted(command) {
            return Err(CoreApiAuthorityWireError::CommandNotAdmitted(command));
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "outcome", content = "body", rename_all = "snake_case", deny_unknown_fields)]
pub enum CoreApiAuthorityOutcomeV2 {
    Success(Box<ResultBody>),
    Error(ErrorBody),
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CoreApiAuthorityResponseV2 {
    pub format_version: u16,
    pub outcome: CoreApiAuthorityOutcomeV2,
}

impl CoreApiAuthorityResponseV2 {
    pub fn success(request: &CoreApiAuthorityRequestV2, result: ResultBody) -> Result<Self, CoreApiAuthorityWireError> {
        if command_for_result(&result) != command_for_request(&request.request) {
            return Err(CoreApiAuthorityWireError::SubstitutedResult);
        }
        Ok(Self {
            format_version: CORE_API_AUTHORITY_FORMAT_VERSION,
            outcome: CoreApiAuthorityOutcomeV2::Success(Box::new(result)),
        })
    }

    #[must_use]
    pub fn error(error: ErrorBody) -> Self {
        Self {
            format_version: CORE_API_AUTHORITY_FORMAT_VERSION,
            outcome: CoreApiAuthorityOutcomeV2::Error(error),
        }
    }

    pub fn validate_for(&self, request: &CoreApiAuthorityRequestV2) -> Result<(), CoreApiAuthorityWireError> {
        if self.format_version != CORE_API_AUTHORITY_FORMAT_VERSION {
            return Err(CoreApiAuthorityWireError::UnsupportedVersion(self.format_version));
        }
        if let CoreApiAuthorityOutcomeV2::Success(result) = &self.outcome
            && command_for_result(result) != command_for_request(&request.request)
        {
            return Err(CoreApiAuthorityWireError::SubstitutedResult);
        }
        Ok(())
    }
}

pub fn encode_core_api_authority_request(
    request: &CoreApiAuthorityRequestV2,
) -> Result<Vec<u8>, CoreApiAuthorityWireError> {
    request.validate()?;
    encode_canonical(request)
}

pub fn decode_core_api_authority_request(bytes: &[u8]) -> Result<CoreApiAuthorityRequestV2, CoreApiAuthorityWireError> {
    let request: CoreApiAuthorityRequestV2 = decode_bounded(bytes)?;
    request.validate()?;
    require_canonical(&request, bytes)?;
    Ok(request)
}

pub fn encode_core_api_authority_response(
    request: &CoreApiAuthorityRequestV2,
    response: &CoreApiAuthorityResponseV2,
) -> Result<Vec<u8>, CoreApiAuthorityWireError> {
    response.validate_for(request)?;
    encode_canonical(response)
}

pub fn decode_core_api_authority_response(
    request: &CoreApiAuthorityRequestV2,
    bytes: &[u8],
) -> Result<CoreApiAuthorityResponseV2, CoreApiAuthorityWireError> {
    let response: CoreApiAuthorityResponseV2 = decode_bounded(bytes)?;
    response.validate_for(request)?;
    require_canonical(&response, bytes)?;
    Ok(response)
}

fn encode_canonical<T: Serialize>(value: &T) -> Result<Vec<u8>, CoreApiAuthorityWireError> {
    let bytes = serde_json::to_vec(value).map_err(|_| CoreApiAuthorityWireError::Codec)?;
    if bytes.len() > MAX_CORE_API_AUTHORITY_BYTES {
        return Err(CoreApiAuthorityWireError::TooLarge(bytes.len()));
    }
    Ok(bytes)
}

fn decode_bounded<T>(bytes: &[u8]) -> Result<T, CoreApiAuthorityWireError>
where
    T: for<'de> Deserialize<'de>,
{
    if bytes.is_empty() || bytes.len() > MAX_CORE_API_AUTHORITY_BYTES {
        return Err(CoreApiAuthorityWireError::TooLarge(bytes.len()));
    }
    serde_json::from_slice(bytes).map_err(|_| CoreApiAuthorityWireError::Codec)
}

fn require_canonical<T: Serialize>(value: &T, bytes: &[u8]) -> Result<(), CoreApiAuthorityWireError> {
    let canonical = serde_json::to_vec(&value).map_err(|_| CoreApiAuthorityWireError::Codec)?;
    if canonical != bytes {
        return Err(CoreApiAuthorityWireError::NonCanonical);
    }
    Ok(())
}

#[must_use]
pub const fn core_api_authority_command_is_admitted(command: Command) -> bool {
    !matches!(
        command,
        Command::Discover
            | Command::SubmitDocumentMutation
            | Command::GetDocument
            | Command::Search
            | Command::OpenPit
            | Command::RenewPit
            | Command::ClosePit
    )
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CoreApiAuthorityWireError {
    UnsupportedVersion(u16),
    InvalidContext,
    CommandNotAdmitted(Command),
    SubstitutedResult,
    TooLarge(usize),
    Codec,
    NonCanonical,
}

impl Display for CoreApiAuthorityWireError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        write!(formatter, "Core API authority wire refusal: {self:?}")
    }
}

impl Error for CoreApiAuthorityWireError {}

#[cfg(test)]
mod tests {
    use std::error::Error;

    use lodestone_types::{
        AUTHORIZATION_BINDING_FORMAT_VERSION, ApiKeyGeneration, ApiKeyId, CollectionId, CollectionSecurityGeneration,
        PrincipalId, QuotaPolicyGeneration, RecoveryEra, SecurityPolicyRevision, SecurityPrincipalGeneration,
        UnixMillis,
    };

    use super::*;
    use crate::{GetQuotaRequest, GetQuotaResult, PrepareDocumentMutationRequest, RawJson, Resource, U64};

    fn context() -> Result<CoreApiAuthorityContextV2, Box<dyn Error>> {
        Ok(CoreApiAuthorityContextV2 {
            collection_id: CollectionId::new(1)?,
            principal_id: PrincipalId::new(2),
            api_key_id: ApiKeyId::new(3),
            authorization_binding: AuthorizationBindingV1 {
                format_version: AUTHORIZATION_BINDING_FORMAT_VERSION,
                recovery_era: RecoveryEra::new(1)?,
                public_security_root_digest: [7; 32],
                ordering_revision: SecurityPolicyRevision::new(4),
                collection_id: CollectionId::new(1)?,
                collection_security_generation: CollectionSecurityGeneration::new(5),
                principal_id: PrincipalId::new(2),
                principal_generation: SecurityPrincipalGeneration::new(6),
                api_key_id: ApiKeyId::new(3),
                api_key_generation: ApiKeyGeneration::new(7),
                role_generations_digest: [8; 32],
                quota_policy_generation: QuotaPolicyGeneration::new(9),
                authorized_scope_digest: [10; 32],
            },
            workload_class: WorkloadClassV1::BatchAnalytics,
            request_correlation_id: [5; 16],
            response_diagnostic_id: [6; 16],
            absolute_deadline_unix_millis: UnixMillis::new(6),
        })
    }

    #[test]
    fn control_request_and_exact_result_are_canonical() -> Result<(), Box<dyn Error>> {
        let request = CoreApiAuthorityRequestV2::current(context()?, RequestBody::GetQuota(GetQuotaRequest {}))?;
        let bytes = encode_core_api_authority_request(&request)?;
        assert_eq!(decode_core_api_authority_request(&bytes)?, request);
        let response = CoreApiAuthorityResponseV2::success(
            &request,
            ResultBody::GetQuota(GetQuotaResult {
                revision: U64 { value: "4".to_owned() },
                limits: Vec::new(),
                usage: Vec::new(),
            }),
        )?;
        let encoded = encode_core_api_authority_response(&request, &response)?;
        assert_eq!(decode_core_api_authority_response(&request, &encoded)?, response);
        Ok(())
    }

    #[test]
    fn recoverable_prepare_newer_unknown_noncanonical_and_substitution_fail_closed() -> Result<(), Box<dyn Error>> {
        let data = CoreApiAuthorityRequestV2::current(
            context()?,
            RequestBody::PrepareDocumentMutation(PrepareDocumentMutationRequest {
                index: Resource {
                    kind: "index".to_owned(),
                    name: "logs".to_owned(),
                    incarnation: crate::Identity { value: "01".repeat(16) },
                },
                document_id: "1".to_owned(),
                operation: "put".to_owned(),
                source: Some(RawJson {
                    bytes_base64: "e30=".to_owned(),
                }),
                routing: None,
                condition: None,
                pipeline: None,
            }),
        );
        assert!(data.is_ok());
        assert!(!core_api_authority_command_is_admitted(Command::SubmitDocumentMutation));

        let valid = CoreApiAuthorityRequestV2::current(context()?, RequestBody::GetQuota(GetQuotaRequest {}))?;
        let mut newer = serde_json::to_value(&valid)?;
        newer["format_version"] = serde_json::json!(CORE_API_AUTHORITY_FORMAT_VERSION + 1);
        let newer = serde_json::to_vec(&newer)?;
        assert_eq!(
            decode_core_api_authority_request(&newer),
            Err(CoreApiAuthorityWireError::UnsupportedVersion(
                CORE_API_AUTHORITY_FORMAT_VERSION + 1
            ))
        );
        let with_unknown = serde_json::json!({
            "format_version": CORE_API_AUTHORITY_FORMAT_VERSION,
            "context": valid.context,
            "request": valid.request,
            "unknown": true
        });
        assert!(matches!(
            decode_core_api_authority_request(&serde_json::to_vec(&with_unknown)?),
            Err(CoreApiAuthorityWireError::Codec)
        ));
        let mut noncanonical = encode_core_api_authority_request(&valid)?;
        noncanonical.push(b' ');
        assert!(matches!(
            decode_core_api_authority_request(&noncanonical),
            Err(CoreApiAuthorityWireError::NonCanonical)
        ));
        assert!(matches!(
            CoreApiAuthorityResponseV2::success(
                &valid,
                ResultBody::GetRuntimeStatus(crate::GetRuntimeStatusResult {
                    ready: true,
                    recovery_era: crate::Identity { value: "01".repeat(16) },
                    catalog_revision: U64 { value: "1".to_owned() },
                    policy_revision: U64 { value: "1".to_owned() },
                    redacted: crate::RuntimeStatusDetail {
                        metastore_revision: U64 { value: "1".to_owned() },
                        root_revision: U64 { value: "1".to_owned() },
                        logical_time: U64 { value: "1".to_owned() },
                        degraded_dependencies: Vec::new(),
                    },
                })
            ),
            Err(CoreApiAuthorityWireError::SubstitutedResult)
        ));
        Ok(())
    }
}