laminar_db/process_function/remote/python_environment/
mod.rs

1use std::path::{Path, PathBuf};
2
3use super::LocalPythonWorkerConfig;
4use crate::error::DbError;
5use crate::process_function::descriptor::{valid_python_handler, MAX_PYTHON_IMPORT_ROOTS};
6use crate::process_function::{ProcessFunctionDescriptor, PythonEnvironmentBinding};
7
8mod file_guards;
9mod tree;
10
11pub(super) use file_guards::FileGuards;
12pub(super) use tree::file_sha256;
13use tree::{canonical_directory, fingerprint_guarded, InventoryBudget};
14
15#[cfg(all(test, target_os = "linux"))]
16pub(crate) fn read_only_fixture_config() -> LocalPythonWorkerConfig {
17    let package = std::path::PathBuf::from(
18        std::env::var_os("LAMINAR_PROCESS_REPLAY_PACKAGE")
19            .expect("set LAMINAR_PROCESS_REPLAY_PACKAGE to the read-only qualification package"),
20    );
21    LocalPythonWorkerConfig {
22        python: package.join("runtime/bin/python3.13"),
23        runtime_root: Some(package.join("runtime")),
24        manifest: package.join("manifest.json"),
25        handler_file: package.join("handler/replay_handler.py"),
26        function: "handle".into(),
27        python_paths: vec![package.join("runtime/lib/python3.13/site-packages")],
28        max_in_flight: 4,
29        timeout: std::time::Duration::from_secs(10),
30    }
31}
32
33pub(super) struct VerifiedEnvironment {
34    pub(super) python: PathBuf,
35    pub(super) runtime_root: Option<PathBuf>,
36    pub(super) import_roots: Vec<PathBuf>,
37    pub(super) _guards: FileGuards,
38}
39
40impl PythonEnvironmentBinding {
41    /// Fingerprint a quiescent runtime installation and ordered import trees, including all
42    /// source, bytecode, native libraries and data. The first import root is the handler directory.
43    /// `handler` selects the exact `module:function` entry point in that directory.
44    /// This performs blocking filesystem work; call it during packaging or on a blocking task.
45    /// On Windows, read-share handles exclude writers until capture returns. Directory additions
46    /// still require a quiescent deployment, and capture does not retain lifetime protection.
47    ///
48    /// # Errors
49    /// Rejects an interpreter outside the runtime tree, links/reparse points (including path
50    /// ancestors), more than 128 ancestors per configured path, non-UTF-8 paths, more than
51    /// 16 import roots, 32,768 total entries, 4 GiB total bytes or a 512 MiB file.
52    /// Windows also rejects files with an existing incompatible write/delete handle.
53    pub fn capture(
54        runtime_root: &Path,
55        python: &Path,
56        handler: &str,
57        import_roots: &[PathBuf],
58    ) -> Result<Self, DbError> {
59        capture(
60            runtime_root,
61            python,
62            handler,
63            import_roots,
64            &mut FileGuards::default(),
65        )
66    }
67}
68
69fn capture(
70    runtime_root: &Path,
71    python: &Path,
72    handler: &str,
73    import_roots: &[PathBuf],
74    guards: &mut FileGuards,
75) -> Result<PythonEnvironmentBinding, DbError> {
76    if import_roots.is_empty() || import_roots.len() > MAX_PYTHON_IMPORT_ROOTS {
77        return Err(DbError::Config(
78            "Python binding requires 1..=16 import roots".into(),
79        ));
80    }
81    if !valid_python_handler(handler) {
82        return Err(DbError::Config("invalid Python environment handler".into()));
83    }
84    let runtime_root = canonical_directory(runtime_root, guards)?;
85    let python = guards.canonical_file(python)?;
86    let relative = python
87        .strip_prefix(&runtime_root)
88        .map_err(|_| DbError::Config("Python executable is outside the runtime root".into()))?;
89    let executable = relative
90        .to_str()
91        .ok_or_else(|| DbError::Config("Python executable path must be UTF-8".into()))?
92        .replace(std::path::MAIN_SEPARATOR, "/");
93    let mut budget = InventoryBudget::default();
94    let runtime_sha256 = fingerprint_guarded(&runtime_root, &mut budget, guards)?;
95    let mut import_roots_sha256 = Vec::with_capacity(import_roots.len());
96    for root in import_roots {
97        import_roots_sha256.push(fingerprint_guarded(
98            &canonical_directory(root, guards)?,
99            &mut budget,
100            guards,
101        )?);
102    }
103    let binding = PythonEnvironmentBinding {
104        version: 1,
105        executable,
106        handler: handler.to_owned(),
107        runtime_sha256,
108        import_roots_sha256,
109    };
110    binding.validate()?;
111    Ok(binding)
112}
113
114pub(super) fn verify(
115    config: &LocalPythonWorkerConfig,
116    descriptor: &ProcessFunctionDescriptor,
117    handler_directory: &Path,
118    manifest: &Path,
119    module: &str,
120    mut guards: FileGuards,
121) -> Result<VerifiedEnvironment, DbError> {
122    if config.python_paths.len() >= MAX_PYTHON_IMPORT_ROOTS {
123        return Err(DbError::Config(
124            "Python worker supports at most 15 explicit import roots".into(),
125        ));
126    }
127    let mut import_roots = vec![handler_directory.to_path_buf()];
128    for root in &config.python_paths {
129        let root = if descriptor.python_environment.is_some() {
130            canonical_directory(root, &mut guards)?
131        } else {
132            root.canonicalize()
133                .map_err(|error| DbError::Config(format!("resolve Python import root: {error}")))?
134        };
135        import_roots.push(root);
136    }
137    match (&descriptor.python_environment, &config.runtime_root) {
138        (None, None) => Ok(VerifiedEnvironment {
139            python: config.python.clone(),
140            runtime_root: None,
141            import_roots,
142            _guards: guards,
143        }),
144        (Some(expected), Some(root)) => {
145            let runtime_root = canonical_directory(root, &mut guards)?;
146            if manifest.starts_with(&runtime_root)
147                || import_roots.iter().any(|root| manifest.starts_with(root))
148            {
149                return Err(DbError::Config(
150                    "bound process manifest must be outside the hashed trees".into(),
151                ));
152            }
153            let actual = capture(
154                &runtime_root,
155                &config.python,
156                &format!("{module}:{}", config.function),
157                &import_roots,
158                &mut guards,
159            )?;
160            if &actual != expected {
161                return Err(DbError::Config(
162                    "Python environment differs from its process manifest".into(),
163                ));
164            }
165            let python = runtime_root.join(&actual.executable);
166            Ok(VerifiedEnvironment {
167                python,
168                runtime_root: Some(runtime_root),
169                import_roots,
170                _guards: guards,
171            })
172        }
173        _ => Err(DbError::Config(
174            "Python environment binding and runtime_root must be supplied together".into(),
175        )),
176    }
177}
178
179#[cfg(test)]
180mod tests {
181    use super::*;
182    use std::time::Duration;
183
184    fn deployment(root: &Path) -> (LocalPythonWorkerConfig, ProcessFunctionDescriptor) {
185        let root = root.canonicalize().unwrap();
186        let runtime = root.join("runtime");
187        let handlers = root.join("handlers");
188        let sdk = root.join("sdk");
189        for path in [&runtime, &handlers, &sdk] {
190            std::fs::create_dir(path).unwrap();
191        }
192        let python = runtime.join("python");
193        std::fs::write(&python, b"interpreter").unwrap();
194        std::fs::write(runtime.join("stdlib.py"), b"runtime").unwrap();
195        let handler_file = handlers.join("handler.py");
196        std::fs::write(&handler_file, b"handler").unwrap();
197        std::fs::write(sdk.join("worker.py"), b"sdk").unwrap();
198        let manifest = root.join("manifest.json");
199        std::fs::write(&manifest, []).unwrap();
200        let mut descriptor = ProcessFunctionDescriptor::from_manifest_json(
201            &std::fs::read(
202                PathBuf::from(env!("CARGO_MANIFEST_DIR"))
203                    .join("../../examples/process_python/manifest.json"),
204            )
205            .unwrap(),
206        )
207        .unwrap();
208        descriptor.python_environment = Some(
209            PythonEnvironmentBinding::capture(
210                &runtime,
211                &python,
212                "handler:handle",
213                &[handlers, sdk.clone()],
214            )
215            .unwrap(),
216        );
217        (
218            LocalPythonWorkerConfig {
219                python,
220                runtime_root: Some(runtime),
221                manifest,
222                handler_file,
223                function: "handle".into(),
224                python_paths: vec![sdk],
225                max_in_flight: 1,
226                timeout: Duration::from_secs(5),
227            },
228            descriptor,
229        )
230    }
231
232    fn check(
233        config: &LocalPythonWorkerConfig,
234        descriptor: &ProcessFunctionDescriptor,
235    ) -> Result<VerifiedEnvironment, DbError> {
236        verify(
237            config,
238            descriptor,
239            config.handler_file.parent().unwrap(),
240            &config.manifest,
241            "handler",
242            FileGuards::default(),
243        )
244    }
245
246    #[test]
247    fn replay_binding_rejects_writable_deployment() {
248        let root = tempfile::tempdir().unwrap();
249        let (config, _) = deployment(root.path());
250        let result = FileGuards::for_replay()
251            .and_then(|mut guards| guards.canonical_file(&config.python).map(|_| ()));
252        assert!(result.is_err());
253        assert!(result.unwrap_err().to_string().contains("read-only"));
254    }
255
256    #[test]
257    fn binding_rejects_runtime_sdk_addition_and_removal_drift() {
258        let root = tempfile::tempdir().unwrap();
259        let (config, descriptor) = deployment(root.path());
260        check(&config, &descriptor).unwrap();
261        let stdlib = config.runtime_root.as_ref().unwrap().join("stdlib.py");
262        std::fs::write(&stdlib, b"changed").unwrap();
263        assert!(check(&config, &descriptor)
264            .err()
265            .unwrap()
266            .to_string()
267            .contains("environment differs"));
268        std::fs::write(&stdlib, b"runtime").unwrap();
269        let sdk = config.python_paths[0].join("worker.py");
270        std::fs::write(&sdk, b"SDK").unwrap();
271        assert!(check(&config, &descriptor).is_err());
272        std::fs::write(&sdk, b"sdk").unwrap();
273        let extra = config.python_paths[0].join("new.pyc");
274        std::fs::write(&extra, b"new import").unwrap();
275        assert!(check(&config, &descriptor).is_err());
276        std::fs::remove_file(extra).unwrap();
277        check(&config, &descriptor).unwrap();
278        std::fs::remove_file(sdk).unwrap();
279        assert!(check(&config, &descriptor).is_err());
280    }
281
282    #[test]
283    fn binding_is_relocatable_and_requires_both_launch_and_manifest_fields() {
284        let first = tempfile::tempdir().unwrap();
285        let second = tempfile::tempdir().unwrap();
286        let (config, descriptor) = deployment(first.path());
287        let (mut moved, equivalent) = deployment(second.path());
288        assert_eq!(descriptor.python_environment, equivalent.python_environment);
289        check(&moved, &descriptor).unwrap();
290        moved.runtime_root = None;
291        assert!(check(&moved, &descriptor).is_err());
292        let mut unbound = descriptor.clone();
293        unbound.python_environment = None;
294        assert!(check(&config, &unbound).is_err());
295        check(&moved, &unbound).unwrap();
296        let mut changed_entrypoint = config;
297        changed_entrypoint.function = "different_handler".into();
298        assert!(check(&changed_entrypoint, &descriptor).is_err());
299    }
300
301    #[test]
302    fn binding_rejects_uncontained_executable_and_self_referential_manifest() {
303        let root = tempfile::tempdir().unwrap();
304        let (mut config, descriptor) = deployment(root.path());
305        let outside = root.path().join("outside");
306        std::fs::write(&outside, b"interpreter").unwrap();
307        assert!(PythonEnvironmentBinding::capture(
308            config.runtime_root.as_ref().unwrap(),
309            &outside,
310            "handler:handle",
311            &config.python_paths
312        )
313        .unwrap_err()
314        .to_string()
315        .contains("outside"));
316        config.manifest = config.python_paths[0].join("manifest.json");
317        assert!(check(&config, &descriptor)
318            .err()
319            .unwrap()
320            .to_string()
321            .contains("outside the hashed trees"));
322        assert!(PythonEnvironmentBinding::capture(
323            config.runtime_root.as_ref().unwrap(),
324            &config.python,
325            "handler:handle",
326            &vec![config.python_paths[0].clone(); 17]
327        )
328        .is_err());
329    }
330
331    #[cfg(windows)]
332    #[test]
333    fn verified_environment_retains_file_guards_and_failed_checks_release_them() {
334        let root = tempfile::tempdir().unwrap();
335        let (config, descriptor) = deployment(root.path());
336        let files = [
337            config.python.clone(),
338            config.runtime_root.as_ref().unwrap().join("stdlib.py"),
339            config.handler_file.clone(),
340            config.python_paths[0].join("worker.py"),
341        ];
342        let environment = check(&config, &descriptor).unwrap();
343        for file in &files {
344            assert_eq!(
345                std::fs::write(file, b"changed").unwrap_err().raw_os_error(),
346                Some(32)
347            );
348        }
349        drop(environment);
350        std::fs::write(&files[3], b"changed").unwrap();
351        assert!(check(&config, &descriptor).is_err());
352        for file in &files {
353            std::fs::OpenOptions::new().write(true).open(file).unwrap();
354        }
355    }
356}
laminar_db/process_function/remote/python_environment/mod.rs

laminar_db/process_function/remote/python_environment/
mod.rs
