use std::collections::BTreeSet;
use base64::{engine::general_purpose::STANDARD, Engine as _};
use chio_core::capability::scope::MonetaryAmount;
use chio_core::crypto::{Keypair, PublicKey, Signature};
use chio_core::receipt::body::ChioReceipt;
use chio_credit::obligation::{
ObligationAtomV1, ObligationDispositionRecordV1, ObligationDispositionTransitionV1,
ObligationDispositionV1,
};
use chio_settle::channel::derive_channel_receipt_authority_digest;
use serde::{Deserialize, Serialize};
use crate::receipt_store::{AuthorizationReceiptConsumption, PendingSettlementObservation};
use crate::tool_outcome::{VerifiedPreDispatchNoEffect, VerifiedTransportNotAccepted};
use super::*;
const SIGNED_TERMINAL_PROJECTION_SCHEMA: &str = "chio.signed-admission-terminal-projection.v1";
const SIGNED_TERMINAL_PROJECTION_DOMAIN: &[u8] = b"chio.signed-admission-terminal-projection.v1\0";
pub const MAX_ADMISSION_TERMINAL_PROJECTION_BYTES: usize = 4 * 1024 * 1024;
pub const MAX_ADMISSION_TERMINAL_MANIFEST_BYTES: usize = 256 * 1024;
pub const MAX_ADMISSION_TERMINAL_RECORD_BYTES: usize = 1024 * 1024;
pub const MAX_ADMISSION_TERMINAL_RECORDS: usize = 16;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PersistedAdmissionTerminalProjectionRecordV1 {
pub kind: AdmissionProjectionRecordKind,
pub record_id: AdmissionIdentifier,
pub record_digest: AdmissionDigest,
pub canonical_json: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AdmissionTerminalObserverProjectionV1 {
pub receipt_id: AdmissionIdentifier,
pub pending: PendingSettlementObservation,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SignedAdmissionTerminalProjectionBodyV1 {
schema: String,
signer_key: PublicKey,
context: AdmissionProjectionContext,
source_operation: PersistedAdmissionOperationV1,
terminal_operation: PersistedAdmissionOperationV1,
projection_json: String,
manifest_json: String,
records: Vec<PersistedAdmissionTerminalProjectionRecordV1>,
authorization_consumption: Option<AuthorizationReceiptConsumption>,
observer: Option<AdmissionTerminalObserverProjectionV1>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SignedAdmissionTerminalProjectionV1 {
body: SignedAdmissionTerminalProjectionBodyV1,
signature: Signature,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VerifiedAdmissionTerminalProjectionRecordV1 {
kind: AdmissionProjectionRecordKind,
record_id: AdmissionIdentifier,
record_digest: AdmissionDigest,
canonical_json: Vec<u8>,
}
impl VerifiedAdmissionTerminalProjectionRecordV1 {
#[must_use]
pub const fn kind(&self) -> AdmissionProjectionRecordKind {
self.kind
}
#[must_use]
pub const fn record_id(&self) -> &AdmissionIdentifier {
&self.record_id
}
#[must_use]
pub const fn record_digest(&self) -> &AdmissionDigest {
&self.record_digest
}
#[must_use]
pub fn canonical_json(&self) -> &[u8] {
&self.canonical_json
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VerifiedAdmissionTerminalProjectionV1 {
signer_key: PublicKey,
context: AdmissionProjectionContext,
source_operation: AdmissionOperationV1,
terminal_operation: AdmissionOperationV1,
projection_json: Vec<u8>,
manifest_json: Vec<u8>,
records: Vec<VerifiedAdmissionTerminalProjectionRecordV1>,
channel_terminal: Option<VerifiedChannelTerminalProjectionV1>,
anchored_transport_cancellation: bool,
anchored_economic_mutation: bool,
pre_dispatch_release_proof: Option<VerifiedPreDispatchNoEffect>,
authorization_consumption: Option<AuthorizationReceiptConsumption>,
observer: Option<AdmissionTerminalObserverProjectionV1>,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct UntrustedTerminalObligationSourceV1 {
binding: AdmissionExactProjectionBindingV1,
source_authority_digest: AdmissionDigest,
source_record_id: AdmissionIdentifier,
source_record_digest: AdmissionDigest,
source_recorded_at_unix_ms: u64,
consumer_receipt_id: AdmissionIdentifier,
consumer_receipt_digest: AdmissionDigest,
outcome_id: AdmissionDigest,
outcome_version: u64,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct UntrustedTerminalObligationProjectionV1 {
source: UntrustedTerminalObligationSourceV1,
atom: ObligationAtomV1,
disposition_record: ObligationDispositionRecordV1,
}
impl SignedAdmissionTerminalProjectionV1 {
pub fn from_verified(
source_operation: &AdmissionOperationV1,
projection: &AdmissionTerminalProjection,
capabilities: &AdmissionProjectionCapabilities,
signer: &Keypair,
) -> Result<Self, AdmissionOperationError> {
let canonical = projection.canonical_projection()?;
let terminal_operation =
source_operation.apply_terminal_projection(projection, capabilities)?;
let (authorization_consumption, observer) = match projection {
AdmissionTerminalProjection::Completed(completed) => (
completed
.authorization
.as_ref()
.map(|authorization| authorization.consumption().clone()),
completed
.observer_work
.as_ref()
.map(|observer| {
Ok(AdmissionTerminalObserverProjectionV1 {
receipt_id: AdmissionIdentifier::try_new(
"observer_receipt_id",
completed.receipt.receipt().id.clone(),
)?,
pending: *observer.pending(),
})
})
.transpose()?,
),
_ => (None, None),
};
let records = canonical
.records()
.iter()
.map(|record| PersistedAdmissionTerminalProjectionRecordV1 {
kind: record.commitment().kind(),
record_id: record.commitment().record_id().clone(),
record_digest: record.commitment().record_digest().clone(),
canonical_json: STANDARD.encode(record.canonical_bytes()),
})
.collect();
let body = SignedAdmissionTerminalProjectionBodyV1 {
schema: SIGNED_TERMINAL_PROJECTION_SCHEMA.to_owned(),
signer_key: signer.public_key(),
context: projection.context().clone(),
source_operation: source_operation.to_persisted(),
terminal_operation: terminal_operation.to_persisted(),
projection_json: STANDARD.encode(canonical.projection_bytes()),
manifest_json: STANDARD.encode(canonical.manifest_bytes()),
records,
authorization_consumption,
observer,
};
let signature = signer.sign(&signing_preimage(&body)?);
let envelope = Self { body, signature };
envelope.verify()?;
Ok(envelope)
}
pub fn verify(&self) -> Result<VerifiedAdmissionTerminalProjectionV1, AdmissionOperationError> {
let mismatch = || AdmissionOperationError::TerminalProjectionBindingMismatch;
if self.body.schema != SIGNED_TERMINAL_PROJECTION_SCHEMA
|| !self
.body
.signer_key
.verify(&signing_preimage(&self.body)?, &self.signature)
{
return Err(mismatch());
}
self.body.context.validate()?;
let source_operation =
AdmissionOperationV1::from_persisted(self.body.source_operation.clone())?;
let terminal_operation =
AdmissionOperationV1::from_persisted(self.body.terminal_operation.clone())?;
let projection_json = decode_bounded(
&self.body.projection_json,
MAX_ADMISSION_TERMINAL_PROJECTION_BYTES,
)?;
let manifest_json = decode_bounded(
&self.body.manifest_json,
MAX_ADMISSION_TERMINAL_MANIFEST_BYTES,
)?;
let manifest = AdmissionProjectionManifestV1::from_canonical_bytes(&manifest_json)?;
manifest.verify_projection_body(&projection_json)?;
let projection_digest = manifest.projection_digest()?;
validate_terminal_successor(
&source_operation,
&terminal_operation,
&self.body.context,
&projection_digest,
)?;
validate_projection_body(
&projection_json,
&self.body.context,
terminal_operation.state(),
)?;
let records = validate_records(&self.body.records, &manifest)?;
validate_record_set(&source_operation, terminal_operation.state(), &records)?;
validate_receipt_record(
&records,
&source_operation,
&terminal_operation,
&self.body.context,
&self.body.signer_key,
)?;
let channel_terminal =
validate_channel_terminal_record(&records, &source_operation, &self.body.context)?;
let anchored_economic_mutation = validate_economic_mutation_result_record(
&records,
&source_operation,
&terminal_operation,
&self.body.context,
)?;
let (pre_dispatch_release_proof, anchored_transport_cancellation) =
validate_release_proof_record(
&records,
&source_operation,
terminal_operation.state(),
&self.body.context,
)?;
validate_sidecars(
&records,
self.body.authorization_consumption.as_ref(),
self.body.observer.as_ref(),
)?;
Ok(VerifiedAdmissionTerminalProjectionV1 {
signer_key: self.body.signer_key.clone(),
context: self.body.context.clone(),
source_operation,
terminal_operation,
projection_json,
manifest_json,
records,
channel_terminal,
anchored_transport_cancellation,
anchored_economic_mutation,
pre_dispatch_release_proof,
authorization_consumption: self.body.authorization_consumption.clone(),
observer: self.body.observer.clone(),
})
}
}
impl VerifiedAdmissionTerminalProjectionV1 {
#[must_use]
pub const fn signer_key(&self) -> &PublicKey {
&self.signer_key
}
#[must_use]
pub const fn pre_dispatch_release_proof(&self) -> Option<&VerifiedPreDispatchNoEffect> {
self.pre_dispatch_release_proof.as_ref()
}
#[must_use]
pub const fn context(&self) -> &AdmissionProjectionContext {
&self.context
}
#[must_use]
pub const fn source_operation(&self) -> &AdmissionOperationV1 {
&self.source_operation
}
#[must_use]
pub const fn terminal_operation(&self) -> &AdmissionOperationV1 {
&self.terminal_operation
}
#[must_use]
pub fn projection_json(&self) -> &[u8] {
&self.projection_json
}
#[must_use]
pub fn manifest_json(&self) -> &[u8] {
&self.manifest_json
}
#[must_use]
pub fn records(&self) -> &[VerifiedAdmissionTerminalProjectionRecordV1] {
&self.records
}
#[must_use]
pub const fn channel_terminal(&self) -> Option<&VerifiedChannelTerminalProjectionV1> {
self.channel_terminal.as_ref()
}
#[must_use]
pub const fn requires_anchored_economic_commit(&self) -> bool {
self.channel_terminal.is_some()
|| self.anchored_transport_cancellation
|| self.anchored_economic_mutation
}
#[must_use]
pub const fn authorization_consumption(&self) -> Option<&AuthorizationReceiptConsumption> {
self.authorization_consumption.as_ref()
}
#[must_use]
pub const fn observer(&self) -> Option<&AdmissionTerminalObserverProjectionV1> {
self.observer.as_ref()
}
pub fn terminal(&self) -> Result<AdmissionTerminal, AdmissionOperationError> {
Ok(AdmissionTerminal {
operation_id: self.context.operation_id.clone(),
state: self.terminal_operation.state(),
replay: self
.terminal_operation
.terminal_replay()
.cloned()
.ok_or(AdmissionOperationError::TerminalReplayMismatch)?,
})
}
}
fn signing_preimage(
body: &SignedAdmissionTerminalProjectionBodyV1,
) -> Result<Vec<u8>, AdmissionOperationError> {
let canonical = canonical_json_bytes(body)
.map_err(|error| AdmissionOperationError::CanonicalJson(error.to_string()))?;
let mut preimage =
Vec::with_capacity(SIGNED_TERMINAL_PROJECTION_DOMAIN.len() + canonical.len());
preimage.extend_from_slice(SIGNED_TERMINAL_PROJECTION_DOMAIN);
preimage.extend_from_slice(&canonical);
Ok(preimage)
}
fn decode_bounded(value: &str, maximum: usize) -> Result<Vec<u8>, AdmissionOperationError> {
let bytes = STANDARD
.decode(value)
.map_err(|_| AdmissionOperationError::TerminalProjectionBindingMismatch)?;
if bytes.is_empty() || bytes.len() > maximum {
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
}
Ok(bytes)
}
fn validate_terminal_successor(
source: &AdmissionOperationV1,
terminal: &AdmissionOperationV1,
context: &AdmissionProjectionContext,
projection_digest: &AdmissionDigest,
) -> Result<(), AdmissionOperationError> {
let source_persisted = source.to_persisted();
let terminal_persisted = terminal.to_persisted();
if source.state().is_terminal()
|| !terminal.state().is_terminal()
|| context.operation_id != *source.binding().operation_id()
|| context.request_id != source.replay_key().request_id
|| context.expected_operation_version != source.version()
|| context.coordinator_lease_epoch != source.coordinator_lease_epoch()
|| source_persisted.binding != terminal_persisted.binding
|| source_persisted.attachments != terminal_persisted.attachments
|| source_persisted.dispatch_commit != terminal_persisted.dispatch_commit
|| source_persisted.coordinator_lease_epoch != terminal_persisted.coordinator_lease_epoch
|| source_persisted.last_error != terminal_persisted.last_error
|| terminal.version() != next_version(source.version())?
|| !is_legal_transition(
source.binding().kind(),
source.binding().participant_requirements(),
source.state(),
terminal.state(),
)
|| terminal
.terminal_replay()
.is_none_or(|replay| replay.projection_digest() != projection_digest)
|| source.dispatch_commit().is_some_and(|commit| {
commit.store_fence.store_uuid != context.store_fence.store_uuid
|| (context.store_fence != commit.store_fence
&& context.store_fence.owner_epoch <= commit.store_fence.owner_epoch)
})
{
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
}
Ok(())
}
fn validate_projection_body(
bytes: &[u8],
context: &AdmissionProjectionContext,
terminal_state: AdmissionOperationState,
) -> Result<(), AdmissionOperationError> {
let value: serde_json::Value = serde_json::from_slice(bytes)
.map_err(|error| AdmissionOperationError::CanonicalJson(error.to_string()))?;
if canonical_json_bytes(&value)
.map_err(|error| AdmissionOperationError::CanonicalJson(error.to_string()))?
!= bytes
{
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
}
let expected_terminal = match terminal_state {
AdmissionOperationState::Completed => "completed",
AdmissionOperationState::CompensatedBeforeDispatch => "compensated_before_dispatch",
AdmissionOperationState::NotAcceptedAfterDispatchCommit => {
"not_accepted_after_dispatch_commit"
}
AdmissionOperationState::OutcomeUnknownAfterDispatch => "outcome_unknown_after_dispatch",
AdmissionOperationState::EconomicMutationApplied => "economic_mutation_applied",
AdmissionOperationState::EconomicMutationNotApplied => "economic_mutation_not_applied",
_ => return Err(AdmissionOperationError::TerminalProjectionBindingMismatch),
};
if value.get("terminal").and_then(serde_json::Value::as_str) != Some(expected_terminal)
|| value.get("context")
!= Some(
&serde_json::to_value(context)
.map_err(|error| AdmissionOperationError::CanonicalJson(error.to_string()))?,
)
{
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
}
Ok(())
}
fn validate_records(
records: &[PersistedAdmissionTerminalProjectionRecordV1],
manifest: &AdmissionProjectionManifestV1,
) -> Result<Vec<VerifiedAdmissionTerminalProjectionRecordV1>, AdmissionOperationError> {
if records.is_empty()
|| records.len() > MAX_ADMISSION_TERMINAL_RECORDS
|| records.len() != manifest.records().len()
{
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
}
records
.iter()
.zip(manifest.records())
.map(|(record, commitment)| {
let canonical_json =
decode_bounded(&record.canonical_json, MAX_ADMISSION_TERMINAL_RECORD_BYTES)?;
let value: serde_json::Value = serde_json::from_slice(&canonical_json)
.map_err(|_| AdmissionOperationError::TerminalProjectionBindingMismatch)?;
if canonical_json_bytes(&value)
.map_err(|error| AdmissionOperationError::CanonicalJson(error.to_string()))?
!= canonical_json
|| sha256_hex(&canonical_json) != record.record_digest.as_str()
|| record.kind != commitment.kind()
|| record.record_id != *commitment.record_id()
|| record.record_digest != *commitment.record_digest()
{
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
}
Ok(VerifiedAdmissionTerminalProjectionRecordV1 {
kind: record.kind,
record_id: record.record_id.clone(),
record_digest: record.record_digest.clone(),
canonical_json,
})
})
.collect()
}
fn validate_record_set(
source: &AdmissionOperationV1,
terminal_state: AdmissionOperationState,
records: &[VerifiedAdmissionTerminalProjectionRecordV1],
) -> Result<(), AdmissionOperationError> {
validate_record_set_shape(
source.binding().kind(),
source.binding().participant_requirements(),
terminal_state,
records,
)
}
fn validate_record_set_shape(
operation_kind: AdmissionOperationKind,
requirements: AdmissionParticipantRequirements,
terminal_state: AdmissionOperationState,
records: &[VerifiedAdmissionTerminalProjectionRecordV1],
) -> Result<(), AdmissionOperationError> {
let actual = records
.iter()
.map(|record| record.kind)
.collect::<BTreeSet<_>>();
if actual.len() != records.len() {
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
}
let mut expected = BTreeSet::new();
match terminal_state {
AdmissionOperationState::Completed => {
expected.insert(AdmissionProjectionRecordKind::Receipt);
if operation_kind == AdmissionOperationKind::ToolDispatch {
expected.insert(AdmissionProjectionRecordKind::ToolOutcome);
}
for (required, kind) in [
(
requirements.payment,
AdmissionProjectionRecordKind::PaymentTerminal,
),
(
requirements.authorization_consumption,
AdmissionProjectionRecordKind::AuthorizationConsumption,
),
(
requirements.outcome_eligibility,
AdmissionProjectionRecordKind::OutcomeEligibility,
),
(
requirements.observation_attempt_zero,
AdmissionProjectionRecordKind::ObservationAttemptZero,
),
] {
if required {
expected.insert(kind);
}
}
if requirements.channel {
expected.insert(AdmissionProjectionRecordKind::ChannelTerminal);
if channel_terminal_charge(records)?.units > 0 {
expected.insert(AdmissionProjectionRecordKind::Obligation);
}
} else if requirements.obligation {
expected.insert(AdmissionProjectionRecordKind::Obligation);
}
}
AdmissionOperationState::CompensatedBeforeDispatch
| AdmissionOperationState::NotAcceptedAfterDispatchCommit => {
expected.insert(AdmissionProjectionRecordKind::ReleaseProof);
let evidence_count =
usize::from(actual.contains(&AdmissionProjectionRecordKind::Receipt))
+ usize::from(actual.contains(&AdmissionProjectionRecordKind::Incident));
if evidence_count != 1 {
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
}
expected.insert(
if actual.contains(&AdmissionProjectionRecordKind::Receipt) {
AdmissionProjectionRecordKind::Receipt
} else {
AdmissionProjectionRecordKind::Incident
},
);
}
AdmissionOperationState::OutcomeUnknownAfterDispatch => {
expected.insert(AdmissionProjectionRecordKind::Incident);
}
AdmissionOperationState::EconomicMutationApplied
| AdmissionOperationState::EconomicMutationNotApplied => {
expected.insert(AdmissionProjectionRecordKind::EconomicMutationResult);
expected.insert(AdmissionProjectionRecordKind::MutationAudit);
}
_ => return Err(AdmissionOperationError::TerminalProjectionBindingMismatch),
}
if actual != expected {
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
}
Ok(())
}
fn validate_release_proof_record(
records: &[VerifiedAdmissionTerminalProjectionRecordV1],
source: &AdmissionOperationV1,
terminal_state: AdmissionOperationState,
context: &AdmissionProjectionContext,
) -> Result<(Option<VerifiedPreDispatchNoEffect>, bool), AdmissionOperationError> {
let record = records
.iter()
.find(|record| record.kind == AdmissionProjectionRecordKind::ReleaseProof);
match (terminal_state, record) {
(AdmissionOperationState::CompensatedBeforeDispatch, Some(record)) => {
VerifiedPreDispatchNoEffect::from_canonical_record_verified(
record.canonical_json(),
source,
context,
)
.map(|proof| (Some(proof), false))
.map_err(|_| AdmissionOperationError::TerminalProjectionBindingMismatch)
}
(AdmissionOperationState::NotAcceptedAfterDispatchCommit, Some(record)) => {
VerifiedTransportNotAccepted::from_canonical_record_verified(
record.canonical_json(),
source,
context,
)
.map(|proof| (None, proof.uses_economic_effect_cancellation()))
.map_err(|_| AdmissionOperationError::TerminalProjectionBindingMismatch)
}
(
AdmissionOperationState::Completed
| AdmissionOperationState::OutcomeUnknownAfterDispatch
| AdmissionOperationState::EconomicMutationApplied
| AdmissionOperationState::EconomicMutationNotApplied,
None,
) => Ok((None, false)),
_ => Err(AdmissionOperationError::TerminalProjectionBindingMismatch),
}
}
fn validate_economic_mutation_result_record(
records: &[VerifiedAdmissionTerminalProjectionRecordV1],
source: &AdmissionOperationV1,
terminal: &AdmissionOperationV1,
context: &AdmissionProjectionContext,
) -> Result<bool, AdmissionOperationError> {
if !matches!(
terminal.state(),
AdmissionOperationState::EconomicMutationApplied
| AdmissionOperationState::EconomicMutationNotApplied
) {
return Ok(false);
}
let record = records
.iter()
.find(|record| record.kind == AdmissionProjectionRecordKind::EconomicMutationResult)
.ok_or(AdmissionOperationError::TerminalProjectionBindingMismatch)?;
let result: GovernedEconomicMutationResultBinding =
serde_json::from_slice(record.canonical_json())
.map_err(|_| AdmissionOperationError::TerminalProjectionBindingMismatch)?;
result
.validate_remote_terminal(source, context, terminal, record.record_id())
.map_err(|_| AdmissionOperationError::TerminalProjectionBindingMismatch)
}
fn channel_terminal_charge(
records: &[VerifiedAdmissionTerminalProjectionRecordV1],
) -> Result<MonetaryAmount, AdmissionOperationError> {
let record = records
.iter()
.find(|record| record.kind == AdmissionProjectionRecordKind::ChannelTerminal)
.ok_or(AdmissionOperationError::TerminalProjectionBindingMismatch)?;
let value: serde_json::Value = serde_json::from_slice(record.canonical_json())
.map_err(|_| AdmissionOperationError::TerminalProjectionBindingMismatch)?;
let charge = value
.get("actual_charge")
.cloned()
.ok_or(AdmissionOperationError::TerminalProjectionBindingMismatch)
.and_then(|value| {
serde_json::from_value::<MonetaryAmount>(value)
.map_err(|_| AdmissionOperationError::TerminalProjectionBindingMismatch)
})?;
if charge.units > ((1_u64 << 53) - 1)
|| charge.currency.len() != 3
|| !charge
.currency
.bytes()
.all(|byte| byte.is_ascii_uppercase())
{
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
}
Ok(charge)
}
fn validate_channel_terminal_record(
records: &[VerifiedAdmissionTerminalProjectionRecordV1],
source: &AdmissionOperationV1,
context: &AdmissionProjectionContext,
) -> Result<Option<VerifiedChannelTerminalProjectionV1>, AdmissionOperationError> {
let Some(record) = records
.iter()
.find(|record| record.kind == AdmissionProjectionRecordKind::ChannelTerminal)
else {
return Ok(None);
};
let channel = VerifiedChannelTerminalProjectionV1::from_canonical_record_verified(
record.canonical_json(),
source,
context,
)?;
if channel.record_id() != record.record_id() {
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
}
let receipt_record = records
.iter()
.find(|record| record.kind == AdmissionProjectionRecordKind::Receipt)
.ok_or(AdmissionOperationError::TerminalProjectionBindingMismatch)?;
let receipt: ChioReceipt = serde_json::from_slice(receipt_record.canonical_json())
.map_err(|_| AdmissionOperationError::TerminalProjectionBindingMismatch)?;
let receipt_digest = AdmissionDigest::try_new(
"channel_terminal_receipt_digest",
sha256_hex(
&canonical_json_bytes(&receipt)
.map_err(|error| AdmissionOperationError::CanonicalJson(error.to_string()))?,
),
)?;
let expected_authority = AdmissionDigest::try_new(
"channel_terminal_receipt_authority_digest",
derive_channel_receipt_authority_digest(&receipt.kernel_key)
.map_err(|_| AdmissionOperationError::TerminalProjectionBindingMismatch)?,
)?;
let financial = receipt
.financial_metadata()
.ok_or(AdmissionOperationError::TerminalProjectionBindingMismatch)?;
let metadata = receipt
.channel_metadata()
.ok_or(AdmissionOperationError::TerminalProjectionBindingMismatch)?;
let tool_record = records
.iter()
.find(|record| record.kind == AdmissionProjectionRecordKind::ToolOutcome)
.ok_or(AdmissionOperationError::TerminalProjectionBindingMismatch)?;
let Some(chio_core::economic_continuity::EconomicEffectTerminalV1::Completed {
result_id,
result_digest,
result,
}) = channel.completed_effect_slot().terminal.as_ref()
else {
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
};
let chio_core::economic_continuity::EconomicContentV1::Inline { value } = result else {
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
};
let result_bytes = canonical_json_bytes(value)
.map_err(|error| AdmissionOperationError::CanonicalJson(error.to_string()))?;
let expected_result_digest = result
.digest()
.map_err(|_| AdmissionOperationError::TerminalProjectionBindingMismatch)?;
if channel.receipt_id().as_str() != receipt.id
|| channel.receipt_digest() != &receipt_digest
|| channel.receipt_authority_digest() != &expected_authority
|| channel.actual_charge().units != financial.cost_charged
|| channel.actual_charge().currency != financial.currency
|| metadata.channel_id != channel.signed_reservation().body.channel_id
|| metadata.open_digest != channel.signed_reservation().body.open_digest
|| metadata.reservation_id != channel.signed_reservation().body.reservation_id
|| metadata.reservation_digest != channel.reservation_digest().as_str()
|| metadata.sequence != channel.signed_reservation().body.next_sequence
|| result_id != tool_record.record_id().as_str()
|| result_digest != &expected_result_digest
|| result_bytes != tool_record.canonical_json()
{
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
}
validate_channel_obligation_record(records, source, context, &channel, &receipt, tool_record)?;
Ok(Some(channel))
}
fn validate_channel_obligation_record(
records: &[VerifiedAdmissionTerminalProjectionRecordV1],
source: &AdmissionOperationV1,
context: &AdmissionProjectionContext,
channel: &VerifiedChannelTerminalProjectionV1,
receipt: &ChioReceipt,
tool_record: &VerifiedAdmissionTerminalProjectionRecordV1,
) -> Result<(), AdmissionOperationError> {
let mismatch = || AdmissionOperationError::TerminalProjectionBindingMismatch;
let obligation_record = records
.iter()
.find(|record| record.kind == AdmissionProjectionRecordKind::Obligation);
if channel.actual_charge().units == 0 {
if obligation_record.is_some()
|| channel.obligation_atom_id().is_some()
|| channel.obligation_atom_digest().is_some()
{
return Err(mismatch());
}
return Ok(());
}
let obligation_record = obligation_record.ok_or_else(mismatch)?;
let obligation: UntrustedTerminalObligationProjectionV1 =
serde_json::from_slice(obligation_record.canonical_json()).map_err(|_| mismatch())?;
obligation.source.binding.validate_against(
source,
context,
AdmissionOperationState::Completed,
)?;
obligation.atom.validate().map_err(|_| mismatch())?;
obligation
.disposition_record
.validate_against(&obligation.atom)
.map_err(|_| mismatch())?;
let atom_digest = obligation.atom.digest().map_err(|_| mismatch())?;
let proposal_intent_digest = channel
.signed_reservation()
.body
.proposal_digest()
.map_err(|_| mismatch())?;
let tool_value: serde_json::Value =
serde_json::from_slice(tool_record.canonical_json()).map_err(|_| mismatch())?;
let outcome_version = tool_value
.get("outcome_version")
.and_then(serde_json::Value::as_u64)
.ok_or_else(mismatch)?;
validate_positive_ijson("channel_terminal_outcome_version", outcome_version)?;
let disposition_matches = matches!(
obligation.disposition_record.disposition(),
ObligationDispositionV1::Channelized {
channel_id,
reservation_id,
} if channel_id == &channel.signed_reservation().body.channel_id
&& reservation_id == &channel.signed_reservation().body.reservation_id
);
let transition_matches = matches!(
obligation.disposition_record.last_transition(),
ObligationDispositionTransitionV1::ReserveChannel {
channel_id,
reservation_id,
authority_digest,
} if channel_id == &channel.signed_reservation().body.channel_id
&& reservation_id == &channel.signed_reservation().body.reservation_id
&& authority_digest == channel.reservation_digest().as_str()
);
if obligation_record.record_id().as_str() != source.binding().operation_id().as_str()
|| channel
.obligation_atom_id()
.map(AdmissionIdentifier::as_str)
!= Some(obligation.atom.obligation_id())
|| channel
.obligation_atom_digest()
.map(AdmissionDigest::as_str)
!= Some(atom_digest.as_str())
|| obligation.atom.amount() != channel.actual_charge()
|| obligation.atom.economic_intent_digest() != proposal_intent_digest.as_str()
|| obligation.atom.source_receipt_id() != receipt.id.as_str()
|| obligation.atom.source_receipt_id() != channel.receipt_id().as_str()
|| obligation.atom.source_receipt_digest() != channel.receipt_digest().as_str()
|| obligation.atom.pre_action_authority_digest() != channel.reservation_digest().as_str()
|| obligation.atom.created_at_unix_ms() > context.trusted_time_unix_ms
|| obligation.source.source_authority_digest != *channel.reservation_digest()
|| obligation.source.source_record_id.as_str() != obligation.atom.obligation_id()
|| obligation.source.source_record_digest.as_str() != atom_digest.as_str()
|| obligation.source.source_recorded_at_unix_ms != obligation.atom.created_at_unix_ms()
|| obligation.source.consumer_receipt_id != *channel.receipt_id()
|| obligation.source.consumer_receipt_digest != *channel.receipt_digest()
|| obligation.source.outcome_id.as_str() != tool_record.record_id().as_str()
|| obligation.source.outcome_version != outcome_version
|| obligation.disposition_record.version() != 2
|| obligation.disposition_record.lifecycle_fence() != 2
|| !disposition_matches
|| !transition_matches
{
return Err(mismatch());
}
Ok(())
}
fn validate_receipt_record(
records: &[VerifiedAdmissionTerminalProjectionRecordV1],
source: &AdmissionOperationV1,
terminal: &AdmissionOperationV1,
context: &AdmissionProjectionContext,
signer_key: &PublicKey,
) -> Result<(), AdmissionOperationError> {
let Some(record) = records
.iter()
.find(|record| record.kind == AdmissionProjectionRecordKind::Receipt)
else {
return Ok(());
};
let receipt: ChioReceipt = serde_json::from_slice(&record.canonical_json)
.map_err(|_| AdmissionOperationError::TerminalProjectionBindingMismatch)?;
if receipt.id != record.record_id.as_str() || receipt.kernel_key != *signer_key {
return Err(AdmissionOperationError::TerminalProjectionBindingMismatch);
}
let metadata = receipt
.metadata
.as_ref()
.and_then(serde_json::Value::as_object)
.and_then(|object| object.get(ADMISSION_RECEIPT_METADATA_KEY))
.cloned()
.ok_or(AdmissionOperationError::TerminalProjectionBindingMismatch)
.and_then(|value| {
serde_json::from_value::<AdmissionReceiptMetadataV1>(value)
.map_err(|_| AdmissionOperationError::TerminalProjectionBindingMismatch)
})?;
let outcome = metadata
.tool_outcome_id
.as_ref()
.zip(metadata.tool_outcome_version);
let compensation = match terminal.state() {
AdmissionOperationState::Completed => AdmissionCompensationStatus::NotCompensated,
AdmissionOperationState::CompensatedBeforeDispatch => {
AdmissionCompensationStatus::CompensatedBeforeDispatch
}
AdmissionOperationState::NotAcceptedAfterDispatchCommit => {
AdmissionCompensationStatus::NotAcceptedAfterDispatchCommit
}
_ => return Err(AdmissionOperationError::TerminalProjectionBindingMismatch),
};
validate_receipt_projection(
&receipt,
source,
context,
terminal.state(),
compensation,
outcome,
)
}
fn validate_sidecars(
records: &[VerifiedAdmissionTerminalProjectionRecordV1],
authorization: Option<&AuthorizationReceiptConsumption>,
observer: Option<&AdmissionTerminalObserverProjectionV1>,
) -> Result<(), AdmissionOperationError> {
let authorization_record = records
.iter()
.find(|record| record.kind == AdmissionProjectionRecordKind::AuthorizationConsumption);
match (authorization_record, authorization) {
(Some(record), Some(authorization))
if record.record_id.as_str() == authorization.authorization_receipt_id
&& record.canonical_json
== canonical_json_bytes(authorization).map_err(|error| {
AdmissionOperationError::CanonicalJson(error.to_string())
})? => {}
(None, None) => {}
_ => return Err(AdmissionOperationError::TerminalProjectionBindingMismatch),
}
let observer_record = records
.iter()
.find(|record| record.kind == AdmissionProjectionRecordKind::ObservationAttemptZero);
match (observer_record, observer) {
(Some(record), Some(observer))
if record.record_id == observer.receipt_id
&& record.canonical_json
== canonical_json_bytes(&observer.pending).map_err(|error| {
AdmissionOperationError::CanonicalJson(error.to_string())
})? => {}
(None, None) => {}
_ => return Err(AdmissionOperationError::TerminalProjectionBindingMismatch),
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn record(
kind: AdmissionProjectionRecordKind,
value: serde_json::Value,
) -> Result<VerifiedAdmissionTerminalProjectionRecordV1, AdmissionOperationError> {
let canonical_json = canonical_json_bytes(&value)
.map_err(|error| AdmissionOperationError::CanonicalJson(error.to_string()))?;
Ok(VerifiedAdmissionTerminalProjectionRecordV1 {
kind,
record_id: AdmissionIdentifier::try_new("record_id", kind.as_str().to_owned())?,
record_digest: AdmissionDigest::try_new("record_digest", sha256_hex(&canonical_json))?,
canonical_json,
})
}
fn channel_records(
units: u64,
) -> Result<Vec<VerifiedAdmissionTerminalProjectionRecordV1>, AdmissionOperationError> {
Ok(vec![
record(
AdmissionProjectionRecordKind::Receipt,
serde_json::json!({}),
)?,
record(
AdmissionProjectionRecordKind::ToolOutcome,
serde_json::json!({}),
)?,
record(
AdmissionProjectionRecordKind::ChannelTerminal,
serde_json::json!({
"actual_charge": { "currency": "USD", "units": units }
}),
)?,
])
}
#[test]
fn channel_record_set_requires_charge_dependent_obligation(
) -> Result<(), AdmissionOperationError> {
let requirements = AdmissionParticipantRequirements {
broker_attempt: true,
budget_capture: true,
obligation: true,
channel: true,
..AdmissionParticipantRequirements::NONE
};
let validate = |records: &[VerifiedAdmissionTerminalProjectionRecordV1]| {
validate_record_set_shape(
AdmissionOperationKind::ToolDispatch,
requirements,
AdmissionOperationState::Completed,
records,
)
};
let mut positive = channel_records(7)?;
assert_eq!(
validate(&positive),
Err(AdmissionOperationError::TerminalProjectionBindingMismatch)
);
positive.push(record(
AdmissionProjectionRecordKind::Obligation,
serde_json::json!({}),
)?);
validate(&positive)?;
let mut zero = channel_records(0)?;
validate(&zero)?;
zero.push(record(
AdmissionProjectionRecordKind::Obligation,
serde_json::json!({}),
)?);
assert_eq!(
validate(&zero),
Err(AdmissionOperationError::TerminalProjectionBindingMismatch)
);
Ok(())
}
#[test]
fn channel_record_set_rejects_missing_duplicate_and_malformed_terminal_records(
) -> Result<(), AdmissionOperationError> {
let requirements = AdmissionParticipantRequirements {
broker_attempt: true,
budget_capture: true,
obligation: true,
channel: true,
..AdmissionParticipantRequirements::NONE
};
let validate = |records: &[VerifiedAdmissionTerminalProjectionRecordV1]| {
validate_record_set_shape(
AdmissionOperationKind::ToolDispatch,
requirements,
AdmissionOperationState::Completed,
records,
)
};
let mut missing = channel_records(0)?;
missing.retain(|record| record.kind != AdmissionProjectionRecordKind::ChannelTerminal);
assert!(validate(&missing).is_err());
let mut duplicate = channel_records(0)?;
duplicate.push(record(
AdmissionProjectionRecordKind::ChannelTerminal,
serde_json::json!({
"actual_charge": { "currency": "USD", "units": 0 }
}),
)?);
assert!(validate(&duplicate).is_err());
let mut malformed = channel_records(0)?;
let channel_record = malformed
.iter_mut()
.find(|record| record.kind == AdmissionProjectionRecordKind::ChannelTerminal)
.ok_or(AdmissionOperationError::TerminalProjectionBindingMismatch)?;
channel_record.canonical_json = canonical_json_bytes(&serde_json::json!({
"actual_charge": { "currency": "usd", "units": 0 }
}))
.map_err(|error| AdmissionOperationError::CanonicalJson(error.to_string()))?;
assert!(validate(&malformed).is_err());
Ok(())
}
#[test]
fn non_channel_record_set_preserves_legacy_obligation_shape(
) -> Result<(), AdmissionOperationError> {
let requirements = AdmissionParticipantRequirements {
broker_attempt: true,
budget_capture: true,
obligation: true,
..AdmissionParticipantRequirements::NONE
};
let mut records = vec![
record(
AdmissionProjectionRecordKind::Receipt,
serde_json::json!({}),
)?,
record(
AdmissionProjectionRecordKind::ToolOutcome,
serde_json::json!({}),
)?,
record(
AdmissionProjectionRecordKind::Obligation,
serde_json::json!({}),
)?,
];
let validate = |records: &[VerifiedAdmissionTerminalProjectionRecordV1]| {
validate_record_set_shape(
AdmissionOperationKind::ToolDispatch,
requirements,
AdmissionOperationState::Completed,
records,
)
};
validate(&records)?;
records.push(record(
AdmissionProjectionRecordKind::ChannelTerminal,
serde_json::json!({
"actual_charge": { "currency": "USD", "units": 0 }
}),
)?);
assert_eq!(
validate(&records),
Err(AdmissionOperationError::TerminalProjectionBindingMismatch)
);
Ok(())
}
}