lodestone_client/

credential_rotation.rs

//! Create-new local publication for service credential rotations.

use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::path::{Component, Path, PathBuf};

use aws_lc_rs::digest;
use lodestone_wire::{Digest, Identity, ServiceCredentialReservation, U64};
use serde::{Deserialize, Serialize};

use crate::credential::{
    CredentialActivationError, CredentialProviderError, SecretBytes, credential_verifier, decode_file_credential,
    encode_file_credential,
};
use crate::{
    LocalOperatorWorkflowKind, ValidatedConnectionProfile, WorkflowJournal, WorkflowJournalError,
    WorkflowJournalIntent, WorkflowJournalTransition, WorkflowTransactionId, profile_sha256,
    secure_publication::{PublishedFileIdentity, SecureCreateNewFilePublisher, SecurePublicationError},
};

pub const CREDENTIAL_ROTATION_FORMAT_VERSION: u16 = 1;
const TRANSACTION_PREFIX: &str = ".lodestone-credential-rotation-v1-";
const PROFILE_STAGE_PREFIX: &str = ".lodestone-profile-rotation-v1-";
const JOURNAL_NAME: &str = "journal";
const CREDENTIAL_STAGE_NAME: &str = "credential.stage";
const PROFILE_STAGE_NAME: &str = "profile.stage";

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CredentialRotationIntent {
    pub format_version: u16,
    pub current_profile_sha256: [u8; 32],
    pub current_profile_generation: U64,
    pub cluster_id: Identity,
    pub collection_id: Identity,
    pub current_api_key_id: Identity,
    pub current_credential_generation: U64,
    pub credential_expires_at: U64,
    pub overlap_until: U64,
    pub reservation_expires_at: U64,
    pub credential_output: String,
    pub profile_output: String,
}

impl CredentialRotationIntent {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        current_profile: &ValidatedConnectionProfile,
        cluster_id: Identity,
        collection_id: Identity,
        credential_expires_at: U64,
        overlap_until: U64,
        reservation_expires_at: U64,
        credential_output: String,
        profile_output: String,
    ) -> Result<Self, CredentialRotationError> {
        let wire = current_profile.as_wire();
        let intent = Self {
            format_version: CREDENTIAL_ROTATION_FORMAT_VERSION,
            current_profile_sha256: profile_sha256(current_profile).map_err(CredentialRotationError::Journal)?,
            current_profile_generation: wire.profile_generation.clone(),
            cluster_id,
            collection_id,
            current_api_key_id: wire.credential_id.clone(),
            current_credential_generation: wire.credential_generation.clone(),
            credential_expires_at,
            overlap_until,
            reservation_expires_at,
            credential_output,
            profile_output,
        };
        intent.validate().map_err(CredentialRotationError::Journal)?;
        Ok(intent)
    }
}

impl crate::workflow_journal::sealed::Sealed for CredentialRotationIntent {}

impl WorkflowJournalIntent for CredentialRotationIntent {
    type Transition = CredentialRotationTransition;

    const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::RotateCredential;

    fn validate(&self) -> Result<(), WorkflowJournalError> {
        let profile_generation = canonical_u64(&self.current_profile_generation);
        let current_generation = canonical_u64(&self.current_credential_generation);
        let credential_expiry = canonical_u64(&self.credential_expires_at);
        let overlap = canonical_u64(&self.overlap_until);
        let reservation_expiry = canonical_u64(&self.reservation_expires_at);
        if self.format_version != CREDENTIAL_ROTATION_FORMAT_VERSION
            || !valid_sha256(&self.current_profile_sha256)
            || profile_generation.is_none_or(|value| value == 0)
            || current_generation.is_none_or(|value| value == 0)
            || credential_expiry.is_none_or(|value| value == 0)
            || overlap.is_none_or(|value| value == 0)
            || reservation_expiry.is_none_or(|value| value == 0)
            || !valid_identity(&self.cluster_id)
            || !valid_identity(&self.collection_id)
            || !valid_identity(&self.current_api_key_id)
            || !valid_output_path(&self.credential_output)
            || !valid_output_path(&self.profile_output)
            || !valid_file_reference_path(&self.credential_output)
            || self.credential_output == self.profile_output
        {
            return Err(WorkflowJournalError::InvalidIntent);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
pub enum CredentialRotationTransition {
    ActionsAllocated {
        attempt: u32,
        reservation_action_id: Digest,
        activation_action_id: Digest,
    },
    ReservationRecorded {
        reservation: ServiceCredentialReservation,
    },
    CredentialStaged {
        file: PublishedFileIdentity,
    },
    ReservationExpiredUnactivated {
        reservation_action_id: Digest,
    },
    Activated {
        principal_id: Identity,
        api_key_id: Identity,
        generation: U64,
    },
    CredentialPublished {
        file: PublishedFileIdentity,
    },
    ProfileStaged {
        file: PublishedFileIdentity,
    },
    ProfilePublished {
        file: PublishedFileIdentity,
    },
    Verified {
        principal_id: Identity,
        api_key_id: Identity,
        generation: U64,
    },
    Completed,
}

impl crate::workflow_journal::sealed::Sealed for CredentialRotationTransition {}

impl WorkflowJournalTransition for CredentialRotationTransition {
    fn validate(&self) -> Result<(), WorkflowJournalError> {
        let valid = match self {
            Self::ActionsAllocated {
                attempt: _,
                reservation_action_id,
                activation_action_id,
            } => {
                valid_digest(reservation_action_id)
                    && valid_digest(activation_action_id)
                    && reservation_action_id != activation_action_id
            }
            Self::ReservationRecorded { reservation } => valid_reservation(reservation),
            Self::CredentialStaged { file }
            | Self::CredentialPublished { file }
            | Self::ProfileStaged { file }
            | Self::ProfilePublished { file } => valid_file_identity(file),
            Self::ReservationExpiredUnactivated { reservation_action_id } => valid_digest(reservation_action_id),
            Self::Activated {
                principal_id,
                api_key_id,
                generation,
            }
            | Self::Verified {
                principal_id,
                api_key_id,
                generation,
            } => valid_identity(principal_id) && valid_identity(api_key_id) && canonical_u64(generation).is_some(),
            Self::Completed => true,
        };
        if valid {
            Ok(())
        } else {
            Err(WorkflowJournalError::InvalidTransition)
        }
    }
}

#[derive(Clone, Debug, Default, PartialEq)]
pub struct CredentialRotationState {
    attempt: u32,
    action_ids: Option<(Digest, Digest)>,
    reservation: Option<ServiceCredentialReservation>,
    credential_staged: Option<PublishedFileIdentity>,
    activated: Option<(Identity, Identity, U64)>,
    credential_published: Option<PublishedFileIdentity>,
    profile_staged: Option<PublishedFileIdentity>,
    profile_published: Option<PublishedFileIdentity>,
    verified: bool,
    completed: bool,
}

impl CredentialRotationState {
    pub fn replay(transitions: &[CredentialRotationTransition]) -> Result<Self, WorkflowJournalError> {
        let mut state = Self::default();
        for transition in transitions {
            transition.validate()?;
            state.apply(transition)?;
        }
        Ok(state)
    }

    fn apply(&mut self, transition: &CredentialRotationTransition) -> Result<(), WorkflowJournalError> {
        if self.completed {
            return Err(WorkflowJournalError::Corrupt);
        }
        match transition {
            CredentialRotationTransition::ActionsAllocated {
                attempt,
                reservation_action_id,
                activation_action_id,
            } if self.action_ids.is_none() && *attempt == self.attempt => {
                self.action_ids = Some((reservation_action_id.clone(), activation_action_id.clone()));
            }
            CredentialRotationTransition::ReservationRecorded { reservation }
                if self.reservation.is_none()
                    && self.action_ids.as_ref().is_some_and(|(reservation_id, activation_id)| {
                        &reservation.reservation_action_id == reservation_id
                            && &reservation.activation_action_id == activation_id
                    }) =>
            {
                self.reservation = Some(reservation.clone());
            }
            CredentialRotationTransition::CredentialStaged { file }
                if self.reservation.is_some() && self.credential_staged.is_none() =>
            {
                self.credential_staged = Some(file.clone());
            }
            CredentialRotationTransition::ReservationExpiredUnactivated { reservation_action_id }
                if self
                    .action_ids
                    .as_ref()
                    .is_some_and(|(current, _)| current == reservation_action_id)
                    && self
                        .reservation
                        .as_ref()
                        .is_none_or(|reservation| &reservation.reservation_action_id == reservation_action_id)
                    && self.activated.is_none()
                    && self.credential_published.is_none() =>
            {
                self.attempt = self.attempt.checked_add(1).ok_or(WorkflowJournalError::Corrupt)?;
                self.action_ids = None;
                self.reservation = None;
                self.credential_staged = None;
            }
            CredentialRotationTransition::Activated {
                principal_id,
                api_key_id,
                generation,
            } if self.activated.is_none()
                && self.credential_staged.is_some()
                && self.reservation.as_ref().is_some_and(|reservation| {
                    reservation.principal_id == *principal_id
                        && reservation.api_key_id == *api_key_id
                        && reservation.generation == *generation
                }) =>
            {
                self.activated = Some((principal_id.clone(), api_key_id.clone(), generation.clone()));
            }
            CredentialRotationTransition::CredentialPublished { file }
                if self.activated.is_some()
                    && self.credential_published.is_none()
                    && self.credential_staged.as_ref() == Some(file) =>
            {
                self.credential_published = Some(file.clone());
            }
            CredentialRotationTransition::ProfileStaged { file }
                if self.credential_published.is_some() && self.profile_staged.is_none() =>
            {
                self.profile_staged = Some(file.clone());
            }
            CredentialRotationTransition::ProfilePublished { file }
                if self.profile_published.is_none() && self.profile_staged.as_ref() == Some(file) =>
            {
                self.profile_published = Some(file.clone());
            }
            CredentialRotationTransition::Verified {
                principal_id,
                api_key_id,
                generation,
            } if !self.verified
                && self.profile_published.is_some()
                && self.activated.as_ref() == Some(&(principal_id.clone(), api_key_id.clone(), generation.clone())) =>
            {
                self.verified = true;
            }
            CredentialRotationTransition::Completed if self.verified => {
                self.completed = true;
            }
            _ => return Err(WorkflowJournalError::Corrupt),
        }
        Ok(())
    }

    #[must_use]
    pub fn action_ids(&self) -> Option<&(Digest, Digest)> {
        self.action_ids.as_ref()
    }

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

    #[must_use]
    pub fn reservation(&self) -> Option<&ServiceCredentialReservation> {
        self.reservation.as_ref()
    }

    #[must_use]
    pub fn credential_staged(&self) -> Option<&PublishedFileIdentity> {
        self.credential_staged.as_ref()
    }

    #[must_use]
    pub fn activated(&self) -> Option<&(Identity, Identity, U64)> {
        self.activated.as_ref()
    }

    #[must_use]
    pub fn credential_published(&self) -> Option<&PublishedFileIdentity> {
        self.credential_published.as_ref()
    }

    #[must_use]
    pub fn profile_staged(&self) -> Option<&PublishedFileIdentity> {
        self.profile_staged.as_ref()
    }

    #[must_use]
    pub fn profile_published(&self) -> Option<&PublishedFileIdentity> {
        self.profile_published.as_ref()
    }

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

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

pub struct StagedCredential {
    verifier: Digest,
}

impl StagedCredential {
    #[must_use]
    pub const fn verifier(&self) -> &Digest {
        &self.verifier
    }
}

/// Secret-owning create-new credential file used by workflows whose durable
/// journal is owned outside the credential-rotation workflow.
pub struct RecoverableServiceCredentialFile {
    publisher: SecureCreateNewFilePublisher,
}

pub struct StagedServiceCredentialFile {
    verifier: Digest,
    identity: PublishedFileIdentity,
}

impl StagedServiceCredentialFile {
    #[must_use]
    pub const fn verifier(&self) -> &Digest {
        &self.verifier
    }

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

impl RecoverableServiceCredentialFile {
    pub async fn open(
        stage_directory: PathBuf,
        stage_name: String,
        destination: PathBuf,
    ) -> Result<Self, CredentialRotationError> {
        let publisher = SecureCreateNewFilePublisher::open(stage_directory, stage_name, destination)
            .await
            .map_err(map_publication_error)?;
        Ok(Self { publisher })
    }

    pub async fn stage(
        &self,
        reservation: &ServiceCredentialReservation,
        expected: Option<PublishedFileIdentity>,
    ) -> Result<StagedServiceCredentialFile, CredentialRotationError> {
        let mut retried_invalid_unjournaled_stage = false;
        loop {
            match self
                .publisher
                .read_stage(4 * 1024)
                .await
                .map_err(map_publication_error)?
            {
                Some(mut bytes) => {
                    let material = match decode_file_credential(&bytes) {
                        Ok(material) => material,
                        Err(_error) if expected.is_none() && !retried_invalid_unjournaled_stage => {
                            bytes.fill(0);
                            std::hint::black_box(&mut bytes);
                            self.publisher
                                .remove_unpublished_stage(None)
                                .await
                                .map_err(map_publication_error)?;
                            retried_invalid_unjournaled_stage = true;
                            continue;
                        }
                        Err(error) => {
                            bytes.fill(0);
                            std::hint::black_box(&mut bytes);
                            return Err(CredentialRotationError::Credential(error));
                        }
                    };
                    if material.api_key_id() != &reservation.api_key_id
                        || material.generation() != &reservation.generation
                    {
                        bytes.fill(0);
                        std::hint::black_box(&mut bytes);
                        return Err(CredentialRotationError::Conflict);
                    }
                    let verifier = credential_verifier(&reservation.api_key_id, material.secret())
                        .map_err(CredentialRotationError::Activation)?;
                    let identity = self
                        .publisher
                        .stage_sensitive(bytes, 4 * 1024, expected)
                        .await
                        .map_err(map_publication_error)?;
                    return Ok(StagedServiceCredentialFile { verifier, identity });
                }
                None => {
                    if expected.is_some() {
                        return Err(CredentialRotationError::InvalidState);
                    }
                    let secret = SecretBytes::generate().map_err(CredentialRotationError::Credential)?;
                    let verifier = credential_verifier(&reservation.api_key_id, &secret)
                        .map_err(CredentialRotationError::Activation)?;
                    let bytes = encode_file_credential(&reservation.api_key_id, &reservation.generation, &secret)
                        .map_err(CredentialRotationError::Credential)?;
                    let identity = self
                        .publisher
                        .stage_sensitive(bytes, 4 * 1024, None)
                        .await
                        .map_err(map_publication_error)?;
                    return Ok(StagedServiceCredentialFile { verifier, identity });
                }
            }
        }
    }

    pub async fn publish(&self, expected: PublishedFileIdentity) -> Result<(), CredentialRotationError> {
        self.publisher.publish(expected).await.map_err(map_publication_error)
    }

    pub async fn verify_published(&self, expected: PublishedFileIdentity) -> Result<(), CredentialRotationError> {
        self.publisher
            .verify_published(expected)
            .await
            .map_err(map_publication_error)
    }
}

/// Locked create-new transaction for one exact pair of output paths.
pub struct AtomicFileCredentialMaterialStore {
    intent: CredentialRotationIntent,
    journal: WorkflowJournal<CredentialRotationIntent>,
    transitions: Vec<CredentialRotationTransition>,
    state: CredentialRotationState,
    credential_publisher: SecureCreateNewFilePublisher,
    profile_publisher: SecureCreateNewFilePublisher,
}

impl AtomicFileCredentialMaterialStore {
    pub async fn open_or_create(intent: CredentialRotationIntent) -> Result<Self, CredentialRotationError> {
        intent.validate().map_err(CredentialRotationError::Journal)?;

        #[cfg(any(target_os = "linux", target_os = "macos"))]
        {
            let setup_intent = intent.clone();
            let setup = tokio::task::spawn_blocking(move || unix::prepare_transaction(&setup_intent))
                .await
                .map_err(|_| CredentialRotationError::WorkerUnavailable)??;
            let journal_path = setup.transaction_path.join(JOURNAL_NAME);
            let replay = WorkflowJournal::open_or_create(journal_path, intent.clone())
                .await
                .map_err(CredentialRotationError::Journal)?;
            let (journal, transitions) = replay.into_parts();
            let state = CredentialRotationState::replay(&transitions).map_err(CredentialRotationError::Journal)?;
            let setup_state = state.clone();
            let paths = tokio::task::spawn_blocking(move || unix::finish_transaction_setup(setup, &setup_state))
                .await
                .map_err(|_| CredentialRotationError::WorkerUnavailable)??;
            let credential_publisher = SecureCreateNewFilePublisher::open(
                paths.transaction_path,
                CREDENTIAL_STAGE_NAME.to_owned(),
                PathBuf::from(&intent.credential_output),
            )
            .await
            .map_err(map_publication_error)?;
            let profile_publisher = SecureCreateNewFilePublisher::open(
                paths.profile_stage_path,
                PROFILE_STAGE_NAME.to_owned(),
                PathBuf::from(&intent.profile_output),
            )
            .await
            .map_err(map_publication_error)?;
            let credential_destination_identity = state
                .credential_published()
                .cloned()
                .or_else(|| state.activated().and_then(|_| state.credential_staged().cloned()));
            credential_publisher
                .validate_destination(credential_destination_identity, state.credential_published().is_some())
                .await
                .map_err(map_publication_error)?;
            let profile_destination_identity = state
                .profile_published()
                .cloned()
                .or_else(|| state.profile_staged().cloned());
            profile_publisher
                .validate_destination(profile_destination_identity, state.profile_published().is_some())
                .await
                .map_err(map_publication_error)?;
            Ok(Self {
                intent,
                journal,
                transitions,
                state,
                credential_publisher,
                profile_publisher,
            })
        }

        #[cfg(not(any(target_os = "linux", target_os = "macos")))]
        {
            let _ = intent;
            Err(CredentialRotationError::UnsupportedPlatform)
        }
    }

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

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

    pub async fn ensure_action_ids(&mut self) -> Result<(Digest, Digest), CredentialRotationError> {
        if let Some(ids) = self.state.action_ids() {
            return Ok(ids.clone());
        }
        let transaction_id = self.journal.transaction_id();
        let attempt = self.state.attempt();
        let reservation_action_id = action_id(
            b"lodestone/credential-rotation/reservation/v1\0",
            transaction_id,
            attempt,
        );
        let activation_action_id = action_id(
            b"lodestone/credential-rotation/activation/v1\0",
            transaction_id,
            attempt,
        );
        self.record(CredentialRotationTransition::ActionsAllocated {
            attempt,
            reservation_action_id: reservation_action_id.clone(),
            activation_action_id: activation_action_id.clone(),
        })
        .await?;
        Ok((reservation_action_id, activation_action_id))
    }

    pub async fn record_reservation(
        &mut self,
        reservation: ServiceCredentialReservation,
    ) -> Result<(), CredentialRotationError> {
        if let Some(current) = self.state.reservation() {
            return if current == &reservation {
                Ok(())
            } else {
                Err(CredentialRotationError::Conflict)
            };
        }
        self.record(CredentialRotationTransition::ReservationRecorded { reservation })
            .await
    }

    pub async fn stage_credential(&mut self) -> Result<StagedCredential, CredentialRotationError> {
        #[cfg(not(any(target_os = "linux", target_os = "macos")))]
        {
            return Err(CredentialRotationError::UnsupportedPlatform);
        }

        #[cfg(any(target_os = "linux", target_os = "macos"))]
        {
            let reservation = self
                .state
                .reservation()
                .cloned()
                .ok_or(CredentialRotationError::InvalidState)?;
            let expected = self.state.credential_staged().cloned();
            let mut retried_invalid_unjournaled_stage = false;
            let result = loop {
                match self
                    .credential_publisher
                    .read_stage(4 * 1024)
                    .await
                    .map_err(map_publication_error)?
                {
                    Some(mut bytes) => {
                        let material = match decode_file_credential(&bytes) {
                            Ok(material) => material,
                            Err(_error) if expected.is_none() && !retried_invalid_unjournaled_stage => {
                                bytes.fill(0);
                                std::hint::black_box(&mut bytes);
                                self.credential_publisher
                                    .remove_unpublished_stage(None)
                                    .await
                                    .map_err(map_publication_error)?;
                                retried_invalid_unjournaled_stage = true;
                                continue;
                            }
                            Err(error) => {
                                bytes.fill(0);
                                std::hint::black_box(&mut bytes);
                                return Err(CredentialRotationError::Credential(error));
                            }
                        };
                        if material.api_key_id() != &reservation.api_key_id
                            || material.generation() != &reservation.generation
                        {
                            bytes.fill(0);
                            std::hint::black_box(&mut bytes);
                            return Err(CredentialRotationError::Conflict);
                        }
                        let verifier = match credential_verifier(&reservation.api_key_id, material.secret()) {
                            Ok(verifier) => verifier,
                            Err(error) => {
                                bytes.fill(0);
                                std::hint::black_box(&mut bytes);
                                return Err(CredentialRotationError::Activation(error));
                            }
                        };
                        let identity = self
                            .credential_publisher
                            .stage_sensitive(bytes, 4 * 1024, expected.clone())
                            .await
                            .map_err(map_publication_error)?;
                        break unix::StagedCredentialResult { identity, verifier };
                    }
                    None => {
                        if expected.is_some() {
                            return Err(CredentialRotationError::InvalidState);
                        }
                        let secret = SecretBytes::generate().map_err(CredentialRotationError::Credential)?;
                        let verifier = credential_verifier(&reservation.api_key_id, &secret)
                            .map_err(CredentialRotationError::Activation)?;
                        let bytes = encode_file_credential(&reservation.api_key_id, &reservation.generation, &secret)
                            .map_err(CredentialRotationError::Credential)?;
                        let identity = self
                            .credential_publisher
                            .stage_sensitive(bytes, 4 * 1024, None)
                            .await
                            .map_err(map_publication_error)?;
                        break unix::StagedCredentialResult { identity, verifier };
                    }
                }
            };
            if self.state.credential_staged().is_none() {
                self.record(CredentialRotationTransition::CredentialStaged {
                    file: result.identity.clone(),
                })
                .await?;
            }
            Ok(StagedCredential {
                verifier: result.verifier,
            })
        }
    }

    pub async fn retire_expired_unactivated(&mut self) -> Result<(), CredentialRotationError> {
        let reservation_action_id = self
            .state
            .action_ids()
            .map(|(reservation_action_id, _)| reservation_action_id.clone())
            .ok_or(CredentialRotationError::InvalidState)?;
        let expected = self.state.credential_staged().cloned();

        #[cfg(not(any(target_os = "linux", target_os = "macos")))]
        {
            let _ = expected;
            return Err(CredentialRotationError::UnsupportedPlatform);
        }

        #[cfg(any(target_os = "linux", target_os = "macos"))]
        {
            self.credential_publisher
                .remove_unpublished_stage(expected)
                .await
                .map_err(map_publication_error)?;
            self.record(CredentialRotationTransition::ReservationExpiredUnactivated { reservation_action_id })
                .await
        }
    }

    pub async fn record_activation(
        &mut self,
        principal_id: Identity,
        api_key_id: Identity,
        generation: U64,
    ) -> Result<(), CredentialRotationError> {
        if let Some(current) = self.state.activated() {
            return if current == &(principal_id, api_key_id, generation) {
                Ok(())
            } else {
                Err(CredentialRotationError::Conflict)
            };
        }
        self.record(CredentialRotationTransition::Activated {
            principal_id,
            api_key_id,
            generation,
        })
        .await
    }

    pub async fn publish_credential(&mut self) -> Result<(), CredentialRotationError> {
        self.publish(FileKind::Credential).await
    }

    pub async fn stage_profile(&mut self, bytes: Vec<u8>) -> Result<(), CredentialRotationError> {
        #[cfg(not(any(target_os = "linux", target_os = "macos")))]
        {
            let _ = bytes;
            return Err(CredentialRotationError::UnsupportedPlatform);
        }

        #[cfg(any(target_os = "linux", target_os = "macos"))]
        {
            if bytes.is_empty() || bytes.len() > 64 * 1024 {
                return Err(CredentialRotationError::InvalidMaterial);
            }
            let expected = self.state.profile_staged().cloned();
            let identity = self
                .profile_publisher
                .stage(bytes, 64 * 1024, expected)
                .await
                .map_err(map_publication_error)?;
            if self.state.profile_staged().is_none() {
                self.record(CredentialRotationTransition::ProfileStaged { file: identity })
                    .await?;
            }
            Ok(())
        }
    }

    pub async fn publish_profile(&mut self) -> Result<(), CredentialRotationError> {
        self.publish(FileKind::Profile).await
    }

    pub async fn record_verified(
        &mut self,
        principal_id: Identity,
        api_key_id: Identity,
        generation: U64,
    ) -> Result<(), CredentialRotationError> {
        if self.state.verified() {
            return Ok(());
        }
        self.record(CredentialRotationTransition::Verified {
            principal_id,
            api_key_id,
            generation,
        })
        .await
    }

    pub async fn complete(&mut self) -> Result<(), CredentialRotationError> {
        if !self.state.completed() {
            self.record(CredentialRotationTransition::Completed).await?;
        }
        Ok(())
    }

    async fn publish(&mut self, kind: FileKind) -> Result<(), CredentialRotationError> {
        #[cfg(not(any(target_os = "linux", target_os = "macos")))]
        {
            let _ = kind;
            return Err(CredentialRotationError::UnsupportedPlatform);
        }

        #[cfg(any(target_os = "linux", target_os = "macos"))]
        {
            let expected = match kind {
                FileKind::Credential => self
                    .state
                    .credential_staged()
                    .cloned()
                    .ok_or(CredentialRotationError::InvalidState)?,
                FileKind::Profile => self
                    .state
                    .profile_staged()
                    .cloned()
                    .ok_or(CredentialRotationError::InvalidState)?,
            };
            let published = match kind {
                FileKind::Credential => self.state.credential_published(),
                FileKind::Profile => self.state.profile_published(),
            };
            match kind {
                FileKind::Credential => self.credential_publisher.publish(expected.clone()).await,
                FileKind::Profile => self.profile_publisher.publish(expected.clone()).await,
            }
            .map_err(map_publication_error)?;
            if published.is_none() {
                let transition = match kind {
                    FileKind::Credential => {
                        CredentialRotationTransition::CredentialPublished { file: expected.clone() }
                    }
                    FileKind::Profile => CredentialRotationTransition::ProfilePublished { file: expected.clone() },
                };
                self.record(transition).await?;
            }
            Ok(())
        }
    }

    async fn record(&mut self, transition: CredentialRotationTransition) -> Result<(), CredentialRotationError> {
        self.journal
            .append(transition.clone())
            .await
            .map_err(CredentialRotationError::Journal)?;
        self.transitions.push(transition);
        self.state = CredentialRotationState::replay(&self.transitions).map_err(CredentialRotationError::Journal)?;
        Ok(())
    }
}

#[derive(Clone, Copy)]
enum FileKind {
    Credential,
    Profile,
}

#[derive(Debug)]
pub enum CredentialRotationError {
    UnsupportedPlatform,
    InvalidPath,
    InsecurePath,
    DestinationExists,
    Conflict,
    InvalidState,
    InvalidMaterial,
    Io,
    RandomUnavailable,
    WorkerUnavailable,
    Journal(WorkflowJournalError),
    Credential(CredentialProviderError),
    Activation(CredentialActivationError),
}

impl CredentialRotationError {
    #[must_use]
    pub const fn code(&self) -> &'static str {
        match self {
            Self::UnsupportedPlatform => "unsupported_platform",
            Self::InvalidPath => "invalid_path",
            Self::InsecurePath => "insecure_path",
            Self::DestinationExists => "destination_exists",
            Self::Conflict => "conflict",
            Self::InvalidState => "invalid_state",
            Self::InvalidMaterial => "invalid_material",
            Self::Io => "io",
            Self::RandomUnavailable => "random_unavailable",
            Self::WorkerUnavailable => "worker_unavailable",
            Self::Journal(error) => error.code(),
            Self::Credential(_) => "credential",
            Self::Activation(_) => "activation",
        }
    }
}

impl Display for CredentialRotationError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::UnsupportedPlatform => "the platform cannot provide atomic credential publication",
            Self::InvalidPath => "a credential rotation path is not canonical and absolute",
            Self::InsecurePath => "a credential rotation path is symlinked, foreign-owned, or broadly writable",
            Self::DestinationExists => "a create-new credential rotation destination already exists",
            Self::Conflict => "credential rotation state conflicts with the requested transaction",
            Self::InvalidState => "credential rotation transitions are incomplete or out of order",
            Self::InvalidMaterial => "credential rotation material is invalid",
            Self::Io => "credential rotation filesystem operation failed",
            Self::RandomUnavailable => "secure random credential generation failed",
            Self::WorkerUnavailable => "credential rotation filesystem worker did not complete",
            Self::Journal(_) => "credential rotation journal failed",
            Self::Credential(_) => "credential rotation material decoding failed",
            Self::Activation(_) => "credential rotation verifier derivation failed",
        })
    }
}

impl Error for CredentialRotationError {}

pub fn credential_file_reference(path: &str) -> Result<String, CredentialRotationError> {
    if !valid_output_path(path) || !valid_file_reference_path(path) {
        return Err(CredentialRotationError::InvalidPath);
    }
    Ok(format!("file://{path}"))
}

fn valid_file_reference_path(path: &str) -> bool {
    path.len() <= 505
        && path
            .bytes()
            .all(|byte| byte.is_ascii_graphic() && !matches!(byte, b'@' | b'=' | b'?' | b'#'))
}

fn action_id(domain: &[u8], transaction_id: WorkflowTransactionId, attempt: u32) -> Digest {
    let mut context = digest::Context::new(&digest::SHA256);
    context.update(domain);
    context.update(&transaction_id.as_bytes());
    context.update(&attempt.to_be_bytes());
    Digest {
        value: lower_hex(context.finish().as_ref()),
    }
}

fn canonical_u64(value: &U64) -> Option<u64> {
    value
        .value
        .parse::<u64>()
        .ok()
        .filter(|parsed| parsed.to_string() == value.value)
}

fn valid_identity(value: &Identity) -> bool {
    value.value.len() == 32
        && value
            .value
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
        && u128::from_str_radix(&value.value, 16).is_ok_and(|value| value != 0)
}

fn valid_sha256(value: &[u8; 32]) -> bool {
    value.iter().any(|byte| *byte != 0)
}

fn valid_digest(value: &Digest) -> bool {
    value.value.len() == 64
        && value
            .value
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
        && value.value.bytes().any(|byte| byte != b'0')
}

fn valid_reservation(value: &ServiceCredentialReservation) -> bool {
    value.locator.authority == "service_credential_reservation"
        && !value.locator.value.is_empty()
        && valid_digest(&value.reservation_action_id)
        && valid_digest(&value.activation_action_id)
        && valid_identity(&value.principal_id)
        && valid_identity(&value.api_key_id)
        && canonical_u64(&value.generation).is_some_and(|generation| generation > 0)
        && canonical_u64(&value.reservation_expires_at).is_some_and(|expiry| expiry > 0)
        && canonical_u64(&value.credential_expires_at).is_some_and(|expiry| expiry > 0)
        && value.verifier_algorithm == "sha256_identity_bound_v1"
        && value.verifier_algorithm_version == 1
        && value.required_secret_bytes == 32
}

fn valid_file_identity(value: &PublishedFileIdentity) -> bool {
    value.is_valid()
}

fn map_publication_error(error: SecurePublicationError) -> CredentialRotationError {
    match error {
        SecurePublicationError::UnsupportedPlatform => CredentialRotationError::UnsupportedPlatform,
        SecurePublicationError::InvalidPath => CredentialRotationError::InvalidPath,
        SecurePublicationError::InsecurePath => CredentialRotationError::InsecurePath,
        SecurePublicationError::DestinationExists => CredentialRotationError::DestinationExists,
        SecurePublicationError::Conflict => CredentialRotationError::Conflict,
        SecurePublicationError::InvalidState => CredentialRotationError::InvalidState,
        SecurePublicationError::InvalidMaterial => CredentialRotationError::InvalidMaterial,
        SecurePublicationError::Io => CredentialRotationError::Io,
        SecurePublicationError::WorkerUnavailable => CredentialRotationError::WorkerUnavailable,
    }
}

fn valid_output_path(value: &str) -> bool {
    if value.is_empty() || value.bytes().any(|byte| byte.is_ascii_control()) {
        return false;
    }
    let path = Path::new(value);
    if !path.is_absolute()
        || path
            .components()
            .any(|component| !matches!(component, Component::RootDir | Component::Normal(_)))
    {
        return false;
    }
    path.components().collect::<PathBuf>().as_os_str() == path.as_os_str()
        && path.parent().is_some()
        && path.file_name().is_some_and(|name| !name.is_empty())
}

fn lower_hex(bytes: &[u8]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut output = String::with_capacity(bytes.len().saturating_mul(2));
    for byte in bytes {
        output.push(char::from(HEX[usize::from(byte >> 4)]));
        output.push(char::from(HEX[usize::from(byte & 0x0f)]));
    }
    output
}

#[cfg(any(target_os = "linux", target_os = "macos"))]
mod unix {
    use std::ffi::OsStr;
    use std::fs::File;
    use std::path::{Component, Path, PathBuf};

    use aws_lc_rs::digest;
    use rustix::fs::{FileType, Mode, OFlags, fstat, fsync, mkdirat, open, openat};

    use super::{
        CREDENTIAL_STAGE_NAME, CredentialRotationError, CredentialRotationIntent, CredentialRotationState,
        JOURNAL_NAME, PROFILE_STAGE_NAME, PROFILE_STAGE_PREFIX, PublishedFileIdentity, TRANSACTION_PREFIX, lower_hex,
    };

    const DIRECTORY_MODE: rustix::fs::RawMode = 0o700;

    pub(super) struct Setup {
        pub transaction_path: PathBuf,
        transaction: File,
        profile_stage_path: Option<PathBuf>,
    }

    pub(super) struct PublicationPaths {
        pub transaction_path: PathBuf,
        pub profile_stage_path: PathBuf,
    }

    pub(super) struct StagedCredentialResult {
        pub identity: PublishedFileIdentity,
        pub verifier: lodestone_wire::Digest,
    }

    pub(super) fn prepare_transaction(intent: &CredentialRotationIntent) -> Result<Setup, CredentialRotationError> {
        let credential_path = Path::new(&intent.credential_output);
        let profile_path = Path::new(&intent.profile_output);
        let (credential_parent_path, _credential_name) = split_output(credential_path)?;
        let (profile_parent_path, _profile_name) = split_output(profile_path)?;
        let credential_parent = open_directory_chain(credential_parent_path)?;
        let profile_parent = open_directory_chain(profile_parent_path)?;
        validate_parent(&credential_parent)?;
        validate_parent(&profile_parent)?;

        let key = path_key(&intent.credential_output);
        let transaction_name = format!("{TRANSACTION_PREFIX}{}", &lower_hex(&key)[..24]);
        let transaction_path = credential_parent_path.join(&transaction_name);
        if transaction_path == credential_path || transaction_path == profile_path {
            return Err(CredentialRotationError::InvalidPath);
        }
        let transaction = open_or_create_private_directory(&credential_parent, OsStr::new(&transaction_name))?;
        let credential_parent_stat = fstat(&credential_parent).map_err(|_| CredentialRotationError::Io)?;
        let profile_parent_stat = fstat(&profile_parent).map_err(|_| CredentialRotationError::Io)?;
        let same_parent = credential_parent_stat.st_dev == profile_parent_stat.st_dev
            && credential_parent_stat.st_ino == profile_parent_stat.st_ino;
        let profile_stage_path = if same_parent {
            None
        } else {
            Some(profile_parent_path.join(format!("{PROFILE_STAGE_PREFIX}{}", &lower_hex(&key)[..24])))
        };
        if profile_stage_path
            .as_ref()
            .is_some_and(|path| path == credential_path || path == profile_path)
        {
            return Err(CredentialRotationError::InvalidPath);
        }
        Ok(Setup {
            transaction_path,
            transaction,
            profile_stage_path,
        })
    }

    pub(super) fn finish_transaction_setup(
        setup: Setup,
        _state: &CredentialRotationState,
    ) -> Result<PublicationPaths, CredentialRotationError> {
        let transaction_path = setup.transaction_path.clone();
        let profile_stage_path = setup
            .profile_stage_path
            .clone()
            .unwrap_or_else(|| transaction_path.clone());
        let (profile_stage, profile_stage_is_transaction) = match &setup.profile_stage_path {
            None => (
                setup.transaction.try_clone().map_err(|_| CredentialRotationError::Io)?,
                true,
            ),
            Some(path) => {
                let parent = path.parent().ok_or(CredentialRotationError::InvalidPath)?;
                let name = path.file_name().ok_or(CredentialRotationError::InvalidPath)?;
                let parent = open_directory_chain(parent)?;
                (open_or_create_private_directory(&parent, name)?, false)
            }
        };
        validate_known_names(&setup.transaction, true)?;
        if !profile_stage_is_transaction {
            validate_known_names(&profile_stage, false)?;
        }
        Ok(PublicationPaths {
            transaction_path,
            profile_stage_path,
        })
    }

    fn split_output(path: &Path) -> Result<(&Path, &OsStr), CredentialRotationError> {
        Ok((
            path.parent().ok_or(CredentialRotationError::InvalidPath)?,
            path.file_name().ok_or(CredentialRotationError::InvalidPath)?,
        ))
    }

    fn open_directory_chain(path: &Path) -> Result<File, CredentialRotationError> {
        let mut components = path.components();
        if !matches!(components.next(), Some(Component::RootDir)) {
            return Err(CredentialRotationError::InvalidPath);
        }
        let flags = directory_flags();
        let mut current = File::from(open("/", flags, Mode::empty()).map_err(map_open_error)?);
        for component in components {
            let Component::Normal(name) = component else {
                return Err(CredentialRotationError::InvalidPath);
            };
            current = File::from(openat(&current, name, flags, Mode::empty()).map_err(map_open_error)?);
        }
        Ok(current)
    }

    fn open_or_create_private_directory(parent: &File, name: &OsStr) -> Result<File, CredentialRotationError> {
        match mkdirat(parent, name, Mode::from_raw_mode(DIRECTORY_MODE)) {
            Ok(()) => fsync(parent).map_err(|_| CredentialRotationError::Io)?,
            Err(error) if error == rustix::io::Errno::EXIST => {}
            Err(_) => return Err(CredentialRotationError::Io),
        }
        let directory = File::from(openat(parent, name, directory_flags(), Mode::empty()).map_err(map_open_error)?);
        validate_private_directory(&directory)?;
        Ok(directory)
    }

    const fn directory_flags() -> OFlags {
        OFlags::RDONLY
            .union(OFlags::DIRECTORY)
            .union(OFlags::CLOEXEC)
            .union(OFlags::NOFOLLOW)
    }

    fn validate_parent(parent: &File) -> Result<(), CredentialRotationError> {
        let stat = fstat(parent).map_err(|_| CredentialRotationError::Io)?;
        if FileType::from_raw_mode(stat.st_mode) != FileType::Directory
            || stat.st_uid != rustix::process::geteuid().as_raw()
            || stat.st_mode & 0o022 != 0
        {
            return Err(CredentialRotationError::InsecurePath);
        }
        Ok(())
    }

    fn validate_private_directory(directory: &File) -> Result<(), CredentialRotationError> {
        let stat = fstat(directory).map_err(|_| CredentialRotationError::Io)?;
        if FileType::from_raw_mode(stat.st_mode) != FileType::Directory
            || stat.st_uid != rustix::process::geteuid().as_raw()
            || stat.st_mode & 0o777 != DIRECTORY_MODE
        {
            return Err(CredentialRotationError::InsecurePath);
        }
        Ok(())
    }

    fn validate_known_names(directory: &File, transaction: bool) -> Result<(), CredentialRotationError> {
        let entries = rustix::fs::Dir::read_from(directory).map_err(|_| CredentialRotationError::Io)?;
        for entry in entries {
            let entry = entry.map_err(|_| CredentialRotationError::Io)?;
            let name = entry.file_name().to_bytes();
            if matches!(name, b"." | b"..") {
                continue;
            }
            let allowed = if transaction {
                name == JOURNAL_NAME.as_bytes()
                    || name == CREDENTIAL_STAGE_NAME.as_bytes()
                    || name == PROFILE_STAGE_NAME.as_bytes()
            } else {
                name == PROFILE_STAGE_NAME.as_bytes()
            };
            if !allowed {
                return Err(CredentialRotationError::Conflict);
            }
        }
        Ok(())
    }

    fn path_key(path: &str) -> [u8; 32] {
        let mut context = digest::Context::new(&digest::SHA256);
        context.update(b"lodestone/credential-rotation/path/v1\0");
        context.update(path.as_bytes());
        let observed = context.finish();
        let mut output = [0; 32];
        output.copy_from_slice(observed.as_ref());
        output
    }

    fn map_open_error(error: rustix::io::Errno) -> CredentialRotationError {
        match error {
            rustix::io::Errno::LOOP | rustix::io::Errno::NOTDIR => CredentialRotationError::InsecurePath,
            rustix::io::Errno::NOENT => CredentialRotationError::InvalidPath,
            _ => CredentialRotationError::Io,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{CredentialProvider, FileCredentialProvider};

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

    fn digest(value: u8) -> Digest {
        Digest {
            value: format!("{value:02x}").repeat(32),
        }
    }

    fn reservation() -> ServiceCredentialReservation {
        ServiceCredentialReservation {
            locator: lodestone_wire::Locator {
                authority: "service_credential_reservation".to_owned(),
                value: "locator".to_owned(),
            },
            reservation_action_id: digest(1),
            activation_action_id: digest(2),
            principal_id: identity(3),
            api_key_id: identity(4),
            generation: U64 { value: "2".to_owned() },
            reservation_expires_at: U64 {
                value: "100".to_owned(),
            },
            credential_expires_at: U64 {
                value: "200".to_owned(),
            },
            verifier_algorithm: "sha256_identity_bound_v1".to_owned(),
            verifier_algorithm_version: 1,
            required_secret_bytes: 32,
        }
    }

    #[test]
    fn reducer_accepts_only_the_atomic_publication_order() -> Result<(), WorkflowJournalError> {
        let file = PublishedFileIdentity {
            device: 1,
            inode: 2,
            size: 3,
            sha256: [4; 32],
        };
        let reservation = reservation();
        let transitions = vec![
            CredentialRotationTransition::ActionsAllocated {
                attempt: 0,
                reservation_action_id: reservation.reservation_action_id.clone(),
                activation_action_id: reservation.activation_action_id.clone(),
            },
            CredentialRotationTransition::ReservationRecorded {
                reservation: reservation.clone(),
            },
            CredentialRotationTransition::CredentialStaged { file: file.clone() },
            CredentialRotationTransition::Activated {
                principal_id: reservation.principal_id.clone(),
                api_key_id: reservation.api_key_id.clone(),
                generation: reservation.generation.clone(),
            },
            CredentialRotationTransition::CredentialPublished { file: file.clone() },
            CredentialRotationTransition::ProfileStaged { file: file.clone() },
            CredentialRotationTransition::ProfilePublished { file },
            CredentialRotationTransition::Verified {
                principal_id: reservation.principal_id,
                api_key_id: reservation.api_key_id,
                generation: reservation.generation,
            },
            CredentialRotationTransition::Completed,
        ];
        let state = CredentialRotationState::replay(&transitions)?;
        assert!(state.completed());
        assert_eq!(
            CredentialRotationState::replay(&transitions[1..]),
            Err(WorkflowJournalError::Corrupt)
        );
        Ok(())
    }

    #[test]
    fn reducer_retires_only_the_current_unactivated_attempt() -> Result<(), WorkflowJournalError> {
        let first_reservation = reservation();
        let second_reservation = ServiceCredentialReservation {
            reservation_action_id: digest(5),
            activation_action_id: digest(6),
            ..reservation()
        };
        let transitions = vec![
            CredentialRotationTransition::ActionsAllocated {
                attempt: 0,
                reservation_action_id: first_reservation.reservation_action_id.clone(),
                activation_action_id: first_reservation.activation_action_id.clone(),
            },
            CredentialRotationTransition::ReservationExpiredUnactivated {
                reservation_action_id: first_reservation.reservation_action_id,
            },
            CredentialRotationTransition::ActionsAllocated {
                attempt: 1,
                reservation_action_id: second_reservation.reservation_action_id.clone(),
                activation_action_id: second_reservation.activation_action_id.clone(),
            },
            CredentialRotationTransition::ReservationRecorded {
                reservation: second_reservation,
            },
        ];
        let state = CredentialRotationState::replay(&transitions)?;
        assert_eq!(state.attempt(), 1);
        assert!(state.reservation().is_some());
        assert_eq!(
            CredentialRotationState::replay(&[
                transitions[0].clone(),
                CredentialRotationTransition::ReservationExpiredUnactivated {
                    reservation_action_id: digest(7),
                },
            ]),
            Err(WorkflowJournalError::Corrupt)
        );
        Ok(())
    }

    #[test]
    fn output_paths_are_absolute_canonical_and_distinct() {
        assert!(valid_output_path("/var/lib/lodestone/credential.json"));
        assert!(!valid_output_path("relative/credential.json"));
        assert!(!valid_output_path("/var/lib/../credential.json"));
        assert!(!valid_output_path("/var/lib//credential.json"));
    }

    #[test]
    fn intent_rejects_a_zero_profile_digest() {
        let intent = CredentialRotationIntent {
            format_version: CREDENTIAL_ROTATION_FORMAT_VERSION,
            current_profile_sha256: [0; 32],
            current_profile_generation: U64 { value: "1".to_owned() },
            cluster_id: identity(10),
            collection_id: identity(11),
            current_api_key_id: identity(12),
            current_credential_generation: U64 { value: "1".to_owned() },
            credential_expires_at: U64 {
                value: "200".to_owned(),
            },
            overlap_until: U64 {
                value: "150".to_owned(),
            },
            reservation_expires_at: U64 {
                value: "100".to_owned(),
            },
            credential_output: "/var/lib/lodestone/credential.json".to_owned(),
            profile_output: "/var/lib/lodestone/profile.json".to_owned(),
        };
        assert_eq!(intent.validate(), Err(WorkflowJournalError::InvalidIntent));
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    #[tokio::test]
    async fn atomic_store_recovers_a_link_before_its_journal_record() -> Result<(), Box<dyn Error>> {
        use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};

        struct Cleanup(PathBuf);
        impl Drop for Cleanup {
            fn drop(&mut self) {
                let _ = std::fs::remove_dir_all(&self.0);
            }
        }

        let mut nonce = [0_u8; 8];
        getrandom::fill(&mut nonce)?;
        let root =
            std::fs::canonicalize(std::env::temp_dir())?.join(format!("lodestone-rotation-test-{}", lower_hex(&nonce)));
        std::fs::create_dir(&root)?;
        std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o700))?;
        let _cleanup = Cleanup(root.clone());
        let credential_path = root.join("credential.json");
        let profile_path = root.join("profile.json");
        let intent = CredentialRotationIntent {
            format_version: CREDENTIAL_ROTATION_FORMAT_VERSION,
            current_profile_sha256: [9; 32],
            current_profile_generation: U64 { value: "1".to_owned() },
            cluster_id: identity(10),
            collection_id: identity(11),
            current_api_key_id: identity(12),
            current_credential_generation: U64 { value: "1".to_owned() },
            credential_expires_at: U64 {
                value: "200".to_owned(),
            },
            overlap_until: U64 {
                value: "150".to_owned(),
            },
            reservation_expires_at: U64 {
                value: "100".to_owned(),
            },
            credential_output: credential_path.to_string_lossy().into_owned(),
            profile_output: profile_path.to_string_lossy().into_owned(),
        };
        let mut store = AtomicFileCredentialMaterialStore::open_or_create(intent.clone()).await?;
        let first_action_ids = store.ensure_action_ids().await?;
        let first_reservation = ServiceCredentialReservation {
            reservation_action_id: first_action_ids.0,
            activation_action_id: first_action_ids.1,
            ..reservation()
        };
        store.record_reservation(first_reservation).await?;
        let first_staged = store.stage_credential().await?;
        store.retire_expired_unactivated().await?;
        assert_eq!(store.state().attempt(), 1);
        assert!(store.state().action_ids().is_none());
        assert!(store.state().reservation().is_none());
        assert!(store.state().credential_staged().is_none());

        let second_action_ids = store.ensure_action_ids().await?;
        assert_ne!(
            second_action_ids,
            action_ids_for_attempt(store.journal.transaction_id(), 0)
        );
        let reservation = ServiceCredentialReservation {
            reservation_action_id: second_action_ids.0,
            activation_action_id: second_action_ids.1,
            ..reservation()
        };
        store.record_reservation(reservation.clone()).await?;
        let second_staged = store.stage_credential().await?;
        assert_ne!(first_staged.verifier(), second_staged.verifier());
        store
            .record_activation(
                reservation.principal_id.clone(),
                reservation.api_key_id.clone(),
                reservation.generation.clone(),
            )
            .await?;
        let expected = store
            .state()
            .credential_staged()
            .cloned()
            .ok_or(CredentialRotationError::InvalidState)?;
        store.credential_publisher.publish(expected).await?;
        drop(store);

        let mut resumed = AtomicFileCredentialMaterialStore::open_or_create(intent.clone()).await?;
        assert!(resumed.state().credential_published().is_none());
        resumed.publish_credential().await?;
        let metadata = std::fs::metadata(&credential_path)?;
        assert_eq!(metadata.permissions().mode() & 0o777, 0o600);
        assert_eq!(metadata.nlink(), 1);
        let reference = credential_file_reference(&credential_path.to_string_lossy())?;
        let material = FileCredentialProvider.provide(&reference).await?;
        assert_eq!(material.api_key_id(), &reservation.api_key_id);

        let profile_bytes = br#"{"format_version":1,"redacted":true}"#.to_vec();
        resumed.stage_profile(profile_bytes.clone()).await?;
        resumed.publish_profile().await?;
        assert_eq!(std::fs::read(&profile_path)?, profile_bytes);
        resumed
            .record_verified(reservation.principal_id, reservation.api_key_id, reservation.generation)
            .await?;
        resumed.complete().await?;
        assert!(resumed.state().completed());
        drop(resumed);
        let reopened = AtomicFileCredentialMaterialStore::open_or_create(intent).await?;
        assert!(reopened.state().completed());
        Ok(())
    }

    fn action_ids_for_attempt(transaction_id: WorkflowTransactionId, attempt: u32) -> (Digest, Digest) {
        (
            action_id(
                b"lodestone/credential-rotation/reservation/v1\0",
                transaction_id,
                attempt,
            ),
            action_id(
                b"lodestone/credential-rotation/activation/v1\0",
                transaction_id,
                attempt,
            ),
        )
    }
}