laminar_db/process_function/
descriptor.rs

1use laminar_core::state::PARTITIONING_ABI_VERSION;
2use serde::{Deserialize, Serialize};
3use sha2::{Digest, Sha256};
4
5use super::schema::{schema_from_canonical_fields, CanonicalField};
6use super::{
7    canonical_fields, ProcessDeterminism, ProcessFunctionDescriptor, ProcessFunctionLimits,
8    ProcessRuntime, PythonEnvironmentBinding, STATE_CODEC_VERSION,
9};
10use crate::error::DbError;
11
12pub(crate) const MAX_MANIFEST_BYTES: usize = 64 * 1024;
13const PROTOCOL_VERSION: u32 = 1;
14pub(crate) const MAX_PYTHON_IMPORT_ROOTS: usize = 16;
15
16#[derive(Deserialize, Serialize)]
17#[serde(deny_unknown_fields)]
18struct Manifest {
19    version: u32,
20    protocol_version: u32,
21    runtime: String,
22    function_id: String,
23    pipeline_state_id: String,
24    implementation_digest: String,
25    #[serde(
26        default,
27        skip_serializing_if = "Option::is_none",
28        deserialize_with = "decode_python_environment"
29    )]
30    python_environment: Option<PythonEnvironmentBinding>,
31    input_schema: Vec<CanonicalField>,
32    output_schema: Vec<CanonicalField>,
33    key_columns: Vec<String>,
34    partitioning_abi: u16,
35    event_time_column: String,
36    output_event_time_column: String,
37    late_event_policy: String,
38    input_changelog: String,
39    output_changelog: String,
40    value_state_name: String,
41    state_codec_version: u32,
42    timer_names: Vec<String>,
43    determinism: ProcessDeterminism,
44    limits: ProcessFunctionLimits,
45}
46
47impl Manifest {
48    fn from_descriptor(descriptor: &ProcessFunctionDescriptor) -> Result<Self, DbError> {
49        Ok(Self {
50            version: descriptor.version,
51            protocol_version: PROTOCOL_VERSION,
52            runtime: match descriptor.runtime {
53                ProcessRuntime::NativeRust => "trusted_native_rust",
54                ProcessRuntime::RemoteRust => "remote_rust",
55                ProcessRuntime::RemotePython => "remote_python",
56            }
57            .into(),
58            function_id: descriptor.function_id.clone(),
59            pipeline_state_id: descriptor.pipeline_state_id.clone(),
60            implementation_digest: descriptor.implementation_digest.clone(),
61            python_environment: descriptor.python_environment.clone(),
62            input_schema: canonical_fields(&descriptor.input_schema)?,
63            output_schema: canonical_fields(&descriptor.output_schema)?,
64            key_columns: descriptor.key_columns.clone(),
65            partitioning_abi: PARTITIONING_ABI_VERSION,
66            event_time_column: descriptor.event_time_column.clone(),
67            output_event_time_column: descriptor.output_event_time_column.clone(),
68            late_event_policy: "reject".into(),
69            input_changelog: "append_only".into(),
70            output_changelog: "append_only".into(),
71            value_state_name: descriptor.value_state_name.clone(),
72            state_codec_version: STATE_CODEC_VERSION,
73            timer_names: descriptor.timer_names.clone(),
74            determinism: descriptor.determinism,
75            limits: descriptor.limits,
76        })
77    }
78
79    fn into_descriptor(self) -> Result<ProcessFunctionDescriptor, DbError> {
80        if self.version != 1
81            || self.protocol_version != PROTOCOL_VERSION
82            || self.partitioning_abi != PARTITIONING_ABI_VERSION
83            || self.late_event_policy != "reject"
84            || self.input_changelog != "append_only"
85            || self.output_changelog != "append_only"
86            || self.state_codec_version != STATE_CODEC_VERSION
87        {
88            return Err(DbError::Unsupported(
89                "unsupported process function manifest contract".into(),
90            ));
91        }
92        let runtime = match self.runtime.as_str() {
93            "trusted_native_rust" => ProcessRuntime::NativeRust,
94            "remote_rust" => ProcessRuntime::RemoteRust,
95            "remote_python" => ProcessRuntime::RemotePython,
96            _ => {
97                return Err(DbError::Unsupported(
98                    "unsupported process function runtime".into(),
99                ))
100            }
101        };
102        let descriptor = ProcessFunctionDescriptor {
103            version: self.version,
104            runtime,
105            function_id: self.function_id,
106            pipeline_state_id: self.pipeline_state_id,
107            implementation_digest: self.implementation_digest,
108            determinism: self.determinism,
109            python_environment: self.python_environment,
110            input_schema: schema_from_canonical_fields(self.input_schema)?,
111            output_schema: schema_from_canonical_fields(self.output_schema)?,
112            key_columns: self.key_columns,
113            event_time_column: self.event_time_column,
114            output_event_time_column: self.output_event_time_column,
115            value_state_name: self.value_state_name,
116            timer_names: self.timer_names,
117            limits: self.limits,
118        };
119        super::operator::validate_descriptor(&descriptor)?;
120        Ok(descriptor)
121    }
122}
123
124impl PythonEnvironmentBinding {
125    pub(crate) fn validate(&self) -> Result<(), DbError> {
126        if self.version != 1
127            || !valid_relative_python_path(&self.executable)
128            || !valid_python_handler(&self.handler)
129            || !valid_sha256(&self.runtime_sha256)
130            || self.import_roots_sha256.is_empty()
131            || self.import_roots_sha256.len() > MAX_PYTHON_IMPORT_ROOTS
132            || !self
133                .import_roots_sha256
134                .iter()
135                .all(|digest| valid_sha256(digest))
136        {
137            return Err(DbError::InvalidOperation(
138                "invalid Python environment binding".into(),
139            ));
140        }
141        Ok(())
142    }
143}
144
145pub(crate) fn valid_relative_python_path(path: &str) -> bool {
146    !path.is_empty()
147        && path.len() <= 1024
148        && !path.bytes().any(|byte| matches!(byte, b'\\' | b':' | 0))
149        && path.split('/').all(|part| !matches!(part, "" | "." | ".."))
150}
151
152pub(crate) fn valid_python_identifier(value: &str) -> bool {
153    let mut bytes = value.bytes();
154    matches!(bytes.next(), Some(b'a'..=b'z' | b'A'..=b'Z' | b'_'))
155        && bytes.all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
156}
157
158pub(crate) fn valid_python_handler(value: &str) -> bool {
159    value.split_once(':').is_some_and(|(module, function)| {
160        valid_python_identifier(module) && valid_python_identifier(function)
161    })
162}
163
164fn valid_sha256(digest: &str) -> bool {
165    digest.len() == 64
166        && digest
167            .bytes()
168            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
169}
170
171fn decode_python_environment<'de, D>(
172    deserializer: D,
173) -> Result<Option<PythonEnvironmentBinding>, D::Error>
174where
175    D: serde::Deserializer<'de>,
176{
177    PythonEnvironmentBinding::deserialize(deserializer).map(Some)
178}
179
180impl ProcessFunctionDescriptor {
181    /// Encode the validated v1 function contract as canonical UTF-8 JSON. This is a portable
182    /// package and negotiation manifest; it does not package native code or certify replay.
183    ///
184    /// # Errors
185    /// Rejects unsupported schemas or invalid descriptor fields.
186    pub fn to_manifest_json(&self) -> Result<Vec<u8>, DbError> {
187        super::operator::validate_descriptor(self)?;
188        let bytes = serde_json::to_vec(&Manifest::from_descriptor(self)?).map_err(|error| {
189            DbError::InvalidOperation(format!("encode process manifest: {error}"))
190        })?;
191        if bytes.len() > MAX_MANIFEST_BYTES {
192            return Err(DbError::InvalidOperation(
193                "process manifest exceeds 64 KiB".into(),
194            ));
195        }
196        Ok(bytes)
197    }
198
199    /// Decode and validate a v1 function manifest. Unknown fields and unsupported semantics
200    /// fail closed; the caller must separately bind the implementation digest to trusted code.
201    ///
202    /// # Errors
203    /// Rejects malformed, oversized or incompatible manifests.
204    pub fn from_manifest_json(bytes: &[u8]) -> Result<Self, DbError> {
205        if bytes.len() > MAX_MANIFEST_BYTES {
206            return Err(DbError::InvalidOperation(
207                "process manifest exceeds 64 KiB".into(),
208            ));
209        }
210        let manifest: Manifest = serde_json::from_slice(bytes).map_err(|error| {
211            DbError::InvalidOperation(format!("decode process manifest: {error}"))
212        })?;
213        manifest.into_descriptor()
214    }
215
216    pub(crate) fn binding_sha256(&self) -> Result<String, DbError> {
217        Ok(format!("{:x}", Sha256::digest(self.to_manifest_json()?)))
218    }
219}
laminar_db/process_function/descriptor.rs

laminar_db/process_function/
descriptor.rs
