lodestone-proto 0.0.16

//! Authenticated RPC negotiation and authorization-before-allocation.
//!
//! The TLS implementation is separately owned. This module freezes the bounded
//! wire/authentication profile consumed by that listener. The authenticated
//! [`WorkloadIdentity`] is channel-derived; caller-controlled payload fields
//! can never select a role or node incarnation.

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

use lodestone_types::{ClusterStorageId, CredentialGeneration, NodeRole, NodeToken};
use serde::{Deserialize, Serialize};

pub const WORKLOAD_IDENTITY_FORMAT_VERSION: u16 = 1;
pub const RPC_PRELUDE_FORMAT_VERSION: u16 = 1;
pub const RPC_AUTH_PROFILE_V1: u16 = 1;
pub const RPC_WIRE_VERSION_MIN: u16 = 1;
pub const RPC_WIRE_VERSION_MAX: u16 = 1;
pub const RPC_PRELUDE_BYTES: usize = 24;
pub const RPC_MAX_PAYLOAD_BYTES: usize = 16 * 1024 * 1024;
pub const RPC_MAX_CONTROL_PAYLOAD_BYTES: usize = 2 * 1024 * 1024;
pub const RPC_SEARCH_REQUEST_FRAME_BYTES: usize = RPC_MAX_PAYLOAD_BYTES;
pub const RPC_METASTORE_SNAPSHOT_REQUEST_FRAME_BYTES: usize = 128;
pub const RPC_METASTORE_SNAPSHOT_RESPONSE_FRAME_BYTES: usize = 4 * 1024;
pub const RPC_SEARCH_RESPONSE_FRAME_BYTES: usize = RPC_MAX_PAYLOAD_BYTES;
const RPC_PRELUDE_MAGIC: &[u8; 8] = b"LDSP6RPC";

/// Exact authenticated claims yielded by the transport channel.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct WorkloadIdentity {
    pub format_version: u16,
    pub cluster_id: ClusterStorageId,
    pub node_token: NodeToken,
    pub node_role: NodeRole,
    pub credential_generation: CredentialGeneration,
    pub credential_binding_digest: [u8; 32],
}

impl WorkloadIdentity {
    pub fn validate(&self) -> Result<(), RpcAuthorizationError> {
        if self.format_version != WORKLOAD_IDENTITY_FORMAT_VERSION {
            return Err(RpcAuthorizationError::UnsupportedIdentityVersion(self.format_version));
        }
        if self.credential_generation.get() == 0 {
            return Err(RpcAuthorizationError::CredentialMismatch);
        }
        if self.credential_binding_digest == [0; 32] {
            return Err(RpcAuthorizationError::CredentialMismatch);
        }
        Ok(())
    }
}

/// Negotiated RPC wire offer. Identity and authorization profiles negotiate
/// independently; every current runtime version requires auth profile v1.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RpcVersionOffer {
    pub minimum_wire_version: u16,
    pub maximum_wire_version: u16,
    pub authorization_profile: u16,
}

impl RpcVersionOffer {
    pub fn v1() -> Self {
        Self {
            minimum_wire_version: RPC_WIRE_VERSION_MIN,
            maximum_wire_version: RPC_WIRE_VERSION_MAX,
            authorization_profile: RPC_AUTH_PROFILE_V1,
        }
    }

    pub fn negotiate(self, peer: Self) -> Result<RpcNegotiatedProfile, RpcAuthorizationError> {
        if self.authorization_profile != RPC_AUTH_PROFILE_V1 || peer.authorization_profile != RPC_AUTH_PROFILE_V1 {
            return Err(RpcAuthorizationError::IncompatibleAuthProfile {
                offered: peer.authorization_profile,
            });
        }
        if self.minimum_wire_version == 0
            || peer.minimum_wire_version == 0
            || self.minimum_wire_version > self.maximum_wire_version
            || peer.minimum_wire_version > peer.maximum_wire_version
        {
            return Err(RpcAuthorizationError::IncompatibleWireVersion);
        }
        let minimum = self.minimum_wire_version.max(peer.minimum_wire_version);
        let maximum = self.maximum_wire_version.min(peer.maximum_wire_version);
        if minimum > maximum {
            return Err(RpcAuthorizationError::IncompatibleWireVersion);
        }
        Ok(RpcNegotiatedProfile {
            wire_version: maximum,
            authorization_profile: RPC_AUTH_PROFILE_V1,
        })
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RpcNegotiatedProfile {
    pub wire_version: u16,
    pub authorization_profile: u16,
}

/// Method identity available from the fixed prelude before payload work.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RpcMethod {
    RaftVote,
    RaftAppend,
    RaftInstallSnapshot,
    MetastoreCommand,
    MetastoreSnapshot,
    MetastorePointRead,
    MetastoreWatch,
    NodeSelfHealth,
    OrchestratorPlacementIntent,
    OrchestratorCatalogIntent,
    ProposePlacementRecovery,
    WriterIngest,
    WriterRealtimeRead,
    WriterDrainAdvisory,
    TaskDestinationMutation,
    SearchCoordinate,
    SearchShardExecute,
    SearchMount,
    InternalHealth,
    CompactorExecuteTask,
    CoreApiAuthority,
}

impl RpcMethod {
    const fn to_wire(self) -> u8 {
        match self {
            Self::RaftVote => 1,
            Self::RaftAppend => 2,
            Self::RaftInstallSnapshot => 3,
            Self::MetastoreCommand => 4,
            Self::MetastoreSnapshot => 5,
            Self::MetastorePointRead => 6,
            Self::MetastoreWatch => 7,
            Self::NodeSelfHealth => 8,
            Self::OrchestratorPlacementIntent => 9,
            Self::OrchestratorCatalogIntent => 10,
            Self::ProposePlacementRecovery => 11,
            Self::WriterIngest => 12,
            Self::WriterRealtimeRead => 13,
            Self::WriterDrainAdvisory => 14,
            Self::TaskDestinationMutation => 15,
            Self::SearchCoordinate => 16,
            Self::SearchShardExecute => 17,
            Self::SearchMount => 18,
            Self::InternalHealth => 19,
            Self::CompactorExecuteTask => 20,
            Self::CoreApiAuthority => 21,
        }
    }

    fn from_wire(value: u8) -> Result<Self, RpcAuthorizationError> {
        match value {
            1 => Ok(Self::RaftVote),
            2 => Ok(Self::RaftAppend),
            3 => Ok(Self::RaftInstallSnapshot),
            4 => Ok(Self::MetastoreCommand),
            5 => Ok(Self::MetastoreSnapshot),
            6 => Ok(Self::MetastorePointRead),
            7 => Ok(Self::MetastoreWatch),
            8 => Ok(Self::NodeSelfHealth),
            9 => Ok(Self::OrchestratorPlacementIntent),
            10 => Ok(Self::OrchestratorCatalogIntent),
            11 => Ok(Self::ProposePlacementRecovery),
            12 => Ok(Self::WriterIngest),
            13 => Ok(Self::WriterRealtimeRead),
            14 => Ok(Self::WriterDrainAdvisory),
            15 => Ok(Self::TaskDestinationMutation),
            16 => Ok(Self::SearchCoordinate),
            17 => Ok(Self::SearchShardExecute),
            18 => Ok(Self::SearchMount),
            19 => Ok(Self::InternalHealth),
            20 => Ok(Self::CompactorExecuteTask),
            21 => Ok(Self::CoreApiAuthority),
            _ => Err(RpcAuthorizationError::UnknownMethod(value)),
        }
    }

    pub const fn payload_limit(self) -> usize {
        match self {
            Self::RaftInstallSnapshot => RPC_MAX_PAYLOAD_BYTES,
            Self::MetastoreSnapshot => RPC_METASTORE_SNAPSHOT_REQUEST_FRAME_BYTES,
            Self::SearchShardExecute => RPC_SEARCH_REQUEST_FRAME_BYTES,
            _ => RPC_MAX_CONTROL_PAYLOAD_BYTES,
        }
    }
}

/// Bounded discriminant decoded and authorized before a metastore command
/// body. A generic metastore method is never a generic permission.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum MetastoreCommandFamily {
    SelfNode,
    RaftMembership,
    PlacementRecoveryVote,
    PlacementClaim,
    WriterLease,
    SearchAssignment,
    NodeLifecycle,
    SplitMerge,
    SystemTaskEnqueue,
    WalRegistration,
    ManifestPublication,
    SearcherMount,
    TaskLease,
    PitLifecycle,
    Catalog,
    Security,
}

impl MetastoreCommandFamily {
    const fn to_wire(self) -> u8 {
        match self {
            Self::SelfNode => 1,
            Self::RaftMembership => 2,
            Self::PlacementRecoveryVote => 3,
            Self::PlacementClaim => 4,
            Self::WriterLease => 5,
            Self::SearchAssignment => 6,
            Self::NodeLifecycle => 7,
            Self::SplitMerge => 8,
            Self::SystemTaskEnqueue => 9,
            Self::WalRegistration => 10,
            Self::ManifestPublication => 11,
            Self::SearcherMount => 12,
            Self::TaskLease => 13,
            Self::PitLifecycle => 14,
            Self::Catalog => 15,
            Self::Security => 16,
        }
    }

    fn from_wire(value: u8) -> Result<Option<Self>, RpcAuthorizationError> {
        match value {
            0 => Ok(None),
            1 => Ok(Some(Self::SelfNode)),
            2 => Ok(Some(Self::RaftMembership)),
            3 => Ok(Some(Self::PlacementRecoveryVote)),
            4 => Ok(Some(Self::PlacementClaim)),
            5 => Ok(Some(Self::WriterLease)),
            6 => Ok(Some(Self::SearchAssignment)),
            7 => Ok(Some(Self::NodeLifecycle)),
            8 => Ok(Some(Self::SplitMerge)),
            9 => Ok(Some(Self::SystemTaskEnqueue)),
            10 => Ok(Some(Self::WalRegistration)),
            11 => Ok(Some(Self::ManifestPublication)),
            12 => Ok(Some(Self::SearcherMount)),
            13 => Ok(Some(Self::TaskLease)),
            14 => Ok(Some(Self::PitLifecycle)),
            15 => Ok(Some(Self::Catalog)),
            16 => Ok(Some(Self::Security)),
            _ => Err(RpcAuthorizationError::UnknownCommandFamily(value)),
        }
    }
}

/// Fixed-size prelude decoded without allocating attacker-declared payload
/// bytes. The target role is also known from the listener and is exact-matched
/// before authorization.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct RpcPrelude {
    pub format_version: u16,
    pub authorization_profile: u16,
    pub minimum_wire_version: u16,
    pub maximum_wire_version: u16,
    pub target_role: NodeRole,
    pub method: RpcMethod,
    pub command_family: Option<MetastoreCommandFamily>,
    pub declared_payload_bytes: u32,
}

impl RpcPrelude {
    pub fn encode(self) -> Result<[u8; RPC_PRELUDE_BYTES], RpcAuthorizationError> {
        validate_method_family(self.method, self.command_family)?;
        let mut bytes = [0_u8; RPC_PRELUDE_BYTES];
        bytes[..8].copy_from_slice(RPC_PRELUDE_MAGIC);
        bytes[8..10].copy_from_slice(&self.format_version.to_le_bytes());
        bytes[10..12].copy_from_slice(&self.authorization_profile.to_le_bytes());
        bytes[12..14].copy_from_slice(&self.minimum_wire_version.to_le_bytes());
        bytes[14..16].copy_from_slice(&self.maximum_wire_version.to_le_bytes());
        bytes[16] = node_role_to_wire(self.target_role);
        bytes[17] = self.method.to_wire();
        bytes[18] = self.command_family.map_or(0, MetastoreCommandFamily::to_wire);
        bytes[19] = 0;
        bytes[20..24].copy_from_slice(&self.declared_payload_bytes.to_le_bytes());
        Ok(bytes)
    }

    pub fn decode(bytes: &[u8]) -> Result<Self, RpcAuthorizationError> {
        if bytes.len() != RPC_PRELUDE_BYTES {
            return Err(RpcAuthorizationError::PreludeLength { actual: bytes.len() });
        }
        if bytes.get(..8) != Some(RPC_PRELUDE_MAGIC.as_slice()) {
            return Err(RpcAuthorizationError::BadMagic);
        }
        let format_version = read_u16(bytes, 8)?;
        if format_version != RPC_PRELUDE_FORMAT_VERSION {
            return Err(RpcAuthorizationError::UnsupportedPreludeVersion(format_version));
        }
        if bytes[19] != 0 {
            return Err(RpcAuthorizationError::ReservedBits);
        }
        let method = RpcMethod::from_wire(bytes[17])?;
        let command_family = MetastoreCommandFamily::from_wire(bytes[18])?;
        validate_method_family(method, command_family)?;
        Ok(Self {
            format_version,
            authorization_profile: read_u16(bytes, 10)?,
            minimum_wire_version: read_u16(bytes, 12)?,
            maximum_wire_version: read_u16(bytes, 14)?,
            target_role: node_role_from_wire(bytes[16])?,
            method,
            command_family,
            declared_payload_bytes: read_u32(bytes, 20)?,
        })
    }
}

/// Authorizes one already decoded fixed prelude. ACL evaluation deliberately
/// precedes the payload length check so an oversized forbidden call is denied
/// without allocating or revealing a different path.
pub fn authorize_rpc(
    identity: &WorkloadIdentity,
    listener_role: NodeRole,
    prelude: RpcPrelude,
) -> Result<RpcNegotiatedProfile, RpcAuthorizationError> {
    identity.validate()?;
    if prelude.target_role != listener_role {
        return Err(RpcAuthorizationError::WrongRole);
    }
    let negotiated = RpcVersionOffer::v1().negotiate(RpcVersionOffer {
        minimum_wire_version: prelude.minimum_wire_version,
        maximum_wire_version: prelude.maximum_wire_version,
        authorization_profile: prelude.authorization_profile,
    })?;
    if !acl_allows(
        identity.node_role,
        listener_role,
        prelude.method,
        prelude.command_family,
    ) {
        return Err(RpcAuthorizationError::PermissionDenied);
    }
    let actual =
        usize::try_from(prelude.declared_payload_bytes).map_err(|_| RpcAuthorizationError::PayloadTooLarge {
            actual: usize::MAX,
            maximum: prelude.method.payload_limit(),
        })?;
    let maximum = prelude.method.payload_limit();
    if actual > maximum {
        return Err(RpcAuthorizationError::PayloadTooLarge { actual, maximum });
    }
    Ok(negotiated)
}

#[allow(
    clippy::match_like_matches_macro,
    reason = "the row-oriented deny-by-default ACL is reviewed and modeled as an explicit matrix"
)]
fn acl_allows(source: NodeRole, target: NodeRole, method: RpcMethod, family: Option<MetastoreCommandFamily>) -> bool {
    use MetastoreCommandFamily as Family;
    use NodeRole::{Compactor, Gateway, Indexer, MetastoreVoter, Orchestrator, Placement, Searcher};
    use RpcMethod as Method;

    match (source, target, method, family) {
        (MetastoreVoter, MetastoreVoter, Method::RaftVote | Method::RaftAppend | Method::RaftInstallSnapshot, None) => {
            true
        }
        (MetastoreVoter, MetastoreVoter, Method::MetastoreSnapshot, None) => true,
        (_, MetastoreVoter, Method::MetastorePointRead | Method::MetastoreWatch, None) => true,
        (_, _, Method::InternalHealth, None) => true,
        (_, MetastoreVoter, Method::NodeSelfHealth, None) => true,
        (_, MetastoreVoter, Method::MetastoreCommand, Some(Family::SelfNode)) => true,
        (
            MetastoreVoter,
            MetastoreVoter,
            Method::MetastoreCommand,
            Some(Family::RaftMembership | Family::PlacementRecoveryVote),
        ) => true,
        (
            Placement,
            MetastoreVoter,
            Method::MetastoreCommand,
            Some(
                Family::PlacementClaim
                | Family::WriterLease
                | Family::SearchAssignment
                | Family::NodeLifecycle
                | Family::SplitMerge
                | Family::SystemTaskEnqueue,
            ),
        ) => true,
        (
            Indexer,
            MetastoreVoter,
            Method::MetastoreCommand,
            Some(Family::WriterLease | Family::WalRegistration | Family::ManifestPublication | Family::SplitMerge),
        ) => true,
        (Searcher, MetastoreVoter, Method::MetastoreCommand, Some(Family::SearcherMount)) => true,
        (Compactor, MetastoreVoter, Method::MetastoreCommand, Some(Family::TaskLease | Family::SplitMerge)) => true,
        (
            Gateway,
            MetastoreVoter,
            Method::MetastoreCommand,
            Some(Family::PitLifecycle | Family::Catalog | Family::Security),
        ) => true,
        (Orchestrator, MetastoreVoter, Method::MetastoreCommand, Some(Family::Security)) => true,
        (Orchestrator, Placement, Method::OrchestratorPlacementIntent, None) => true,
        (Orchestrator, Gateway, Method::OrchestratorCatalogIntent, None) => true,
        (Orchestrator, MetastoreVoter, Method::ProposePlacementRecovery, None) => true,
        (Gateway, Indexer, Method::WriterIngest | Method::WriterRealtimeRead, None) => true,
        (Searcher, Indexer, Method::WriterRealtimeRead, None) => true,
        (Placement, Indexer, Method::WriterDrainAdvisory, None) => true,
        (Compactor, Indexer, Method::TaskDestinationMutation, None) => true,
        (Placement, Compactor, Method::CompactorExecuteTask, None) => true,
        (Gateway, MetastoreVoter, Method::CoreApiAuthority, None) => true,
        (Gateway, Searcher, Method::SearchCoordinate, None) => true,
        (Searcher, Searcher, Method::SearchShardExecute, None) => true,
        (Placement, Searcher, Method::SearchMount, None) => true,
        _ => false,
    }
}

fn validate_method_family(
    method: RpcMethod,
    family: Option<MetastoreCommandFamily>,
) -> Result<(), RpcAuthorizationError> {
    match (method, family) {
        (RpcMethod::MetastoreCommand, Some(_)) => Ok(()),
        (method, None) if method != RpcMethod::MetastoreCommand => Ok(()),
        _ => Err(RpcAuthorizationError::MalformedCommandFamily),
    }
}

fn node_role_to_wire(role: NodeRole) -> u8 {
    match role {
        NodeRole::MetastoreVoter => 1,
        NodeRole::Placement => 2,
        NodeRole::Orchestrator => 3,
        NodeRole::Gateway => 4,
        NodeRole::Indexer => 5,
        NodeRole::Searcher => 6,
        NodeRole::Compactor => 7,
    }
}

fn node_role_from_wire(value: u8) -> Result<NodeRole, RpcAuthorizationError> {
    match value {
        1 => Ok(NodeRole::MetastoreVoter),
        2 => Ok(NodeRole::Placement),
        3 => Ok(NodeRole::Orchestrator),
        4 => Ok(NodeRole::Gateway),
        5 => Ok(NodeRole::Indexer),
        6 => Ok(NodeRole::Searcher),
        7 => Ok(NodeRole::Compactor),
        _ => Err(RpcAuthorizationError::UnknownRole(value)),
    }
}

fn read_u16(bytes: &[u8], offset: usize) -> Result<u16, RpcAuthorizationError> {
    let value = bytes
        .get(offset..offset + 2)
        .ok_or(RpcAuthorizationError::PreludeLength { actual: bytes.len() })?;
    let mut array = [0; 2];
    array.copy_from_slice(value);
    Ok(u16::from_le_bytes(array))
}

fn read_u32(bytes: &[u8], offset: usize) -> Result<u32, RpcAuthorizationError> {
    let value = bytes
        .get(offset..offset + 4)
        .ok_or(RpcAuthorizationError::PreludeLength { actual: bytes.len() })?;
    let mut array = [0; 4];
    array.copy_from_slice(value);
    Ok(u32::from_le_bytes(array))
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum RpcAuthorizationError {
    BadMagic,
    CredentialMismatch,
    IncompatibleAuthProfile { offered: u16 },
    IncompatibleWireVersion,
    MalformedCommandFamily,
    PayloadTooLarge { actual: usize, maximum: usize },
    PermissionDenied,
    PreludeLength { actual: usize },
    ReservedBits,
    UnknownCommandFamily(u8),
    UnknownMethod(u8),
    UnknownRole(u8),
    UnsupportedIdentityVersion(u16),
    UnsupportedPreludeVersion(u16),
    WrongRole,
}

impl Display for RpcAuthorizationError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::BadMagic => write!(formatter, "RPC prelude magic is invalid"),
            Self::CredentialMismatch => write!(formatter, "workload credential binding is invalid"),
            Self::IncompatibleAuthProfile { offered } => {
                write!(formatter, "authorization profile {offered} is incompatible")
            }
            Self::IncompatibleWireVersion => write!(formatter, "RPC wire version ranges do not overlap"),
            Self::MalformedCommandFamily => write!(formatter, "RPC method and command family are inconsistent"),
            Self::PayloadTooLarge { actual, maximum } => {
                write!(formatter, "RPC payload is {actual} bytes; maximum is {maximum}")
            }
            Self::PermissionDenied => write!(formatter, "workload role is not permitted to invoke this method"),
            Self::PreludeLength { actual } => write!(formatter, "RPC prelude has {actual} bytes"),
            Self::ReservedBits => write!(formatter, "RPC prelude reserved bits are nonzero"),
            Self::UnknownCommandFamily(value) => write!(formatter, "unknown metastore command family {value}"),
            Self::UnknownMethod(value) => write!(formatter, "unknown RPC method {value}"),
            Self::UnknownRole(value) => write!(formatter, "unknown RPC role {value}"),
            Self::UnsupportedIdentityVersion(value) => {
                write!(formatter, "unsupported workload identity version {value}")
            }
            Self::UnsupportedPreludeVersion(value) => write!(formatter, "unsupported RPC prelude version {value}"),
            Self::WrongRole => write!(formatter, "RPC target role does not match the listener"),
        }
    }
}

impl Error for RpcAuthorizationError {}

#[cfg(test)]
mod tests {
    use lodestone_types::{
        ClusterIdentity, CredentialGeneration, NodeId, NodeIncarnation, NodeRole, NodeToken, RecoveryEra,
        RecoveryGeneration,
    };

    use super::{
        MetastoreCommandFamily, RPC_METASTORE_SNAPSHOT_REQUEST_FRAME_BYTES, RpcAuthorizationError, RpcMethod,
        RpcPrelude, RpcVersionOffer, WorkloadIdentity, authorize_rpc,
    };

    fn identity(role: NodeRole) -> Result<WorkloadIdentity, Box<dyn std::error::Error>> {
        let cluster = ClusterIdentity::new(RecoveryEra::new(1)?, RecoveryGeneration::genesis());
        Ok(WorkloadIdentity {
            format_version: 1,
            cluster_id: lodestone_types::ClusterStorageId::derive_from_genesis(cluster)?,
            node_token: NodeToken::new(RecoveryEra::new(1)?, NodeId::new(7), NodeIncarnation::new(1)?)?,
            node_role: role,
            credential_generation: CredentialGeneration::new(1),
            credential_binding_digest: [9; 32],
        })
    }

    fn prelude(target_role: NodeRole, method: RpcMethod) -> RpcPrelude {
        RpcPrelude {
            format_version: 1,
            authorization_profile: 1,
            minimum_wire_version: 1,
            maximum_wire_version: 1,
            target_role,
            method,
            command_family: None,
            declared_payload_bytes: 64,
        }
    }

    #[test]
    fn prelude_round_trips_and_refuses_newer_versions() -> Result<(), Box<dyn std::error::Error>> {
        let value = prelude(NodeRole::Indexer, RpcMethod::WriterIngest);
        assert_eq!(RpcPrelude::decode(&value.encode()?)?, value);
        let mut newer = value.encode()?;
        newer[8..10].copy_from_slice(&2_u16.to_le_bytes());
        assert!(matches!(
            RpcPrelude::decode(&newer),
            Err(RpcAuthorizationError::UnsupportedPreludeVersion(2))
        ));
        Ok(())
    }

    #[test]
    fn role_matrix_denies_before_oversized_payload_check() -> Result<(), Box<dyn std::error::Error>> {
        let mut denied = prelude(NodeRole::MetastoreVoter, RpcMethod::MetastoreCommand);
        denied.command_family = Some(MetastoreCommandFamily::WriterLease);
        denied.declared_payload_bytes = u32::MAX;
        assert_eq!(
            authorize_rpc(&identity(NodeRole::Gateway)?, NodeRole::MetastoreVoter, denied),
            Err(RpcAuthorizationError::PermissionDenied)
        );
        Ok(())
    }

    #[test]
    fn metastore_snapshot_requires_an_authenticated_voter_caller() -> Result<(), Box<dyn std::error::Error>> {
        let mut denied = prelude(NodeRole::MetastoreVoter, RpcMethod::MetastoreSnapshot);
        denied.declared_payload_bytes = u32::MAX;
        assert_eq!(
            authorize_rpc(&identity(NodeRole::Gateway)?, NodeRole::MetastoreVoter, denied),
            Err(RpcAuthorizationError::PermissionDenied)
        );

        authorize_rpc(
            &identity(NodeRole::MetastoreVoter)?,
            NodeRole::MetastoreVoter,
            prelude(NodeRole::MetastoreVoter, RpcMethod::MetastoreSnapshot),
        )?;
        let mut oversized = prelude(NodeRole::MetastoreVoter, RpcMethod::MetastoreSnapshot);
        oversized.declared_payload_bytes = u32::try_from(RPC_METASTORE_SNAPSHOT_REQUEST_FRAME_BYTES + 1)?;
        assert_eq!(
            authorize_rpc(
                &identity(NodeRole::MetastoreVoter)?,
                NodeRole::MetastoreVoter,
                oversized,
            ),
            Err(RpcAuthorizationError::PayloadTooLarge {
                actual: RPC_METASTORE_SNAPSHOT_REQUEST_FRAME_BYTES + 1,
                maximum: RPC_METASTORE_SNAPSHOT_REQUEST_FRAME_BYTES,
            })
        );
        Ok(())
    }

    #[test]
    fn admitted_metastore_family_round_trips_before_authorization() -> Result<(), Box<dyn std::error::Error>> {
        let mut allowed = prelude(NodeRole::MetastoreVoter, RpcMethod::MetastoreCommand);
        allowed.command_family = Some(MetastoreCommandFamily::WalRegistration);
        let decoded = RpcPrelude::decode(&allowed.encode()?)?;
        assert_eq!(decoded, allowed);
        authorize_rpc(&identity(NodeRole::Indexer)?, NodeRole::MetastoreVoter, decoded)?;
        Ok(())
    }

    #[test]
    fn security_family_admits_only_orchestrator_and_gateway_leaf_clients() -> Result<(), Box<dyn std::error::Error>> {
        let mut security = prelude(NodeRole::MetastoreVoter, RpcMethod::MetastoreCommand);
        security.command_family = Some(MetastoreCommandFamily::Security);
        assert_eq!(RpcPrelude::decode(&security.encode()?)?, security);
        authorize_rpc(&identity(NodeRole::Orchestrator)?, NodeRole::MetastoreVoter, security)?;
        authorize_rpc(&identity(NodeRole::Gateway)?, NodeRole::MetastoreVoter, security)?;
        for denied_role in [
            NodeRole::MetastoreVoter,
            NodeRole::Placement,
            NodeRole::Indexer,
            NodeRole::Searcher,
            NodeRole::Compactor,
        ] {
            assert_eq!(
                authorize_rpc(&identity(denied_role)?, NodeRole::MetastoreVoter, security),
                Err(RpcAuthorizationError::PermissionDenied)
            );
        }
        Ok(())
    }

    #[test]
    fn indexer_range_settlement_uses_split_merge_not_task_lease() -> Result<(), Box<dyn std::error::Error>> {
        let mut split_merge = prelude(NodeRole::MetastoreVoter, RpcMethod::MetastoreCommand);
        split_merge.command_family = Some(MetastoreCommandFamily::SplitMerge);
        authorize_rpc(&identity(NodeRole::Indexer)?, NodeRole::MetastoreVoter, split_merge)?;

        let mut task_lease = split_merge;
        task_lease.command_family = Some(MetastoreCommandFamily::TaskLease);
        assert_eq!(
            authorize_rpc(&identity(NodeRole::Indexer)?, NodeRole::MetastoreVoter, task_lease,),
            Err(RpcAuthorizationError::PermissionDenied)
        );
        Ok(())
    }

    #[test]
    fn admitted_edges_and_profile_downgrade_are_explicit() -> Result<(), Box<dyn std::error::Error>> {
        let allowed = prelude(NodeRole::Indexer, RpcMethod::WriterIngest);
        authorize_rpc(&identity(NodeRole::Gateway)?, NodeRole::Indexer, allowed)?;

        assert!(matches!(
            RpcVersionOffer::v1().negotiate(RpcVersionOffer {
                minimum_wire_version: 1,
                maximum_wire_version: 1,
                authorization_profile: 0,
            }),
            Err(RpcAuthorizationError::IncompatibleAuthProfile { offered: 0 })
        ));
        Ok(())
    }

    #[test]
    fn compactor_execution_is_placement_only_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
        let allowed = prelude(NodeRole::Compactor, RpcMethod::CompactorExecuteTask);
        assert_eq!(RpcPrelude::decode(&allowed.encode()?)?, allowed);
        authorize_rpc(&identity(NodeRole::Placement)?, NodeRole::Compactor, allowed)?;

        for denied_role in [
            NodeRole::MetastoreVoter,
            NodeRole::Orchestrator,
            NodeRole::Gateway,
            NodeRole::Indexer,
            NodeRole::Searcher,
            NodeRole::Compactor,
        ] {
            assert_eq!(
                authorize_rpc(&identity(denied_role)?, NodeRole::Compactor, allowed),
                Err(RpcAuthorizationError::PermissionDenied)
            );
        }
        Ok(())
    }

    #[test]
    fn core_api_authority_is_gateway_to_metastore_only_and_bounded() -> Result<(), Box<dyn std::error::Error>> {
        let allowed = prelude(NodeRole::MetastoreVoter, RpcMethod::CoreApiAuthority);
        assert_eq!(RpcPrelude::decode(&allowed.encode()?)?, allowed);
        assert_eq!(allowed.command_family, None);
        assert_eq!(allowed.method.payload_limit(), super::RPC_MAX_CONTROL_PAYLOAD_BYTES);
        authorize_rpc(&identity(NodeRole::Gateway)?, NodeRole::MetastoreVoter, allowed)?;

        for denied_role in [
            NodeRole::MetastoreVoter,
            NodeRole::Placement,
            NodeRole::Orchestrator,
            NodeRole::Indexer,
            NodeRole::Searcher,
            NodeRole::Compactor,
        ] {
            assert_eq!(
                authorize_rpc(&identity(denied_role)?, NodeRole::MetastoreVoter, allowed),
                Err(RpcAuthorizationError::PermissionDenied)
            );
        }
        Ok(())
    }
}