lodestone_client/

config.rs

use std::collections::BTreeSet;
use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use aws_lc_rs::digest;
use http::Uri;
use lodestone_wire::{
    ConnectionProfile as WireConnectionProfile, LODESTONE_API_CONNECTION_PROFILE_FORMAT_VERSION,
    LODESTONE_API_SCHEMA_ID, LODESTONE_API_SPEC_SHA256, LODESTONE_API_STORAGE_PROFILE_EXPECTATION_FORMAT_VERSION,
};
use rustls::pki_types::ServerName;

use crate::CredentialProvider;

const MAX_PROFILE_BYTES: usize = 64 * 1024;
const MAX_ENDPOINTS: usize = 8;
const MAX_ENDPOINT_BYTES: usize = 512;
const MAX_REFERENCE_BYTES: usize = 512;
const MAX_TRUST_CERTIFICATES: usize = 16;
const MAX_TRUST_CERTIFICATE_BYTES: usize = 128 * 1024;
const TRUST_DIGEST_DOMAIN: &[u8] = b"lodestone/client/tls-trust/v1";

#[derive(Clone, Debug, PartialEq)]
pub struct ValidatedConnectionProfile(WireConnectionProfile);

impl ValidatedConnectionProfile {
    pub fn new(profile: WireConnectionProfile) -> Result<Self, ClientConfigError> {
        validate_profile(&profile)?;
        Ok(Self(profile))
    }

    pub fn decode_current(bytes: &[u8]) -> Result<Self, ClientConfigError> {
        if bytes.is_empty() || bytes.len() > MAX_PROFILE_BYTES {
            return Err(ClientConfigError::ProfileSize);
        }
        let profile: WireConnectionProfile =
            serde_json::from_slice(bytes).map_err(|_| ClientConfigError::ProfileCodec)?;
        let value = Self::new(profile)?;
        let canonical = serde_json::to_vec(value.as_wire()).map_err(|_| ClientConfigError::ProfileCodec)?;
        if canonical != bytes {
            return Err(ClientConfigError::NonCanonicalProfile);
        }
        Ok(value)
    }

    pub fn encode_current(&self) -> Result<Vec<u8>, ClientConfigError> {
        validate_profile(&self.0)?;
        let bytes = serde_json::to_vec(&self.0).map_err(|_| ClientConfigError::ProfileCodec)?;
        if bytes.len() > MAX_PROFILE_BYTES {
            return Err(ClientConfigError::ProfileSize);
        }
        Ok(bytes)
    }

    #[must_use]
    pub const fn as_wire(&self) -> &WireConnectionProfile {
        &self.0
    }

    #[must_use]
    pub fn profile_generation(&self) -> u64 {
        parse_u64(&self.0.profile_generation.value).unwrap_or(0)
    }

    pub fn compare_and_swap(
        current: Option<&Self>,
        expected_generation: u64,
        replacement: &Self,
    ) -> Result<(), ClientConfigError> {
        validate_profile(replacement.as_wire())?;
        match current {
            None if expected_generation == 0 && replacement.profile_generation() == 1 => Ok(()),
            Some(current) => {
                validate_profile(current.as_wire())?;
                let next = current
                    .profile_generation()
                    .checked_add(1)
                    .ok_or(ClientConfigError::GenerationExhausted)?;
                let prior = current.as_wire();
                let next_profile = replacement.as_wire();
                if expected_generation != current.profile_generation()
                    || replacement.profile_generation() != next
                    || prior.expected_cluster_id != next_profile.expected_cluster_id
                    || prior.expected_collection_id != next_profile.expected_collection_id
                    || prior.expected_principal_id != next_profile.expected_principal_id
                    || prior.storage_profile_expectation != next_profile.storage_profile_expectation
                {
                    return Err(ClientConfigError::ProfileCasConflict);
                }
                Ok(())
            }
            None => Err(ClientConfigError::ProfileCasConflict),
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct TlsTrust {
    certificates_der: Vec<Vec<u8>>,
    digest: [u8; 32],
}

impl TlsTrust {
    pub fn new(certificates_der: Vec<Vec<u8>>) -> Result<Self, ClientConfigError> {
        if certificates_der.is_empty()
            || certificates_der.len() > MAX_TRUST_CERTIFICATES
            || certificates_der
                .iter()
                .any(|certificate| certificate.is_empty() || certificate.len() > MAX_TRUST_CERTIFICATE_BYTES)
        {
            return Err(ClientConfigError::InvalidTrustMaterial);
        }
        let mut hasher = digest::Context::new(&digest::SHA256);
        hasher.update(TRUST_DIGEST_DOMAIN);
        for certificate in &certificates_der {
            let length = u64::try_from(certificate.len()).map_err(|_| ClientConfigError::InvalidTrustMaterial)?;
            hasher.update(&length.to_be_bytes());
            hasher.update(certificate);
        }
        let observed = hasher.finish();
        let mut trust_digest = [0; 32];
        trust_digest.copy_from_slice(observed.as_ref());
        Ok(Self {
            certificates_der,
            digest: trust_digest,
        })
    }

    #[must_use]
    pub const fn digest(&self) -> [u8; 32] {
        self.digest
    }

    #[must_use]
    pub fn certificates_der(&self) -> &[Vec<u8>] {
        &self.certificates_der
    }
}

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub struct PoolConfig {
    pub maximum_connections: usize,
    pub maximum_idle_connections: usize,
    pub connect_timeout: Duration,
    pub maximum_read_retries: u8,
    pub maximum_subscription_reconnects: u8,
}

impl Default for PoolConfig {
    fn default() -> Self {
        Self {
            maximum_connections: 4,
            maximum_idle_connections: 4,
            connect_timeout: Duration::from_secs(5),
            maximum_read_retries: 2,
            maximum_subscription_reconnects: 4,
        }
    }
}

impl PoolConfig {
    fn validate(self) -> Result<Self, ClientConfigError> {
        if self.maximum_connections == 0
            || self.maximum_connections > 32
            || self.maximum_idle_connections > self.maximum_connections
            || self.connect_timeout.is_zero()
            || self.connect_timeout > Duration::from_secs(60)
            || self.maximum_read_retries > 3
            || self.maximum_subscription_reconnects > 8
        {
            return Err(ClientConfigError::InvalidPoolConfig);
        }
        Ok(self)
    }
}

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub enum CellOperationClass {
    Ordinary,
    CollectionProvisioner,
    CapacityObserver,
    RecoveryOperator,
}

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
pub struct CellTransportIdentity {
    native_endpoints: Vec<String>,
    selected_stream_endpoint: String,
    tls_server_name: String,
    tls_trust_sha256: [u8; 32],
    expected_cluster_id: String,
    schema_id: String,
    schema_digest: String,
    operation_class: CellOperationClass,
    pool: PoolConfig,
}

impl CellTransportIdentity {
    #[must_use]
    pub fn native_endpoints(&self) -> &[String] {
        &self.native_endpoints
    }

    #[must_use]
    pub fn selected_stream_endpoint(&self) -> &str {
        &self.selected_stream_endpoint
    }

    #[must_use]
    pub fn tls_server_name(&self) -> &str {
        &self.tls_server_name
    }

    #[must_use]
    pub const fn tls_trust_sha256(&self) -> [u8; 32] {
        self.tls_trust_sha256
    }

    #[must_use]
    pub fn expected_cluster_id(&self) -> &str {
        &self.expected_cluster_id
    }

    #[must_use]
    pub fn schema_id(&self) -> &str {
        &self.schema_id
    }

    #[must_use]
    pub fn schema_digest(&self) -> &str {
        &self.schema_digest
    }

    #[must_use]
    pub const fn operation_class(&self) -> CellOperationClass {
        self.operation_class
    }

    #[must_use]
    pub const fn pool(&self) -> PoolConfig {
        self.pool
    }
}

pub struct CellTransportConfig {
    pub(crate) identity: CellTransportIdentity,
    pub(crate) trust: TlsTrust,
    pub(crate) pool: PoolConfig,
    pub(crate) native_endpoints: Vec<ValidatedEndpoint>,
    pub(crate) stream_endpoint: ValidatedEndpoint,
}

impl CellTransportConfig {
    pub fn new(
        profile: &ValidatedConnectionProfile,
        trust: TlsTrust,
        operation_class: CellOperationClass,
        pool: PoolConfig,
    ) -> Result<Self, ClientConfigError> {
        validate_profile(profile.as_wire())?;
        let expected_trust = decode_digest(&profile.as_wire().tls_trust_sha256.value)?;
        if expected_trust != trust.digest() {
            return Err(ClientConfigError::TrustDigestMismatch);
        }
        let pool = pool.validate()?;
        let native_endpoints = profile
            .as_wire()
            .native_endpoints
            .iter()
            .map(|endpoint| ValidatedEndpoint::parse(endpoint, EndpointScheme::NativeTls))
            .collect::<Result<Vec<_>, _>>()?;
        let stream_endpoint =
            ValidatedEndpoint::parse(&profile.as_wire().selected_stream_endpoint, EndpointScheme::Https)?;
        let identity = CellTransportIdentity {
            native_endpoints: canonical_endpoint_set(&profile.as_wire().native_endpoints),
            selected_stream_endpoint: profile.as_wire().selected_stream_endpoint.clone(),
            tls_server_name: profile.as_wire().tls_server_name.clone(),
            tls_trust_sha256: trust.digest(),
            expected_cluster_id: profile.as_wire().expected_cluster_id.value.clone(),
            schema_id: profile.as_wire().schema_id.clone(),
            schema_digest: profile.as_wire().schema_digest.value.clone(),
            operation_class,
            pool,
        };
        Ok(Self {
            identity,
            trust,
            pool,
            native_endpoints,
            stream_endpoint,
        })
    }

    #[must_use]
    pub const fn identity(&self) -> &CellTransportIdentity {
        &self.identity
    }
}

pub struct CollectionSessionConfig {
    pub(crate) profile: ValidatedConnectionProfile,
    pub(crate) credentials: Arc<dyn CredentialProvider>,
}

impl CollectionSessionConfig {
    pub fn new(
        profile: ValidatedConnectionProfile,
        credentials: Arc<dyn CredentialProvider>,
    ) -> Result<Self, ClientConfigError> {
        validate_profile(profile.as_wire())?;
        if profile
            .as_wire()
            .credential_expires_at
            .as_ref()
            .is_some_and(|expiry| parse_u64(&expiry.value).is_none_or(|expiry| expiry <= unix_millis()))
        {
            return Err(ClientConfigError::CredentialExpired);
        }
        Ok(Self { profile, credentials })
    }

    #[must_use]
    pub const fn profile(&self) -> &ValidatedConnectionProfile {
        &self.profile
    }

    pub(crate) fn validate_transport(&self, transport: &CellTransportConfig) -> Result<(), ClientConfigError> {
        let profile = self.profile.as_wire();
        let identity = transport.identity();
        if identity.operation_class != CellOperationClass::Ordinary
            || identity.native_endpoints != canonical_endpoint_set(&profile.native_endpoints)
            || identity.selected_stream_endpoint != profile.selected_stream_endpoint
            || identity.tls_server_name != profile.tls_server_name
            || identity.tls_trust_sha256 != decode_digest(&profile.tls_trust_sha256.value)?
            || identity.expected_cluster_id != profile.expected_cluster_id.value
            || identity.schema_id != profile.schema_id
            || identity.schema_digest != profile.schema_digest.value
        {
            return Err(ClientConfigError::CellTransportMismatch);
        }
        Ok(())
    }
}

fn canonical_endpoint_set(endpoints: &[String]) -> Vec<String> {
    endpoints.iter().cloned().collect::<BTreeSet<_>>().into_iter().collect()
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum EndpointScheme {
    NativeTls,
    Https,
}

impl EndpointScheme {
    const fn value(self) -> &'static str {
        match self {
            Self::NativeTls => "lodestone+tls",
            Self::Https => "https",
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ValidatedEndpoint {
    original: String,
    host: String,
    port: u16,
    scheme: EndpointScheme,
}

impl ValidatedEndpoint {
    fn parse(value: &str, scheme: EndpointScheme) -> Result<Self, ClientConfigError> {
        if value.is_empty()
            || value.len() > MAX_ENDPOINT_BYTES
            || value
                .bytes()
                .any(|byte| byte.is_ascii_control() || byte.is_ascii_whitespace())
            || value.contains('@')
            || value.contains('?')
            || value.contains('#')
        {
            return Err(ClientConfigError::InvalidEndpoint);
        }
        let uri: Uri = value.parse().map_err(|_| ClientConfigError::InvalidEndpoint)?;
        if uri.scheme_str() != Some(scheme.value()) || uri.path() != "" && uri.path() != "/" {
            return Err(ClientConfigError::InvalidEndpoint);
        }
        let host = uri.host().ok_or(ClientConfigError::InvalidEndpoint)?.to_owned();
        let port = uri.port_u16().ok_or(ClientConfigError::InvalidEndpoint)?;
        if host.is_empty() || host.len() > 253 || port == 0 {
            return Err(ClientConfigError::InvalidEndpoint);
        }
        Ok(Self {
            original: value.to_owned(),
            host,
            port,
            scheme,
        })
    }

    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.original
    }

    pub(crate) fn host(&self) -> &str {
        &self.host
    }

    pub(crate) const fn port(&self) -> u16 {
        self.port
    }
}

fn validate_profile(profile: &WireConnectionProfile) -> Result<(), ClientConfigError> {
    if profile.format_version != u32::from(LODESTONE_API_CONNECTION_PROFILE_FORMAT_VERSION) {
        return Err(ClientConfigError::UnsupportedProfileVersion);
    }
    if parse_u64(&profile.profile_generation.value).is_none_or(|value| value == 0)
        || profile.native_endpoints.is_empty()
        || profile.native_endpoints.len() > MAX_ENDPOINTS
        || profile.schema_id != LODESTONE_API_SCHEMA_ID
        || profile.schema_digest.value != LODESTONE_API_SPEC_SHA256
        || profile.expected_cluster_id.validate().is_err()
        || profile.expected_collection_id.validate().is_err()
        || profile.expected_principal_id.validate().is_err()
        || profile.credential_id.validate().is_err()
        || parse_u64(&profile.credential_generation.value).is_none_or(|value| value == 0)
        || profile
            .credential_expires_at
            .as_ref()
            .is_some_and(|value| parse_u64(&value.value).is_none_or(|value| value == 0))
        || decode_digest(&profile.tls_trust_sha256.value).is_err()
        || !valid_reference(&profile.secret_reference)
    {
        return Err(ClientConfigError::InvalidProfileBinding);
    }
    if profile.storage_profile_expectation.as_ref().is_some_and(|expectation| {
        expectation.format_version != u32::from(LODESTONE_API_STORAGE_PROFILE_EXPECTATION_FORMAT_VERSION)
            || expectation.required_posture.is_none() && expectation.expected_storage_profile_digest.is_none()
            || expectation
                .expected_storage_profile_digest
                .as_ref()
                .is_some_and(|digest| digest.validate().is_err())
    }) {
        return Err(ClientConfigError::InvalidProfileBinding);
    }
    let mut endpoints = BTreeSet::new();
    for endpoint in &profile.native_endpoints {
        ValidatedEndpoint::parse(endpoint, EndpointScheme::NativeTls)?;
        if !endpoints.insert(endpoint) {
            return Err(ClientConfigError::InvalidEndpoint);
        }
    }
    ValidatedEndpoint::parse(&profile.selected_stream_endpoint, EndpointScheme::Https)?;
    let tls_server_name = ServerName::try_from(profile.tls_server_name.clone());
    if profile.tls_server_name.is_empty()
        || profile.tls_server_name.len() > 253
        || profile
            .tls_server_name
            .bytes()
            .any(|byte| byte.is_ascii_control() || byte.is_ascii_whitespace())
        || !matches!(tls_server_name, Ok(ServerName::DnsName(_)))
    {
        return Err(ClientConfigError::InvalidServerName);
    }
    Ok(())
}

fn valid_reference(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= MAX_REFERENCE_BYTES
        && value.contains("://")
        && !value.contains('@')
        && !value.contains('=')
        && value
            .bytes()
            .all(|byte| byte.is_ascii_graphic() && !matches!(byte, b'?' | b'#'))
}

pub(crate) fn parse_u64(value: &str) -> Option<u64> {
    value.parse::<u64>().ok().filter(|parsed| parsed.to_string() == value)
}

pub(crate) fn parse_identity(value: &str) -> Option<u128> {
    if value.len() != 32 {
        return None;
    }
    u128::from_str_radix(value, 16).ok().filter(|value| *value != 0)
}

fn decode_digest(value: &str) -> Result<[u8; 32], ClientConfigError> {
    if value.len() != 64 {
        return Err(ClientConfigError::InvalidProfileBinding);
    }
    let mut output = [0; 32];
    for (index, pair) in value.as_bytes().as_chunks::<2>().0.iter().enumerate() {
        let text = std::str::from_utf8(pair).map_err(|_| ClientConfigError::InvalidProfileBinding)?;
        output[index] = u8::from_str_radix(text, 16).map_err(|_| ClientConfigError::InvalidProfileBinding)?;
    }
    Ok(output)
}

fn unix_millis() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .ok()
        .and_then(|duration| u64::try_from(duration.as_millis()).ok())
        .unwrap_or(u64::MAX)
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ClientConfigError {
    UnsupportedProfileVersion,
    InvalidProfileBinding,
    InvalidEndpoint,
    InvalidServerName,
    InvalidTrustMaterial,
    TrustDigestMismatch,
    InvalidPoolConfig,
    CredentialExpired,
    ProfileSize,
    ProfileCodec,
    NonCanonicalProfile,
    ProfileCasConflict,
    GenerationExhausted,
    CellTransportMismatch,
}

impl Display for ClientConfigError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::UnsupportedProfileVersion => "Core connection profile version is unsupported",
            Self::InvalidProfileBinding => "Core connection profile binding is invalid",
            Self::InvalidEndpoint => "Core connection endpoint is invalid",
            Self::InvalidServerName => "Core TLS server name is invalid",
            Self::InvalidTrustMaterial => "Core TLS trust material is invalid",
            Self::TrustDigestMismatch => "Core TLS trust material does not match the connection profile",
            Self::InvalidPoolConfig => "Core client pool configuration is invalid",
            Self::CredentialExpired => "Core connection profile credential is expired",
            Self::ProfileSize => "Core connection profile size is outside the bound",
            Self::ProfileCodec => "Core connection profile could not be decoded",
            Self::NonCanonicalProfile => "Core connection profile encoding is noncanonical",
            Self::ProfileCasConflict => "Core connection profile generation CAS conflicted",
            Self::GenerationExhausted => "Core connection profile generation is exhausted",
            Self::CellTransportMismatch => "Core Collection session does not match the selected cell transport",
        })
    }
}

impl Error for ClientConfigError {}

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

    use lodestone_wire::{
        Digest, Identity, LODESTONE_API_CONNECTION_PROFILE_FORMAT_VERSION, LODESTONE_API_SCHEMA_ID,
        LODESTONE_API_SPEC_SHA256, U64,
    };

    use super::*;
    use crate::FileCredentialProvider;

    fn profile(generation: u64, trust: [u8; 32]) -> Result<ValidatedConnectionProfile, ClientConfigError> {
        ValidatedConnectionProfile::new(WireConnectionProfile {
            format_version: u32::from(LODESTONE_API_CONNECTION_PROFILE_FORMAT_VERSION),
            profile_generation: U64 {
                value: generation.to_string(),
            },
            native_endpoints: vec!["lodestone+tls://core.example:8443".to_owned()],
            selected_stream_endpoint: "https://core.example:9443".to_owned(),
            tls_server_name: "core.example".to_owned(),
            tls_trust_sha256: Digest {
                value: lower_hex(&trust),
            },
            expected_cluster_id: identity(1),
            expected_collection_id: identity(2),
            expected_principal_id: identity(4),
            schema_id: LODESTONE_API_SCHEMA_ID.to_owned(),
            schema_digest: Digest {
                value: LODESTONE_API_SPEC_SHA256.to_owned(),
            },
            credential_id: identity(3),
            credential_generation: U64 { value: "1".to_owned() },
            credential_expires_at: None,
            secret_reference: "memory://core/admin".to_owned(),
            storage_profile_expectation: None,
        })
    }

    fn identity(value: u128) -> Identity {
        Identity {
            value: format!("{value:032x}"),
        }
    }

    fn lower_hex(bytes: &[u8]) -> String {
        bytes.iter().map(|byte| format!("{byte:02x}")).collect()
    }

    #[test]
    fn profile_codec_is_canonical_and_generation_cas_is_strict() -> Result<(), Box<dyn Error>> {
        let trust = TlsTrust::new(vec![vec![1, 2, 3]])?;
        let first = profile(1, trust.digest())?;
        let encoded = first.encode_current()?;
        assert_eq!(ValidatedConnectionProfile::decode_current(&encoded)?, first);
        let second = profile(2, trust.digest())?;
        ValidatedConnectionProfile::compare_and_swap(Some(&first), 1, &second)?;
        assert_eq!(
            ValidatedConnectionProfile::compare_and_swap(Some(&first), 0, &second),
            Err(ClientConfigError::ProfileCasConflict)
        );
        let mut other_principal_wire = second.as_wire().clone();
        other_principal_wire.expected_principal_id = identity(99);
        let other_principal = ValidatedConnectionProfile::new(other_principal_wire)?;
        assert_eq!(
            ValidatedConnectionProfile::compare_and_swap(Some(&first), 1, &other_principal),
            Err(ClientConfigError::ProfileCasConflict)
        );
        let mut noncanonical = encoded;
        noncanonical.push(b' ');
        assert_eq!(
            ValidatedConnectionProfile::decode_current(&noncanonical),
            Err(ClientConfigError::NonCanonicalProfile)
        );
        Ok(())
    }

    #[test]
    fn endpoints_have_no_downgrade_or_ambient_authority_forms() -> Result<(), Box<dyn Error>> {
        let trust = TlsTrust::new(vec![vec![1]])?;
        let mut value = profile(1, trust.digest())?.as_wire().clone();
        value.native_endpoints[0] = "tcp://core.example:8443".to_owned();
        assert_eq!(
            ValidatedConnectionProfile::new(value),
            Err(ClientConfigError::InvalidEndpoint)
        );
        Ok(())
    }

    #[test]
    fn explicit_tls_server_name_is_validated_independently_of_endpoint_hosts() -> Result<(), Box<dyn Error>> {
        let trust = TlsTrust::new(vec![vec![1]])?;
        let mut value = profile(1, trust.digest())?.as_wire().clone();
        value.native_endpoints[0] = "lodestone+tls://127.0.0.1:8443".to_owned();
        value.selected_stream_endpoint = "https://127.0.0.1:9443".to_owned();
        assert!(ValidatedConnectionProfile::new(value.clone()).is_ok());
        value.tls_server_name = "bad name".to_owned();
        assert_eq!(
            ValidatedConnectionProfile::new(value),
            Err(ClientConfigError::InvalidServerName)
        );
        let mut value = profile(1, trust.digest())?.as_wire().clone();
        value.tls_server_name = "127.0.0.1".to_owned();
        assert_eq!(
            ValidatedConnectionProfile::new(value),
            Err(ClientConfigError::InvalidServerName)
        );
        Ok(())
    }

    #[test]
    fn cell_identity_excludes_collection_and_credential_but_sessions_stay_exact() -> Result<(), Box<dyn Error>> {
        let trust = TlsTrust::new(vec![vec![1, 2, 3]])?;
        let first = profile(1, trust.digest())?;
        let mut second_wire = first.as_wire().clone();
        second_wire.profile_generation = U64 { value: "7".to_owned() };
        second_wire.expected_collection_id = identity(20);
        second_wire.expected_principal_id = identity(40);
        second_wire.credential_id = identity(30);
        second_wire.credential_generation = U64 { value: "4".to_owned() };
        second_wire.secret_reference = "memory://core/second".to_owned();
        let second = ValidatedConnectionProfile::new(second_wire)?;

        let first_transport = CellTransportConfig::new(
            &first,
            trust.clone(),
            CellOperationClass::Ordinary,
            PoolConfig::default(),
        )?;
        let second_transport = CellTransportConfig::new(
            &second,
            trust.clone(),
            CellOperationClass::Ordinary,
            PoolConfig::default(),
        )?;
        assert_eq!(first_transport.identity(), second_transport.identity());

        let second_session = CollectionSessionConfig::new(second, Arc::new(FileCredentialProvider))?;
        assert!(second_session.validate_transport(&first_transport).is_ok());

        let control_transport = CellTransportConfig::new(
            &first,
            trust,
            CellOperationClass::CollectionProvisioner,
            PoolConfig::default(),
        )?;
        let first_session = CollectionSessionConfig::new(first, Arc::new(FileCredentialProvider))?;
        assert_eq!(
            first_session.validate_transport(&control_transport),
            Err(ClientConfigError::CellTransportMismatch)
        );
        Ok(())
    }

    #[test]
    fn cluster_or_trust_omission_cannot_alias_a_cell_transport() -> Result<(), Box<dyn Error>> {
        let trust = TlsTrust::new(vec![vec![1, 2, 3]])?;
        let first = profile(1, trust.digest())?;
        let first_transport = CellTransportConfig::new(
            &first,
            trust.clone(),
            CellOperationClass::Ordinary,
            PoolConfig::default(),
        )?;

        let mut wrong_cluster_wire = first.as_wire().clone();
        wrong_cluster_wire.expected_cluster_id = identity(99);
        let wrong_cluster = ValidatedConnectionProfile::new(wrong_cluster_wire)?;
        let wrong_cluster_transport = CellTransportConfig::new(
            &wrong_cluster,
            trust,
            CellOperationClass::Ordinary,
            PoolConfig::default(),
        )?;
        assert_ne!(first_transport.identity(), wrong_cluster_transport.identity());

        let other_trust = TlsTrust::new(vec![vec![9, 8, 7]])?;
        let mut wrong_trust_wire = first.as_wire().clone();
        wrong_trust_wire.tls_trust_sha256 = Digest {
            value: lower_hex(&other_trust.digest()),
        };
        let wrong_trust = ValidatedConnectionProfile::new(wrong_trust_wire)?;
        let wrong_trust_transport = CellTransportConfig::new(
            &wrong_trust,
            other_trust,
            CellOperationClass::Ordinary,
            PoolConfig::default(),
        )?;
        assert_ne!(first_transport.identity(), wrong_trust_transport.identity());
        Ok(())
    }

    #[test]
    fn endpoint_order_does_not_split_one_cell_transport_identity() -> Result<(), Box<dyn Error>> {
        let trust = TlsTrust::new(vec![vec![1, 2, 3]])?;
        let mut first_wire = profile(1, trust.digest())?.as_wire().clone();
        first_wire
            .native_endpoints
            .push("lodestone+tls://core.example:8444".to_owned());
        let first = ValidatedConnectionProfile::new(first_wire.clone())?;
        first_wire.native_endpoints.reverse();
        first_wire.expected_collection_id = identity(20);
        first_wire.expected_principal_id = identity(40);
        first_wire.credential_id = identity(30);
        first_wire.secret_reference = "memory://core/second".to_owned();
        let second = ValidatedConnectionProfile::new(first_wire)?;

        let first_transport = CellTransportConfig::new(
            &first,
            trust.clone(),
            CellOperationClass::Ordinary,
            PoolConfig::default(),
        )?;
        let second_transport =
            CellTransportConfig::new(&second, trust, CellOperationClass::Ordinary, PoolConfig::default())?;
        assert_eq!(first_transport.identity(), second_transport.identity());
        Ok(())
    }

    #[test]
    fn storage_expectation_is_nonempty_current_and_preserved_by_rotation() -> Result<(), Box<dyn Error>> {
        let trust = TlsTrust::new(vec![vec![1, 2, 3]])?;
        let first = profile(1, trust.digest())?;
        let mut invalid = first.as_wire().clone();
        invalid.storage_profile_expectation = Some(lodestone_wire::StorageProfileExpectation {
            format_version: u32::from(LODESTONE_API_STORAGE_PROFILE_EXPECTATION_FORMAT_VERSION),
            required_posture: None,
            expected_storage_profile_digest: None,
        });
        assert_eq!(
            ValidatedConnectionProfile::new(invalid),
            Err(ClientConfigError::InvalidProfileBinding)
        );

        let mut pinned_wire = first.as_wire().clone();
        pinned_wire.storage_profile_expectation = Some(lodestone_wire::StorageProfileExpectation {
            format_version: u32::from(LODESTONE_API_STORAGE_PROFILE_EXPECTATION_FORMAT_VERSION),
            required_posture: Some(lodestone_wire::StorageOverallPosture::FilesystemDevelopment),
            expected_storage_profile_digest: None,
        });
        let pinned = ValidatedConnectionProfile::new(pinned_wire)?;
        let mut rotated_wire = pinned.as_wire().clone();
        rotated_wire.profile_generation = U64 { value: "2".to_owned() };
        let rotated = ValidatedConnectionProfile::new(rotated_wire)?;
        ValidatedConnectionProfile::compare_and_swap(Some(&pinned), 1, &rotated)?;

        let mut dropped_wire = rotated.as_wire().clone();
        dropped_wire.storage_profile_expectation = None;
        let dropped = ValidatedConnectionProfile::new(dropped_wire)?;
        assert_eq!(
            ValidatedConnectionProfile::compare_and_swap(Some(&pinned), 1, &dropped),
            Err(ClientConfigError::ProfileCasConflict)
        );
        Ok(())
    }
}