You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/10/22 05:15:13 UTC
(maka) branch rust-preview updated: fix(cli): preserve validated initialization settings through observer
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch rust-preview
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/rust-preview by this push:
new 67286c167f fix(cli): preserve validated initialization settings through observer
67286c167f is described below
commit 67286c167f355f7e4d4d49f9bce358c81d009cee
Author: Wang <m4...@gmail.com>
AuthorDate: Thu Oct 22 13:14:54 2026 +0800
fix(cli): preserve validated initialization settings through observer
---
products/maka/cli/src/args.rs | 3 +-
products/maka/cli/src/initialize.rs | 43 ++++--
products/maka/cli/src/operation.rs | 112 +++++++++-------
products/maka/cli/tests/integration/initialize.rs | 153 ++++++++++++++++++++++
products/maka/cli/tests/integration/main.rs | 1 +
scripts/rust/host-comparison/README.md | 6 +-
scripts/rust/host-comparison/compare.py | 4 +-
7 files changed, 258 insertions(+), 64 deletions(-)
diff --git a/products/maka/cli/src/args.rs b/products/maka/cli/src/args.rs
index 8c53065c10..5d4ec99216 100644
--- a/products/maka/cli/src/args.rs
+++ b/products/maka/cli/src/args.rs
@@ -183,7 +183,8 @@ impl Cli {
};
let timeout = std::time::Duration::from_millis(self.timeout_ms.unwrap_or(default));
if !self.operation_worker {
- return crate::operation::observe(timeout)
+ let settings_stdin = matches!(&self.command, Some(Command::Host(HostCommand::Init(args))) if args.settings_stdin);
+ return crate::operation::observe(timeout, settings_stdin)
.await
.map(|()| std::process::ExitCode::SUCCESS);
}
diff --git a/products/maka/cli/src/initialize.rs b/products/maka/cli/src/initialize.rs
index 15fcf7f729..a36760fc59 100644
--- a/products/maka/cli/src/initialize.rs
+++ b/products/maka/cli/src/initialize.rs
@@ -41,9 +41,11 @@ pub(super) struct Init {
root: Root,
/// Read initial privacy/proxy settings from stdin. Requires a new, empty root.
#[arg(long)]
- settings_stdin: bool,
+ pub(super) settings_stdin: bool,
}
+pub(crate) const SETTINGS_LIMIT: u64 = 65536;
+
/// Bootstrap secrets travel through stdin, never command arguments or diagnostics.
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
@@ -57,19 +59,7 @@ impl Init {
pub async fn run(self) -> Result<(), HostError> {
let namespaces = RootNamespaces::for_current_account()?;
let id = if self.settings_stdin {
- let settings = tokio::task::spawn_blocking(|| {
- let mut bytes = Vec::new();
- std::io::stdin()
- .lock()
- .take(65537)
- .read_to_end(&mut bytes)?;
- if bytes.len() > 65536 {
- return Err(std::io::Error::other("initial settings exceed 64 KiB"));
- }
- serde_json::from_slice::<Settings>(&bytes)
- .map_err(|_| std::io::Error::other("invalid initial settings"))
- })
- .await??;
+ let settings = Settings::parse(&settings_input().await?)?;
let update = settings.proxy_update()?;
// Never reconfigure an existing root or race its running Host.
let owner = Arc::new(RootOwner::create(&self.root.root, &namespaces)?);
@@ -100,7 +90,32 @@ impl Init {
}
}
+pub(crate) async fn settings_input() -> Result<Vec<u8>, HostError> {
+ Ok(tokio::task::spawn_blocking(|| {
+ let mut bytes = Vec::new();
+ std::io::stdin()
+ .lock()
+ .take(SETTINGS_LIMIT + 1)
+ .read_to_end(&mut bytes)?;
+ Ok::<_, std::io::Error>(bytes)
+ })
+ .await??)
+}
+
+// Validate the entire input before handing it to a detached worker. If replay
+// is cut short, only trailing whitespace may be lost from an accepted JSON value.
+pub(crate) fn validate_settings_input(bytes: &[u8]) -> std::io::Result<()> {
+ Settings::parse(bytes).map(|_| ())
+}
+
impl Settings {
+ fn parse(bytes: &[u8]) -> std::io::Result<Self> {
+ if bytes.len() as u64 > SETTINGS_LIMIT {
+ return Err(std::io::Error::other("initial settings exceed 64 KiB"));
+ }
+ serde_json::from_slice(bytes).map_err(|_| std::io::Error::other("invalid initial settings"))
+ }
+
fn proxy_update(&self) -> Result<Option<Update>, HostError> {
let Some(proxy) = &self.network_proxy else {
return if self.proxy_password.is_some() {
diff --git a/products/maka/cli/src/operation.rs b/products/maka/cli/src/operation.rs
index 3abe5357e5..c8bc4d251b 100644
--- a/products/maka/cli/src/operation.rs
+++ b/products/maka/cli/src/operation.rs
@@ -23,7 +23,10 @@
use maka_host_runtime::HostError;
use serde::Serialize;
use std::{future::Future, io::Write, process::Stdio, time::Duration};
-use tokio::{io::AsyncReadExt, time::Instant};
+use tokio::{
+ io::{AsyncReadExt, AsyncWriteExt},
+ time::Instant,
+};
tokio::task_local! {
static DEADLINE: Instant;
@@ -92,7 +95,7 @@ pub(crate) fn progress(phase: Phase, completed: Option<u64>, total: Option<u64>)
/// Detach only this finite command so the caller can leave while accepted
/// blocking work retains its existing leases. Never terminate a discovered Host.
-pub(crate) async fn observe(timeout: Duration) -> Result<(), HostError> {
+pub(crate) async fn observe(timeout: Duration, settings_stdin: bool) -> Result<(), HostError> {
// Leave a small part of the same budget for the timeout diagnostic.
let deadline = Instant::now() + timeout.saturating_sub(Duration::from_millis(10));
// Only this observer may be exited at the deadline: it owns no Root or
@@ -105,51 +108,70 @@ pub(crate) async fn observe(timeout: Duration) -> Result<(), HostError> {
std::thread::sleep(timeout);
std::process::exit(70);
})?;
- let mut command = tokio::process::Command::new(std::env::current_exe()?);
- command
- .arg("--operation-worker")
- .args(std::env::args_os().skip(1))
- .stdin(Stdio::null())
- .stdout(Stdio::piped())
- .stderr(Stdio::piped())
- .kill_on_drop(false);
- #[cfg(unix)]
- // SAFETY: setsid is async-signal-safe and accesses no Rust state.
- unsafe {
- command.pre_exec(|| {
- if libc::setsid() == -1 {
- return Err(std::io::Error::last_os_error());
+ let result = tokio::time::timeout_at(deadline, async {
+ // Read and validate bootstrap input before detaching any work. Incomplete
+ // input cannot leave a worker waiting on the caller after this deadline.
+ let input = if settings_stdin {
+ let bytes = crate::initialize::settings_input().await?;
+ crate::initialize::validate_settings_input(&bytes)?;
+ Some(bytes)
+ } else {
+ None
+ };
+ let mut command = tokio::process::Command::new(std::env::current_exe()?);
+ command
+ .arg("--operation-worker")
+ .args(std::env::args_os().skip(1))
+ .stdin(if input.is_some() {
+ Stdio::piped()
+ } else {
+ Stdio::null()
+ })
+ .stdout(Stdio::piped())
+ .stderr(Stdio::piped())
+ .kill_on_drop(false);
+ #[cfg(unix)]
+ // SAFETY: setsid is async-signal-safe and accesses no Rust state.
+ unsafe {
+ command.pre_exec(|| {
+ if libc::setsid() == -1 {
+ return Err(std::io::Error::last_os_error());
+ }
+ Ok(())
+ });
+ }
+ #[cfg(windows)]
+ command.creation_flags(
+ windows_sys::Win32::System::Threading::DETACHED_PROCESS
+ | windows_sys::Win32::System::Threading::CREATE_NEW_PROCESS_GROUP
+ | windows_sys::Win32::System::Threading::CREATE_BREAKAWAY_FROM_JOB,
+ );
+ let child = command.spawn();
+ #[cfg(windows)]
+ let child = match child {
+ Err(error) if error.raw_os_error() == Some(5) => {
+ // A finite worker can remain in an enclosing Job: stopping this
+ // observer does not close somebody else's Job. External Job
+ // termination is ordinary crash recovery from committed state.
+ // Long-lived Host activation still requires independent ownership
+ // and refuses its own breakaway failure; do not weaken that rule.
+ command.creation_flags(
+ windows_sys::Win32::System::Threading::CREATE_NO_WINDOW
+ | windows_sys::Win32::System::Threading::CREATE_NEW_PROCESS_GROUP,
+ );
+ command.spawn()
}
- Ok(())
- });
- }
- #[cfg(windows)]
- command.creation_flags(
- windows_sys::Win32::System::Threading::DETACHED_PROCESS
- | windows_sys::Win32::System::Threading::CREATE_NEW_PROCESS_GROUP
- | windows_sys::Win32::System::Threading::CREATE_BREAKAWAY_FROM_JOB,
- );
- let child = command.spawn();
- #[cfg(windows)]
- let child = match child {
- Err(error) if error.raw_os_error() == Some(5) => {
- // A finite worker can remain in an enclosing Job: stopping this
- // observer does not close somebody else's Job. External Job
- // termination is ordinary crash recovery from committed state.
- // Long-lived Host activation still requires independent ownership
- // and refuses its own breakaway failure; do not weaken that rule.
- command.creation_flags(
- windows_sys::Win32::System::Threading::CREATE_NO_WINDOW
- | windows_sys::Win32::System::Threading::CREATE_NEW_PROCESS_GROUP,
- );
- command.spawn()
+ result => result,
+ };
+ let mut child = child?;
+ if let Some(input) = input {
+ let mut stdin = child.stdin.take().ok_or("missing command stdin")?;
+ stdin.write_all(&input).await?;
+ // EOF marks the end of the complete, validated settings document.
+ drop(stdin);
}
- result => result,
- };
- let mut child = child?;
- let stdout = child.stdout.take().ok_or("missing command stdout")?;
- let stderr = child.stderr.take().ok_or("missing command stderr")?;
- let result = tokio::time::timeout_at(deadline, async {
+ let stdout = child.stdout.take().ok_or("missing command stdout")?;
+ let stderr = child.stderr.take().ok_or("missing command stderr")?;
let (status, (), ()) =
tokio::try_join!(child.wait(), forward(stdout, false), forward(stderr, true),)?;
if !status.success() {
diff --git a/products/maka/cli/tests/integration/initialize.rs b/products/maka/cli/tests/integration/initialize.rs
new file mode 100644
index 0000000000..4f82de97a1
--- /dev/null
+++ b/products/maka/cli/tests/integration/initialize.rs
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::{
+ io::Write,
+ path::Path,
+ process::{Child, Command, Stdio},
+ sync::Arc,
+ time::{Duration, Instant},
+};
+
+use maka_config::ConfigurationStore;
+use maka_event_log::root::{RootNamespaces, RootOwner};
+
+fn initialize(root: &Path, timeout: &str) -> Child {
+ Command::new(env!("CARGO_BIN_EXE_maka"))
+ .args([
+ "host",
+ "init",
+ "--settings-stdin",
+ "--timeout-ms",
+ timeout,
+ "--root",
+ ])
+ .arg(root)
+ .stdin(Stdio::piped())
+ .stdout(Stdio::piped())
+ .stderr(Stdio::piped())
+ .spawn()
+ .unwrap()
+}
+
+#[tokio::test]
+async fn public_init_persists_settings_from_stdin_and_rejects_reconfiguration() {
+ let directory = tempfile::tempdir().unwrap();
+ let root = directory.path().join("root");
+ let mut child = initialize(&root, "15000");
+ let mut input = br#"{"privacy":{"incognitoActive":true}}"#.to_vec();
+ input.resize(65536, b' ');
+ child.stdin.take().unwrap().write_all(&input).unwrap();
+ let output = child.wait_with_output().unwrap();
+ assert!(
+ output.status.success(),
+ "{}",
+ String::from_utf8_lossy(&output.stderr)
+ );
+ let response: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap();
+ let namespaces = RootNamespaces::for_current_account().unwrap();
+ let owner = Arc::new(RootOwner::open_existing(&root, &namespaces).unwrap());
+ assert_eq!(response["rootId"], owner.root_id());
+ let store = ConfigurationStore::for_root(owner).await.unwrap();
+ let policy = store.runtime_policy().await.unwrap();
+ assert_eq!(policy.revision, 1);
+ assert!(policy.policy.privacy.incognito_active);
+ store.close().await.unwrap();
+
+ let mut child = initialize(&root, "15000");
+ child
+ .stdin
+ .take()
+ .unwrap()
+ .write_all(br#"{"privacy":{"incognitoActive":false}}"#)
+ .unwrap();
+ assert!(!child.wait_with_output().unwrap().status.success());
+ let owner = Arc::new(RootOwner::open_existing(&root, &namespaces).unwrap());
+ let store = ConfigurationStore::for_root(owner).await.unwrap();
+ assert_eq!(store.runtime_policy().await.unwrap(), policy);
+ store.close().await.unwrap();
+}
+
+#[test]
+fn public_init_rejects_invalid_and_oversized_settings_without_exposing_input() {
+ let mut oversized = br#"{"privacy":{"incognitoActive":true}}"#.to_vec();
+ oversized.resize(65537, b' ');
+ let mut invalid_suffix = br#"{"privacy":{"incognitoActive":true}}"#.to_vec();
+ invalid_suffix.resize(60000, b' ');
+ invalid_suffix.extend_from_slice(b"private-secret-invalid-suffix");
+ for input in [
+ b"private-secret-invalid-json".to_vec(),
+ oversized,
+ invalid_suffix,
+ ] {
+ let directory = tempfile::tempdir().unwrap();
+ let root = directory.path().join("root");
+ let mut child = initialize(&root, "15000");
+ // An early rejection may close its pipe before the entire input is written.
+ let _ = child.stdin.take().unwrap().write_all(&input);
+ let output = child.wait_with_output().unwrap();
+ assert!(!output.status.success());
+ let diagnostic = String::from_utf8_lossy(&output.stderr);
+ assert!(
+ diagnostic.contains(if input.len() > 65536 {
+ "initial settings exceed 64 KiB"
+ } else {
+ "invalid initial settings"
+ }),
+ "{diagnostic}"
+ );
+ assert!(!diagnostic.contains("private-secret"));
+ assert!(!root.exists());
+ }
+}
+
+#[test]
+fn public_init_times_out_unfinished_input_without_leaving_a_reader_or_root() {
+ let directory = tempfile::tempdir().unwrap();
+ let root = directory.path().join("root");
+ let start = Instant::now();
+ let mut child = initialize(&root, "1000");
+ let mut input = child.stdin.take().unwrap();
+ // Even valid JSON is not accepted until the caller finishes its input.
+ input
+ .write_all(br#"{"privacy":{"incognitoActive":true}}"#)
+ .unwrap();
+ loop {
+ if child.try_wait().unwrap().is_some() {
+ break;
+ }
+ if start.elapsed() > Duration::from_secs(15) {
+ let _ = child.kill();
+ let _ = child.wait();
+ panic!("unfinished stdin defeated the operation deadline");
+ }
+ std::thread::sleep(Duration::from_millis(10));
+ }
+ let output = child.wait_with_output().unwrap();
+ assert!(!output.status.success());
+ assert!(
+ output.status.code() == Some(70)
+ || String::from_utf8_lossy(&output.stderr).contains("observation timed out"),
+ "{}",
+ String::from_utf8_lossy(&output.stderr)
+ );
+ assert!(!root.exists());
+ // A detached worker still reading inherited stdin would keep this pipe open.
+ assert!(input.write_all(b" ").is_err());
+}
diff --git a/products/maka/cli/tests/integration/main.rs b/products/maka/cli/tests/integration/main.rs
index f7fb6f6c02..15f58b4c6a 100644
--- a/products/maka/cli/tests/integration/main.rs
+++ b/products/maka/cli/tests/integration/main.rs
@@ -22,6 +22,7 @@ mod candidate;
mod code_command;
#[cfg(unix)]
mod desktop;
+mod initialize;
#[allow(dead_code)]
#[path = "../../../../../platform/transport/tests/support/mod.rs"]
mod local_fixture;
diff --git a/scripts/rust/host-comparison/README.md b/scripts/rust/host-comparison/README.md
index da59ebb856..ee715d1d50 100644
--- a/scripts/rust/host-comparison/README.md
+++ b/scripts/rust/host-comparison/README.md
@@ -28,7 +28,11 @@ tokens, API cost, or a general advantage of one implementation language.
The TS adapter currently targets main commit
`0d19f242336a31e8aeb300372d3d3c896b1e99ef`. The initial Rust measurements use
`97405171ea8f10cfc0ec8ae469cfca559402ed90`. Record different revisions explicitly;
-the public protocols may change. Production paths remain unchanged by the harness.
+the public protocols may change. The current harness uses public
+`host init --settings-stdin` and requires the CLI's settings-input forwarding fix.
+The initial frozen campaign used its recorded worker-entry workaround during
+untimed setup; retain that harness with those receipts. Production paths remain
+unchanged by the harness.
## Build and run
diff --git a/scripts/rust/host-comparison/compare.py b/scripts/rust/host-comparison/compare.py
index 3b370300a0..c1ecba338f 100755
--- a/scripts/rust/host-comparison/compare.py
+++ b/scripts/rust/host-comparison/compare.py
@@ -1200,11 +1200,9 @@ class RustHost:
self.process = None
async def initialize(self):
- # Current finite-command observer discards stdin. Invoke the same CLI init
- # worker directly during untimed setup; retain the failed public-init receipt.
+ # Exercise the public initializer, including its input and observation boundary.
proc = await asyncio.create_subprocess_exec(
str(RUST / "target/release/maka"),
- "--operation-worker",
"host",
"init",
"--root",
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org