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
}
}
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)
}
}
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(¤t, 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,
),
)
}
}