use serde::{Deserialize, Serialize};
use std::fmt;
use super::KernelBootstrapLimits;
use super::command::HostCommand;
use super::config::OperationConfig;
use super::effect::EffectOutcome;
use super::event::ExternalEvent;
use super::root::{InitialContext, RootEntry};
use super::scalar::{EffectId, InputId, OperationId, SCALAR_ERROR_MARKER, WireU64};
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WireEnvelope {
pub operation_id: OperationId,
pub input_id: InputId,
pub observed_at_ms: WireU64,
pub input: KernelInput,
}
impl WireEnvelope {
pub fn new(
operation_id: OperationId,
input_id: InputId,
observed_at_ms: WireU64,
input: KernelInput,
) -> Self {
Self {
operation_id,
input_id,
observed_at_ms,
input,
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum KernelInput {
ConfigureOperation(ConfigureOperation),
StartOperation(StartOperation),
ResolveEffect(ResolveEffect),
DeliverExternalEvent(DeliverExternalEvent),
HostControl(HostControl),
}
impl KernelInput {
pub fn authority(&self) -> InputAuthority {
match self {
Self::ConfigureOperation(_) => InputAuthority::HostBootstrap,
Self::StartOperation(_) => InputAuthority::HostRoot,
Self::ResolveEffect(_) => InputAuthority::HostEffectResolution,
Self::DeliverExternalEvent(_) => InputAuthority::HostObservedFact,
Self::HostControl(_) => InputAuthority::HostControlPlane,
}
}
pub fn admissible_lifecycles(&self) -> &'static [OperationLifecycle] {
match self {
Self::ConfigureOperation(_) => &[OperationLifecycle::Created],
Self::StartOperation(_) => &[OperationLifecycle::Configured],
Self::ResolveEffect(_) | Self::DeliverExternalEvent(_) => {
&[OperationLifecycle::Running, OperationLifecycle::Suspended]
}
Self::HostControl(_) => &[
OperationLifecycle::Configured,
OperationLifecycle::Running,
OperationLifecycle::Suspended,
],
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum InputAuthority {
HostBootstrap,
HostRoot,
HostEffectResolution,
HostObservedFact,
HostControlPlane,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OperationLifecycle {
Created,
Configured,
Running,
Suspended,
Completed,
Cancelled,
Failed,
}
impl OperationLifecycle {
pub fn is_terminal(self) -> bool {
matches!(self, Self::Completed | Self::Cancelled | Self::Failed)
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConfigureOperation {
pub config: OperationConfig,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct StartOperation {
pub entry: RootEntry,
pub initial_context: InitialContext,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ResolveEffect {
pub effect_id: EffectId,
pub outcome: EffectOutcome,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DeliverExternalEvent {
pub event: ExternalEvent,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct HostControl {
pub command: HostCommand,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WireRejectionKind {
InputTooLarge,
DepthExceeded,
CollectionTooLarge,
MalformedJson,
UnknownField,
UnknownVariant,
MissingField,
InvalidScalar,
TypeMismatch,
PolicyViolation,
}
impl WireRejectionKind {
pub fn as_str(self) -> &'static str {
match self {
Self::InputTooLarge => "input_too_large",
Self::DepthExceeded => "depth_exceeded",
Self::CollectionTooLarge => "collection_too_large",
Self::MalformedJson => "malformed_json",
Self::UnknownField => "unknown_field",
Self::UnknownVariant => "unknown_variant",
Self::MissingField => "missing_field",
Self::InvalidScalar => "invalid_scalar",
Self::TypeMismatch => "type_mismatch",
Self::PolicyViolation => "policy_violation",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WireRejection {
pub kind: WireRejectionKind,
pub message: String,
}
impl WireRejection {
pub fn new(kind: WireRejectionKind, message: impl Into<String>) -> Self {
Self {
kind,
message: message.into(),
}
}
}
impl fmt::Display for WireRejection {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}: {}", self.kind.as_str(), self.message)
}
}
impl std::error::Error for WireRejection {}
pub fn decode_envelope_json(
input_json: &str,
limits: &KernelBootstrapLimits,
) -> Result<WireEnvelope, WireRejection> {
if input_json.len() > limits.absolute_max_input_bytes as usize {
return Err(WireRejection::new(
WireRejectionKind::InputTooLarge,
format!(
"kernel input is {} bytes; the absolute bound is {} bytes",
input_json.len(),
limits.absolute_max_input_bytes
),
));
}
scan_structural_boundary(input_json, limits)?;
serde_json::from_str(input_json).map_err(classify_serde_error)
}
pub fn encode_envelope_json(envelope: &WireEnvelope) -> String {
serde_json::to_string(envelope).expect("wire envelope is always serializable")
}
fn classify_serde_error(error: serde_json::Error) -> WireRejection {
let message = error.to_string();
let kind = if error.classify() == serde_json::error::Category::Syntax
|| error.classify() == serde_json::error::Category::Eof
{
WireRejectionKind::MalformedJson
} else if message.contains(SCALAR_ERROR_MARKER) {
WireRejectionKind::InvalidScalar
} else if message.contains("unknown field") {
WireRejectionKind::UnknownField
} else if message.contains("unknown variant") {
WireRejectionKind::UnknownVariant
} else if message.contains("missing field") {
WireRejectionKind::MissingField
} else {
WireRejectionKind::TypeMismatch
};
WireRejection::new(kind, message)
}
pub fn scan_structural_boundary(
input_json: &str,
limits: &KernelBootstrapLimits,
) -> Result<(), WireRejection> {
type Frame = (u64, bool);
let mut stack: Vec<Frame> = Vec::new();
let mut in_string = false;
let mut escaped = false;
for &byte in input_json.as_bytes() {
if in_string {
if escaped {
escaped = false;
} else if byte == b'\\' {
escaped = true;
} else if byte == b'"' {
in_string = false;
}
continue;
}
match byte {
b'"' => {
in_string = true;
mark_content(&mut stack);
}
b'{' | b'[' => {
mark_content(&mut stack);
stack.push((0, false));
if stack.len() > limits.absolute_max_json_depth as usize {
return Err(WireRejection::new(
WireRejectionKind::DepthExceeded,
format!(
"kernel input nests {} levels deep; the absolute bound is {}",
stack.len(),
limits.absolute_max_json_depth
),
));
}
}
b'}' | b']' => {
let Some((separators, has_content)) = stack.pop() else {
return Ok(());
};
let entries = if has_content { separators + 1 } else { 0 };
if entries > u64::from(limits.absolute_max_collection_entries) {
return Err(WireRejection::new(
WireRejectionKind::CollectionTooLarge,
format!(
"kernel input has a container with {entries} entries; \
the absolute bound is {}",
limits.absolute_max_collection_entries
),
));
}
}
b',' => {
if let Some(frame) = stack.last_mut() {
frame.0 += 1;
frame.1 = true;
}
}
byte if byte.is_ascii_whitespace() => {}
_ => mark_content(&mut stack),
}
}
Ok(())
}
fn mark_content(stack: &mut [(u64, bool)]) {
if let Some(frame) = stack.last_mut() {
frame.1 = true;
}
}