use crate::{
APPLICATION_MODULE_LOCK_PROTOCOL, ApplicationModuleLock, CreateOperationResult,
DESIRED_MODULE_COMPOSITION_PROTOCOL, DesiredModuleComposition, EnvironmentManagementMode,
MODULE_APPROVAL_PROTOCOL, MODULE_CHANGE_PLAN_PROTOCOL, MODULE_ENVIRONMENT_POLICY_PROTOCOL,
MODULE_OPERATION_JOURNAL_PROTOCOL, MODULE_OPERATION_PROTOCOL, MODULE_REPAIR_PLAN_PROTOCOL,
ModuleApproval, ModuleApprovalBoundary, ModuleChangePlan, ModuleEffectReceipt,
ModuleEnvironmentPolicy, ModuleOperation, ModuleOperationError, ModuleOperationJournalEvent,
ModuleOperationKind, ModuleOperationLease, ModuleOperationState, ModuleOperationStore,
ModuleOperationStoreError, ModuleOperationTransition, ModulePlanEffect, ModuleRepairAction,
ModuleRepairPlan, ModuleResumeEvidence, ModuleRiskClass, ModuleRootChange,
ModuleWorkspaceBackup, application_module_lock_digest, desired_composition_digest,
journal_event_digest, module_change_plan_digest, module_repair_plan_digest,
};
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
use chrono::{DateTime, Duration, Utc};
use lenso_contracts::ArtifactReference;
use sha2::{Digest as _, Sha256};
use std::collections::BTreeSet;
use thiserror::Error;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ManagementActor {
pub actor_id: String,
pub verified_authorities: BTreeSet<String>,
}
#[derive(Debug, Clone)]
pub struct StartModuleOperation<'a> {
pub operation_id: &'a str,
pub idempotency_key: &'a str,
pub operation_kind: ModuleOperationKind,
pub plan: &'a ModuleChangePlan,
pub policy: &'a ModuleEnvironmentPolicy,
pub actor: &'a ManagementActor,
pub approvals: Vec<ModuleApproval>,
pub holder_id: &'a str,
pub now: DateTime<Utc>,
}
#[derive(Debug, Clone)]
pub struct AdvanceModuleOperation {
pub operation_id: String,
pub expected_revision: u64,
pub fencing_token: u64,
pub next_state: ModuleOperationState,
pub actor_id: String,
pub outcome_code: String,
pub evidence_references: Vec<ArtifactReference>,
pub error: Option<ModuleOperationError>,
pub next_actions: Vec<String>,
pub now: DateTime<Utc>,
}
#[derive(Debug, Error)]
pub enum ModuleManagementError {
#[error("management contract is invalid: {0}")]
InvalidContract(String),
#[error("environment management is disabled or read-only")]
EnvironmentNotWritable,
#[error("actor lacks `{0}` authority")]
MissingAuthority(String),
#[error("approval boundary `{0}` is not satisfied")]
ApprovalMissing(String),
#[error("approval `{0}` is stale or does not bind the exact plan")]
ApprovalStale(String),
#[error("idempotency key is already bound to another plan")]
IdempotencyConflict,
#[error("application composition lease is held by `{0}`")]
LeaseHeld(String),
#[error("operation fencing token is stale")]
StaleFencingToken,
#[error("operation transition from {from:?} to {to:?} is not legal")]
IllegalTransition {
from: ModuleOperationState,
to: ModuleOperationState,
},
#[error("effect receipt conflicts with retained evidence for `{0}`")]
ReceiptConflict(String),
#[error(transparent)]
Store(#[from] ModuleOperationStoreError),
}
#[derive(Debug)]
pub struct ModuleManagementEngine<S> {
store: S,
}
impl<S> ModuleManagementEngine<S>
where
S: ModuleOperationStore,
{
pub fn new(store: S) -> Self {
Self { store }
}
pub fn store(&self) -> &S {
&self.store
}
pub fn start(
&self,
request: StartModuleOperation<'_>,
) -> Result<ModuleOperation, ModuleManagementError> {
validate_change_plan(request.plan)?;
if request.policy.mode != EnvironmentManagementMode::Full {
return Err(ModuleManagementError::EnvironmentNotWritable);
}
if request.policy.protocol != MODULE_ENVIRONMENT_POLICY_PROTOCOL {
return invalid("unsupported Module Environment Policy protocol");
}
require_authority(request.actor, "module.manage")?;
if request
.plan
.effects
.iter()
.any(|effect| matches!(effect, ModulePlanEffect::ServiceInstallation { .. }))
{
require_authority(request.actor, "service.manage")?;
}
if let Some(existing) = self
.store
.find_by_idempotency_key(&request.plan.application_id, request.idempotency_key)?
{
return if existing.plan_digest == request.plan.plan_digest {
Ok(existing)
} else {
Err(ModuleManagementError::IdempotencyConflict)
};
}
let approvals_satisfied = approvals_satisfied(
request.plan,
request.policy,
request.actor,
&request.approvals,
request.now,
)?;
let initial_state = if approvals_satisfied {
ModuleOperationState::Ready
} else {
ModuleOperationState::AwaitingApproval
};
let fencing_token = if initial_state == ModuleOperationState::Ready {
self.acquire_lease(
&request.plan.application_id,
request.holder_id,
request.policy.maximum_lease_seconds,
request.now,
)?
.fencing_token
} else {
0
};
let operation = ModuleOperation {
protocol: MODULE_OPERATION_PROTOCOL.to_owned(),
operation_id: request.operation_id.to_owned(),
idempotency_key: request.idempotency_key.to_owned(),
application_id: request.plan.application_id.clone(),
environment_id: request.plan.environment_id.clone(),
plan_digest: request.plan.plan_digest.clone(),
operation_kind: request.operation_kind,
expected_target_revision: request.plan.expected_target_revision,
current_lock_digest: request.plan.current_lock_digest.clone(),
target_lock_digest: request.plan.target_lock_digest.clone(),
actor_id: request.actor.actor_id.clone(),
verified_authorities: request.actor.verified_authorities.iter().cloned().collect(),
policy_revision: request.policy.revision.clone(),
state: initial_state,
revision: 0,
fencing_token,
attempt: 1,
approvals: request.approvals,
effect_receipts: Vec::new(),
workspace_backups: Vec::new(),
errors: Vec::new(),
next_actions: if initial_state == ModuleOperationState::AwaitingApproval {
vec!["obtain_plan_bound_approvals".to_owned()]
} else {
vec!["apply_reviewed_plan".to_owned()]
},
repair_operation_id: None,
updated_at: request.now,
};
let event = initial_event(&operation);
match self.store.create_idempotent(&operation, &event)? {
CreateOperationResult::Created => Ok(operation),
CreateOperationResult::Existing(existing)
if existing.plan_digest == request.plan.plan_digest =>
{
Ok(*existing)
}
CreateOperationResult::Existing(_) => Err(ModuleManagementError::IdempotencyConflict),
}
}
#[allow(clippy::too_many_arguments)]
pub fn submit_approval(
&self,
operation_id: &str,
expected_revision: u64,
plan: &ModuleChangePlan,
policy: &ModuleEnvironmentPolicy,
requester: &ManagementActor,
approval: ModuleApproval,
holder_id: &str,
now: DateTime<Utc>,
) -> Result<ModuleOperation, ModuleManagementError> {
validate_change_plan(plan)?;
let current = self.store.load(operation_id)?;
if current.revision != expected_revision {
return Err(ModuleOperationStoreError::RevisionConflict {
operation_id: current.operation_id,
expected: expected_revision,
observed: current.revision,
}
.into());
}
if current.state != ModuleOperationState::AwaitingApproval
|| current.plan_digest != plan.plan_digest
{
return Err(ModuleManagementError::IllegalTransition {
from: current.state,
to: ModuleOperationState::Ready,
});
}
let mut approvals = current.approvals.clone();
approvals.retain(|existing| existing.boundary_id != approval.boundary_id);
approvals.push(approval);
approvals.sort_by(|left, right| left.boundary_id.cmp(&right.boundary_id));
let ready = approvals_satisfied(plan, policy, requester, &approvals, now)?;
let mut updated = current.clone();
updated.approvals = approvals;
updated.revision = updated.revision.saturating_add(1);
updated.updated_at = now;
if ready {
let lease = self.acquire_lease(
¤t.application_id,
holder_id,
policy.maximum_lease_seconds,
now,
)?;
updated.state = ModuleOperationState::Ready;
updated.fencing_token = lease.fencing_token;
updated.next_actions = vec!["apply_reviewed_plan".to_owned()];
} else {
updated.next_actions = vec!["obtain_plan_bound_approvals".to_owned()];
}
let event = next_event(
¤t,
updated,
requester.actor_id.clone(),
"approval_recorded".to_owned(),
Vec::new(),
now,
self.store.journal(operation_id)?.events.last(),
)?;
self.store.compare_and_append(expected_revision, &event)?;
Ok(event.operation_after)
}
pub fn advance(
&self,
request: AdvanceModuleOperation,
) -> Result<ModuleOperation, ModuleManagementError> {
let current = self.store.load(&request.operation_id)?;
if current.revision != request.expected_revision {
return Err(ModuleOperationStoreError::RevisionConflict {
operation_id: current.operation_id,
expected: request.expected_revision,
observed: current.revision,
}
.into());
}
if current.fencing_token != request.fencing_token {
return Err(ModuleManagementError::StaleFencingToken);
}
if !legal_transition(current.state, request.next_state) {
return Err(ModuleManagementError::IllegalTransition {
from: current.state,
to: request.next_state,
});
}
let mut updated = current.clone();
updated.state = request.next_state;
updated.revision = updated.revision.saturating_add(1);
updated.updated_at = request.now;
updated.next_actions = request.next_actions;
if let Some(error) = request.error {
updated.errors.push(error);
}
let event = next_event(
¤t,
updated,
request.actor_id,
request.outcome_code,
request.evidence_references,
request.now,
self.store.journal(&request.operation_id)?.events.last(),
)?;
self.store
.compare_and_append(request.expected_revision, &event)?;
Ok(event.operation_after)
}
#[allow(clippy::too_many_arguments)]
pub fn retry_blocked(
&self,
operation_id: &str,
expected_revision: u64,
next_state: ModuleOperationState,
holder_id: &str,
maximum_lease_seconds: u64,
actor_id: &str,
now: DateTime<Utc>,
) -> Result<ModuleOperation, ModuleManagementError> {
let current = self.store.load(operation_id)?;
if current.revision != expected_revision {
return Err(ModuleOperationStoreError::RevisionConflict {
operation_id: current.operation_id,
expected: expected_revision,
observed: current.revision,
}
.into());
}
if current.state != ModuleOperationState::Blocked
|| !matches!(
next_state,
ModuleOperationState::StagingConfiguration
| ModuleOperationState::Migrating
| ModuleOperationState::Verifying
| ModuleOperationState::Activating
)
{
return Err(ModuleManagementError::IllegalTransition {
from: current.state,
to: next_state,
});
}
let lease = self.acquire_lease(
¤t.application_id,
holder_id,
maximum_lease_seconds,
now,
)?;
let mut updated = current.clone();
updated.state = next_state;
updated.revision = updated.revision.saturating_add(1);
updated.attempt = updated.attempt.saturating_add(1);
updated.fencing_token = lease.fencing_token;
updated.updated_at = now;
updated.next_actions = vec!["retry_incomplete_effects".to_owned()];
let event = next_event(
¤t,
updated,
actor_id.to_owned(),
"blocked_operation_retry_started".to_owned(),
Vec::new(),
now,
self.store.journal(operation_id)?.events.last(),
)?;
self.store.compare_and_append(expected_revision, &event)?;
Ok(event.operation_after)
}
#[allow(clippy::too_many_arguments)]
pub fn begin_workspace_application(
&self,
operation_id: &str,
expected_revision: u64,
fencing_token: u64,
actor_id: &str,
plan: &ModuleChangePlan,
backups: Vec<ModuleWorkspaceBackup>,
evidence_references: Vec<ArtifactReference>,
now: DateTime<Utc>,
) -> Result<ModuleOperation, ModuleManagementError> {
validate_change_plan(plan)?;
let current = self.store.load(operation_id)?;
if current.revision != expected_revision {
return Err(ModuleOperationStoreError::RevisionConflict {
operation_id: current.operation_id,
expected: expected_revision,
observed: current.revision,
}
.into());
}
if current.fencing_token != fencing_token {
return Err(ModuleManagementError::StaleFencingToken);
}
if current.plan_digest != plan.plan_digest {
return Err(ModuleManagementError::IdempotencyConflict);
}
if current.state != ModuleOperationState::Ready {
return Err(ModuleManagementError::IllegalTransition {
from: current.state,
to: ModuleOperationState::ApplyingFiles,
});
}
require_sorted_unique(
backups.iter().map(|backup| backup.path.as_str()),
"workspace backups",
)?;
let backup_paths = backups
.iter()
.map(|backup| backup.path.as_str())
.collect::<BTreeSet<_>>();
if plan.effects.iter().any(|effect| {
matches!(effect, ModulePlanEffect::WorkspaceFile { path, .. } if !backup_paths.contains(path.as_str()))
}) {
return invalid("workspace backups must cover every planned file effect");
}
for backup in &backups {
validate_workspace_backup(backup)?;
}
let mut updated = current.clone();
updated.state = ModuleOperationState::ApplyingFiles;
updated.revision = updated.revision.saturating_add(1);
updated.updated_at = now;
updated.workspace_backups = backups;
updated.next_actions = vec!["apply_next_guarded_file".to_owned()];
let event = next_event(
¤t,
updated,
actor_id.to_owned(),
"workspace_backups_recorded".to_owned(),
evidence_references,
now,
self.store.journal(operation_id)?.events.last(),
)?;
self.store.compare_and_append(expected_revision, &event)?;
Ok(event.operation_after)
}
pub fn record_effect_receipt(
&self,
operation_id: &str,
expected_revision: u64,
fencing_token: u64,
actor_id: &str,
receipt: ModuleEffectReceipt,
now: DateTime<Utc>,
) -> Result<ModuleOperation, ModuleManagementError> {
let current = self.store.load(operation_id)?;
if current.revision != expected_revision {
return Err(ModuleOperationStoreError::RevisionConflict {
operation_id: current.operation_id,
expected: expected_revision,
observed: current.revision,
}
.into());
}
if current.fencing_token != fencing_token || receipt.fencing_token != fencing_token {
return Err(ModuleManagementError::StaleFencingToken);
}
if !matches!(
current.state,
ModuleOperationState::ApplyingFiles
| ModuleOperationState::FilesApplied
| ModuleOperationState::StagingConfiguration
| ModuleOperationState::Migrating
| ModuleOperationState::Verifying
| ModuleOperationState::Activating
) {
return invalid("effect receipts are only accepted while effects are being applied");
}
if receipt.operation_id != current.operation_id || receipt.attempt != current.attempt {
return Err(ModuleManagementError::ReceiptConflict(
receipt.effect_id.clone(),
));
}
if let Some(existing) = current
.effect_receipts
.iter()
.find(|existing| existing.effect_id == receipt.effect_id)
{
return if existing == &receipt {
Ok(current)
} else {
Err(ModuleManagementError::ReceiptConflict(receipt.effect_id))
};
}
let mut updated = current.clone();
updated.revision = updated.revision.saturating_add(1);
updated.updated_at = now;
updated.effect_receipts.push(receipt.clone());
updated
.effect_receipts
.sort_by(|left, right| left.effect_id.cmp(&right.effect_id));
let event = next_event(
¤t,
updated,
actor_id.to_owned(),
"effect_receipt_recorded".to_owned(),
receipt.evidence_references.clone(),
now,
self.store.journal(operation_id)?.events.last(),
)?;
self.store.compare_and_append(expected_revision, &event)?;
Ok(event.operation_after)
}
#[allow(clippy::too_many_arguments)]
pub fn resume_after_crash(
&self,
operation_id: &str,
expected_revision: u64,
evidence: &ModuleResumeEvidence,
holder_id: &str,
maximum_lease_seconds: u64,
actor_id: &str,
now: DateTime<Utc>,
) -> Result<ModuleOperation, ModuleManagementError> {
let current = self.store.load(operation_id)?;
if current.revision != expected_revision {
return Err(ModuleOperationStoreError::RevisionConflict {
operation_id: current.operation_id,
expected: expected_revision,
observed: current.revision,
}
.into());
}
if !matches!(
current.state,
ModuleOperationState::ApplyingFiles
| ModuleOperationState::FilesApplied
| ModuleOperationState::StagingConfiguration
| ModuleOperationState::Migrating
| ModuleOperationState::Verifying
| ModuleOperationState::Activating
) || evidence.plan_digest != current.plan_digest
|| !evidence.next_effect_idempotent
|| evidence.observed_target_digest.is_empty()
{
return Err(ModuleManagementError::InvalidContract(
"crash continuation is not proven safe".to_owned(),
));
}
let completed = current
.effect_receipts
.iter()
.map(|receipt| receipt.effect_id.as_str())
.collect::<Vec<_>>();
if completed
!= evidence
.completed_effect_ids
.iter()
.map(String::as_str)
.collect::<Vec<_>>()
|| completed.contains(&evidence.next_effect_id.as_str())
{
return Err(ModuleManagementError::InvalidContract(
"resume evidence does not match completed effect receipts".to_owned(),
));
}
let lease = self.acquire_lease(
¤t.application_id,
holder_id,
maximum_lease_seconds,
now,
)?;
let mut updated = current.clone();
updated.revision = updated.revision.saturating_add(1);
updated.attempt = updated.attempt.saturating_add(1);
updated.fencing_token = lease.fencing_token;
updated.updated_at = now;
updated.next_actions = vec![format!("resume_effect:{}", evidence.next_effect_id)];
let event = next_event(
¤t,
updated,
actor_id.to_owned(),
"crash_recovery_attempt_started".to_owned(),
Vec::new(),
now,
self.store.journal(operation_id)?.events.last(),
)?;
self.store.compare_and_append(expected_revision, &event)?;
Ok(event.operation_after)
}
#[allow(clippy::too_many_arguments)]
pub fn reconcile_with_succeeded_repair(
&self,
operation_id: &str,
expected_revision: u64,
repair_plan: &ModuleRepairPlan,
repair_change_plan: &ModuleChangePlan,
repair_operation_id: &str,
actor_id: &str,
evidence_references: Vec<ArtifactReference>,
now: DateTime<Utc>,
) -> Result<ModuleOperation, ModuleManagementError> {
validate_repair_plan(repair_plan)?;
validate_change_plan(repair_change_plan)?;
let current = self.store.load(operation_id)?;
if current.revision != expected_revision {
return Err(ModuleOperationStoreError::RevisionConflict {
operation_id: current.operation_id,
expected: expected_revision,
observed: current.revision,
}
.into());
}
let repair_digest_matches = matches!(
&repair_change_plan.request,
ModuleRootChange::Repair { repair_plan_digest }
if repair_plan_digest == &repair_plan.repair_plan_digest
);
if current.state != ModuleOperationState::RepairRequired
|| repair_plan.original_operation_id != current.operation_id
|| repair_plan.original_operation_revision != current.revision
|| repair_plan.application_id != current.application_id
|| repair_plan.environment_id != current.environment_id
|| repair_change_plan.application_id != current.application_id
|| repair_change_plan.environment_id != current.environment_id
|| !repair_digest_matches
{
return invalid("repair plan does not bind the exact repair-required operation");
}
let repair_operation = self.store.load(repair_operation_id)?;
if repair_operation.operation_kind != ModuleOperationKind::Repair
|| repair_operation.state != ModuleOperationState::Succeeded
|| repair_operation.plan_digest != repair_change_plan.plan_digest
|| repair_operation.application_id != current.application_id
|| repair_operation.environment_id != current.environment_id
{
return invalid("reconciliation requires the bound repair operation to succeed");
}
let mut updated = current.clone();
updated.state = ModuleOperationState::Reconciled;
updated.revision = updated.revision.saturating_add(1);
updated.updated_at = now;
updated.next_actions.clear();
updated.repair_operation_id = Some(repair_operation.operation_id);
let event = next_event(
¤t,
updated,
actor_id.to_owned(),
"repair_reconciled".to_owned(),
evidence_references,
now,
self.store.journal(operation_id)?.events.last(),
)?;
self.store.compare_and_append(expected_revision, &event)?;
Ok(event.operation_after)
}
pub fn acquire_lease(
&self,
application_id: &str,
holder_id: &str,
maximum_lease_seconds: u64,
now: DateTime<Utc>,
) -> Result<ModuleOperationLease, ModuleManagementError> {
let current = self.store.load_lease()?;
if let Some(lease) = ¤t
&& lease.expires_at > now
&& (lease.application_id != application_id || lease.holder_id != holder_id)
{
return Err(ModuleManagementError::LeaseHeld(lease.holder_id.clone()));
}
let duration = i64::try_from(maximum_lease_seconds).map_err(|_| {
ModuleManagementError::InvalidContract("lease duration overflows".into())
})?;
let lease = ModuleOperationLease {
application_id: application_id.to_owned(),
holder_id: holder_id.to_owned(),
fencing_token: current.as_ref().map_or(1, |lease| {
if lease.application_id == application_id
&& lease.holder_id == holder_id
&& lease.expires_at > now
{
lease.fencing_token
} else {
lease.fencing_token.saturating_add(1)
}
}),
revision: current
.as_ref()
.map_or(0, |lease| lease.revision.saturating_add(1)),
acquired_at: now,
expires_at: now + Duration::seconds(duration),
};
self.store
.compare_and_set_lease(current.as_ref().map(|lease| lease.revision), Some(&lease))?;
Ok(lease)
}
pub fn release_lease(
&self,
holder_id: &str,
fencing_token: u64,
) -> Result<(), ModuleManagementError> {
let current = self.store.load_lease()?;
let Some(lease) = current else {
return Ok(());
};
if lease.holder_id != holder_id || lease.fencing_token != fencing_token {
return Err(ModuleManagementError::StaleFencingToken);
}
self.store
.compare_and_set_lease(Some(lease.revision), None)?;
Ok(())
}
}
fn validate_workspace_backup(backup: &ModuleWorkspaceBackup) -> Result<(), ModuleManagementError> {
if !safe_relative_path(&backup.path) {
return invalid("workspace backup path is unsafe");
}
let valid = match (backup.existence, backup.file_type) {
(crate::PathExistence::Absent, crate::ManagedFileType::Absent)
| (crate::PathExistence::Present, crate::ManagedFileType::Directory) => {
backup.content_base64.is_none() && backup.content_digest.is_none()
}
(crate::PathExistence::Present, crate::ManagedFileType::Regular) => {
let Some(encoded) = &backup.content_base64 else {
return invalid("regular-file backup has no exact bytes");
};
let Some(digest) = &backup.content_digest else {
return invalid("regular-file backup has no content digest");
};
BASE64
.decode(encoded)
.is_ok_and(|bytes| raw_digest(&bytes) == *digest)
}
_ => false,
};
if valid {
Ok(())
} else {
invalid("workspace backup shape or digest is invalid")
}
}
pub fn validate_change_plan(plan: &ModuleChangePlan) -> Result<(), ModuleManagementError> {
if plan.protocol != MODULE_CHANGE_PLAN_PROTOCOL {
return invalid("unsupported Module Change Plan protocol");
}
if module_change_plan_digest(plan).map_err(json_error)? != plan.plan_digest {
return invalid("Module Change Plan digest mismatch");
}
validate_desired_composition(&plan.target_desired)?;
if desired_composition_digest(&plan.target_desired).map_err(json_error)?
!= plan.target_desired_digest
|| plan.target_desired.application_id != plan.application_id
{
return invalid("target Desired Composition identity or digest mismatch");
}
validate_application_module_lock(&plan.target_lock)?;
if application_module_lock_digest(&plan.target_lock).map_err(json_error)?
!= plan.target_lock_digest
|| plan.target_lock.application_id != plan.application_id
|| plan.target_lock.desired_composition_digest != plan.target_desired_digest
|| plan.target_lock.catalog_snapshot_digest != plan.catalog_snapshot_digest
|| plan.target_lock.trust_policy_digest != plan.trust_policy_digest
{
return invalid("target Application Module Lock binding or digest mismatch");
}
require_sorted_unique(
plan.read_set.iter().map(|entry| entry.path.as_str()),
"read set",
)?;
require_sorted_unique(
plan.effects.iter().map(ModulePlanEffect::effect_id),
"effect ids",
)?;
if plan.validation_commands.is_empty()
|| plan
.read_set
.iter()
.any(|entry| entry.path == ".env" || entry.path.ends_with("/.env"))
{
return invalid("plans require validation commands and must never target .env");
}
for boundary in &plan.approval_boundaries {
validate_boundary(boundary, &plan.effects)?;
}
for effect in &plan.effects {
match effect {
ModulePlanEffect::WorkspaceFile {
path,
change,
before_digest,
after_digest,
after_content,
after_mode,
patch,
..
} => validate_workspace_effect(
path,
*change,
before_digest.as_deref(),
after_digest.as_deref(),
after_content.as_deref(),
*after_mode,
patch,
&plan.read_set,
)?,
ModulePlanEffect::Migration {
artifact_locator,
artifact_digest,
..
} => {
if !safe_relative_path(artifact_locator) || !valid_digest(artifact_digest) {
return invalid(
"migration artifacts require a safe locator and SHA-256 digest",
);
}
}
ModulePlanEffect::ServiceInstallation {
service_id,
service_release_digest,
installation_plan,
adapter,
action,
..
} => {
validate_service_effect(
service_id,
service_release_digest,
*adapter,
action.as_ref(),
)?;
if let Some(installation_plan) = installation_plan {
crate::validate_service_installation_plan(installation_plan).map_err(
|error| ModuleManagementError::InvalidContract(error.to_string()),
)?;
let installation = match &installation_plan.change {
crate::ServiceInstallationChange::Install { installation } => installation,
crate::ServiceInstallationChange::Uninstall { .. } => {
return invalid(
"Module Service installation effect cannot contain an uninstall plan",
);
}
};
if installation_plan.environment_id != plan.environment_id
|| installation.service_ref.service_id != *service_id
|| installation.service_release.digest != *service_release_digest
{
return invalid(
"nested Service Installation Plan differs from the Module plan",
);
}
}
}
ModulePlanEffect::ServiceRemoval {
service_id,
service_release_digest,
adapter,
action,
..
}
| ModulePlanEffect::ServiceRestart {
service_id,
service_release_digest,
adapter,
action,
..
} => {
validate_service_effect(
service_id,
service_release_digest,
*adapter,
action.as_ref(),
)?;
}
_ => {}
}
}
Ok(())
}
fn validate_service_effect(
service_id: &str,
release_digest: &str,
adapter: Option<crate::ServiceDeploymentAdapterKind>,
action: Option<&crate::ServiceDeploymentAction>,
) -> Result<(), ModuleManagementError> {
if service_id.trim().is_empty() || !valid_digest(release_digest) {
return invalid("service effects require a Service identity and release digest");
}
if adapter.is_none() != action.is_none() {
return invalid("service effects must bind adapter and action together");
}
let Some(action) = action else {
return Ok(());
};
match action {
crate::ServiceDeploymentAction::Command {
program,
args,
working_directory,
} => {
let executable = program.rsplit(['/', '\\']).next().unwrap_or_default();
if program.trim().is_empty()
|| program.contains('\0')
|| matches!(
executable.to_ascii_lowercase().as_str(),
"sh" | "bash"
| "dash"
| "zsh"
| "fish"
| "cmd"
| "cmd.exe"
| "powershell"
| "powershell.exe"
| "pwsh"
| "pwsh.exe"
)
|| args.iter().any(|argument| argument.contains('\0'))
|| working_directory
.as_deref()
.is_some_and(|path| !safe_relative_path(path))
|| adapter == Some(crate::ServiceDeploymentAdapterKind::ExternallyManaged)
{
return invalid(
"service command action is unsafe or incompatible with its adapter",
);
}
}
crate::ServiceDeploymentAction::Evidence { receipt } => {
if !safe_relative_path(&receipt.locator) || !valid_digest(&receipt.digest) {
return invalid(
"service evidence action requires a safe content-addressed receipt",
);
}
}
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
fn validate_workspace_effect(
path: &str,
change: crate::ModuleFileChange,
before_digest: Option<&str>,
after_digest: Option<&str>,
after_content: Option<&str>,
after_mode: Option<u32>,
exact_patch: &str,
read_set: &[crate::ModulePathPrecondition],
) -> Result<(), ModuleManagementError> {
if !safe_relative_path(path) || exact_patch.trim().is_empty() {
return invalid("workspace effects require a safe relative path and exact patch");
}
let precondition = read_set
.iter()
.find(|precondition| precondition.path == path)
.ok_or_else(|| {
ModuleManagementError::InvalidContract(format!(
"workspace effect `{path}` is absent from the read set"
))
})?;
let shape_matches = match change {
crate::ModuleFileChange::Create => {
precondition.existence == crate::PathExistence::Absent
&& before_digest.is_none()
&& after_digest.is_some()
&& after_content.is_some()
&& after_mode.is_some()
}
crate::ModuleFileChange::Modify => {
precondition.existence == crate::PathExistence::Present
&& precondition.content_digest.as_deref() == before_digest
&& before_digest.is_some()
&& after_digest.is_some()
&& after_content.is_some()
&& after_mode.is_some()
}
crate::ModuleFileChange::Delete => {
precondition.existence == crate::PathExistence::Present
&& precondition.content_digest.as_deref() == before_digest
&& before_digest.is_some()
&& after_digest.is_none()
&& after_content.is_none()
&& after_mode.is_none()
}
};
if !shape_matches
|| before_digest.is_some_and(|digest| !valid_digest(digest))
|| after_digest.is_some_and(|digest| !valid_digest(digest))
|| after_mode.is_some_and(|mode| mode > 0o777)
|| after_content
.is_some_and(|content| Some(raw_digest(content.as_bytes()).as_str()) != after_digest)
{
return invalid("workspace effect content does not match its exact precondition or digest");
}
Ok(())
}
pub fn validate_desired_composition(
composition: &DesiredModuleComposition,
) -> Result<(), ModuleManagementError> {
if composition.protocol != DESIRED_MODULE_COMPOSITION_PROTOCOL
|| composition.application_id.trim().is_empty()
{
return invalid("invalid Desired Module Composition identity or protocol");
}
require_sorted_unique(
composition
.selected
.iter()
.map(|entry| entry.module_id.as_str()),
"selected Module identities",
)?;
for entry in &composition.selected {
if !valid_module_id(&entry.module_id)
|| semver::VersionReq::parse(&entry.version_requirement).is_err()
{
return invalid("selected Module identity or version requirement is invalid");
}
require_sorted_unique(
entry.optional_requirements.iter().map(String::as_str),
"optional requirements",
)?;
}
require_sorted_unique(
composition
.local_overrides
.iter()
.map(|entry| entry.module_id.as_str()),
"local overrides",
)?;
if composition
.local_overrides
.iter()
.any(|entry| !entry.acknowledged_unverified || !valid_digest(&entry.content_digest))
{
return invalid("local overrides must be content-addressed and explicitly unverified");
}
Ok(())
}
pub fn validate_application_module_lock(
module_lock: &ApplicationModuleLock,
) -> Result<(), ModuleManagementError> {
if module_lock.protocol != APPLICATION_MODULE_LOCK_PROTOCOL
|| module_lock.application_id.trim().is_empty()
|| !valid_digest(&module_lock.desired_composition_digest)
|| !valid_digest(&module_lock.catalog_snapshot_digest)
|| !valid_digest(&module_lock.trust_policy_digest)
{
return invalid("invalid Application Module Lock identity, protocol, or input digest");
}
require_sorted_unique(
module_lock
.modules
.iter()
.map(|entry| entry.module_id.as_str()),
"locked Module identities",
)?;
let module_ids = module_lock
.modules
.iter()
.map(|entry| entry.module_id.as_str())
.collect::<BTreeSet<_>>();
for module in &module_lock.modules {
if !valid_module_id(&module.module_id)
|| semver::Version::parse(&module.version).is_err()
|| !valid_digest(&module.release_digest)
|| !valid_digest(&module.manifest_digest)
{
return invalid("locked Module identity, version, or digest is invalid");
}
require_sorted_unique(
module.dependency_module_ids.iter().map(String::as_str),
"locked dependencies",
)?;
if module
.dependency_module_ids
.iter()
.any(|dependency| !module_ids.contains(dependency.as_str()))
{
return invalid("locked dependency references an absent Module");
}
}
Ok(())
}
pub fn validate_repair_plan(plan: &ModuleRepairPlan) -> Result<(), ModuleManagementError> {
if plan.protocol != MODULE_REPAIR_PLAN_PROTOCOL
|| plan.original_operation_id.trim().is_empty()
|| plan.actions.is_empty()
|| module_repair_plan_digest(plan).map_err(json_error)? != plan.repair_plan_digest
{
return invalid("invalid Module Repair Plan protocol, identity, actions, or digest");
}
require_sorted_unique(
plan.completed_effect_ids.iter().map(String::as_str),
"completed repair-plan effects",
)?;
let completed = plan
.completed_effect_ids
.iter()
.map(String::as_str)
.collect::<BTreeSet<_>>();
for action in &plan.actions {
if let ModuleRepairAction::Resume { effect_ids } = action {
require_sorted_unique(
effect_ids.iter().map(String::as_str),
"repair resume effects",
)?;
if effect_ids
.iter()
.any(|effect_id| completed.contains(effect_id.as_str()))
{
return invalid("repair plan cannot repeat a completed effect");
}
}
}
Ok(())
}
fn approvals_satisfied(
plan: &ModuleChangePlan,
policy: &ModuleEnvironmentPolicy,
requester: &ManagementActor,
approvals: &[ModuleApproval],
now: DateTime<Utc>,
) -> Result<bool, ModuleManagementError> {
let mut satisfied = true;
for boundary in &plan.approval_boundaries {
if policy.require_backup_for_non_local_destructive_effects
&& plan.environment_id != "local"
&& matches!(
boundary.risk_class,
ModuleRiskClass::DestructiveMigration
| ModuleRiskClass::DataDeletion
| ModuleRiskClass::BackupRestore
)
&& boundary.backup_evidence_digest.is_none()
{
return Err(ModuleManagementError::ApprovalMissing(format!(
"{}:backup_evidence",
boundary.boundary_id
)));
}
let Some(approval) = approvals
.iter()
.find(|approval| approval.boundary_id == boundary.boundary_id)
else {
satisfied = false;
continue;
};
if approval.protocol != MODULE_APPROVAL_PROTOCOL
|| approval.plan_digest != plan.plan_digest
|| approval.application_id != plan.application_id
|| approval.environment_id != plan.environment_id
|| approval.expected_target_revision != plan.expected_target_revision
|| approval.risk_class != boundary.risk_class
|| approval.expires_at <= now
|| approval.issued_at > now
|| approval.reason.trim().is_empty()
|| approval.nonce.trim().is_empty()
|| !approval
.verified_authorities
.contains(&boundary.required_authority)
|| policy.require_distinct_approver && approval.actor_id == requester.actor_id
{
return Err(ModuleManagementError::ApprovalStale(
approval.approval_id.clone(),
));
}
let age = now.signed_duration_since(approval.issued_at).num_seconds();
if age < 0 || u64::try_from(age).unwrap_or(u64::MAX) > policy.maximum_approval_age_seconds {
return Err(ModuleManagementError::ApprovalStale(
approval.approval_id.clone(),
));
}
}
Ok(satisfied)
}
fn validate_boundary(
boundary: &ModuleApprovalBoundary,
effects: &[ModulePlanEffect],
) -> Result<(), ModuleManagementError> {
if boundary.risk_class == ModuleRiskClass::Ordinary
|| boundary.boundary_id.trim().is_empty()
|| boundary.effect_ids.is_empty()
{
return invalid("approval boundaries must identify non-ordinary protected effects");
}
let expected_authority = match boundary.risk_class {
ModuleRiskClass::Ordinary => unreachable!("ordinary boundaries rejected above"),
ModuleRiskClass::DestructiveMigration => "module.migrate.destructive",
ModuleRiskClass::DataDeletion
| ModuleRiskClass::BackupRestore
| ModuleRiskClass::BackupWaiver => "module.data.delete",
ModuleRiskClass::TrustOverride => "module.trust.override",
};
if boundary.required_authority != expected_authority {
return invalid("approval boundary uses the wrong scoped authority");
}
require_sorted_unique(
boundary.effect_ids.iter().map(String::as_str),
"boundary effects",
)?;
for effect_id in &boundary.effect_ids {
let effect = effects
.iter()
.find(|effect| effect.effect_id() == effect_id)
.ok_or_else(|| {
ModuleManagementError::InvalidContract(format!(
"approval boundary references unknown effect `{effect_id}`"
))
})?;
if effect.risk_class() != boundary.risk_class {
return invalid("approval risk class does not match its protected effect");
}
}
Ok(())
}
fn require_authority(
actor: &ManagementActor,
authority: &str,
) -> Result<(), ModuleManagementError> {
if actor.verified_authorities.contains(authority) {
Ok(())
} else {
Err(ModuleManagementError::MissingAuthority(
authority.to_owned(),
))
}
}
fn initial_event(operation: &ModuleOperation) -> ModuleOperationJournalEvent {
ModuleOperationJournalEvent {
protocol: MODULE_OPERATION_JOURNAL_PROTOCOL.to_owned(),
event_id: format!("{}-0", operation.operation_id),
operation_id: operation.operation_id.clone(),
attempt: operation.attempt,
revision: 0,
prior_event_digest: None,
plan_digest: operation.plan_digest.clone(),
actor_id: operation.actor_id.clone(),
fencing_token: operation.fencing_token,
transition: ModuleOperationTransition {
from: ModuleOperationState::Planned,
to: operation.state,
},
outcome_code: "operation_created".to_owned(),
evidence_references: Vec::new(),
operation_after: operation.clone(),
recorded_at: operation.updated_at,
}
}
#[allow(clippy::too_many_arguments)]
fn next_event(
before: &ModuleOperation,
after: ModuleOperation,
actor_id: String,
outcome_code: String,
evidence_references: Vec<ArtifactReference>,
now: DateTime<Utc>,
prior: Option<&ModuleOperationJournalEvent>,
) -> Result<ModuleOperationJournalEvent, ModuleManagementError> {
let prior_event_digest = prior
.map(journal_event_digest)
.transpose()
.map_err(json_error)?;
Ok(ModuleOperationJournalEvent {
protocol: MODULE_OPERATION_JOURNAL_PROTOCOL.to_owned(),
event_id: format!("{}-{}", after.operation_id, after.revision),
operation_id: after.operation_id.clone(),
attempt: after.attempt,
revision: after.revision,
prior_event_digest,
plan_digest: after.plan_digest.clone(),
actor_id,
fencing_token: after.fencing_token,
transition: ModuleOperationTransition {
from: before.state,
to: after.state,
},
outcome_code,
evidence_references,
operation_after: after,
recorded_at: now,
})
}
fn legal_transition(from: ModuleOperationState, to: ModuleOperationState) -> bool {
use ModuleOperationState as State;
matches!(
(from, to),
(
State::Planned,
State::AwaitingApproval | State::Ready | State::Blocked | State::Cancelled
) | (
State::AwaitingApproval,
State::Ready | State::Blocked | State::Cancelled
) | (
State::Ready,
State::ApplyingFiles | State::Blocked | State::Cancelled
) | (
State::ApplyingFiles,
State::FilesApplied | State::Restored | State::RepairRequired
) | (
State::FilesApplied,
State::StagingConfiguration
| State::Migrating
| State::Verifying
| State::Blocked
| State::RepairRequired
) | (
State::StagingConfiguration,
State::Migrating | State::Verifying | State::Blocked | State::RepairRequired
) | (
State::Migrating,
State::Verifying | State::Blocked | State::RepairRequired
) | (
State::Verifying,
State::Activating | State::Blocked | State::RepairRequired
) | (
State::Activating,
State::Succeeded | State::Blocked | State::RepairRequired
)
)
}
fn require_sorted_unique<'a>(
values: impl Iterator<Item = &'a str>,
subject: &str,
) -> Result<(), ModuleManagementError> {
let values = values.collect::<Vec<_>>();
if values.windows(2).all(|pair| pair[0] < pair[1]) {
Ok(())
} else {
invalid(format!("{subject} must be sorted and unique"))
}
}
fn valid_module_id(value: &str) -> bool {
value.split_once('/').is_some_and(|(namespace, name)| {
valid_identifier(namespace) && valid_identifier(name) && !name.contains('/')
})
}
fn valid_identifier(value: &str) -> bool {
!value.is_empty()
&& value.bytes().all(|byte| {
byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'-' | b'_')
})
&& value
.bytes()
.next()
.is_some_and(|byte| byte.is_ascii_lowercase())
}
fn valid_digest(value: &str) -> bool {
value.strip_prefix("sha256:").is_some_and(|hex| {
hex.len() == 64
&& hex
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
})
}
fn safe_relative_path(value: &str) -> bool {
!value.is_empty()
&& !value.starts_with('/')
&& !value.contains('\\')
&& value
.split('/')
.all(|segment| !segment.is_empty() && segment != "." && segment != "..")
&& value != ".env"
&& !value.ends_with("/.env")
}
fn raw_digest(bytes: &[u8]) -> String {
const HEX: &[u8; 16] = b"0123456789abcdef";
let digest = Sha256::digest(bytes);
let mut hex = String::with_capacity(digest.len() * 2);
for byte in digest {
hex.push(char::from(HEX[usize::from(byte >> 4)]));
hex.push(char::from(HEX[usize::from(byte & 0x0f)]));
}
format!("sha256:{hex}")
}
fn invalid<T>(message: impl Into<String>) -> Result<T, ModuleManagementError> {
Err(ModuleManagementError::InvalidContract(message.into()))
}
#[allow(clippy::needless_pass_by_value)]
fn json_error(error: serde_json::Error) -> ModuleManagementError {
ModuleManagementError::InvalidContract(error.to_string())
}