From c30337375266bf4f438a17d1159c961441376811 Mon Sep 17 00:00:00 2001 From: wenzr <282277167@qq.com> Date: Sun, 9 Aug 2026 09:06:06 +0800 Subject: [PATCH] =?UTF-8?q?agent:=20=E9=99=90=E5=88=B6=E5=B7=A5=E5=85=B7?= =?UTF-8?q?=E8=B5=84=E6=BA=90=E5=B9=B6=E6=94=B6=E7=B4=A7=E7=BD=91=E7=BB=9C?= =?UTF-8?q?=E4=BF=A1=E4=BB=BB=E8=BE=B9=E7=95=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 1 - README.md | 7 +- crates/kuncode-agent/src/tool/bash.rs | 429 +++++++++++++++--- .../kuncode-agent/src/tool/filesystem/glob.rs | 3 +- .../kuncode-agent/src/tool/filesystem/ls.rs | 3 +- .../src/tool/filesystem/read_file.rs | 257 ++++++++++- crates/kuncode-agent/src/tool/web_fetch.rs | 6 +- .../src/tool/web_fetch/address.rs | 64 +-- .../kuncode-agent/src/tool/web_fetch/tests.rs | 86 ++++ crates/kuncode-cli/Cargo.toml | 1 - crates/kuncode-cli/src/main.rs | 1 - 11 files changed, 715 insertions(+), 143 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 662ba75..63efeff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1894,7 +1894,6 @@ dependencies = [ "async-trait", "clap", "crossterm", - "dotenvy", "futures-util", "kuncode-agent", "kuncode-core", diff --git a/README.md b/README.md index 1f6600b..1d63a33 100644 --- a/README.md +++ b/README.md @@ -35,11 +35,8 @@ kuncode-cli ──▶ kuncode-agent ──▶ kuncode-core ──▶ LLM API export DEEPSEEK_API_KEY="your-api-key" ``` -项目会自动读取当前目录下的 `.env`，因此也可以将变量写入本地 `.env`： - -```dotenv -DEEPSEEK_API_KEY=your-api-key -``` +KunCode 只读取启动进程已有的环境变量，不会自动加载工作区中的 `.env`。 +请在 shell、终端配置或可信的进程管理器中注入密钥，避免未受信项目改变网络与运行时配置。 使用 OpenAI 官方接口时，在 `.kuncode/settings.json` 配置： diff --git a/crates/kuncode-agent/src/tool/bash.rs b/crates/kuncode-agent/src/tool/bash.rs index 1f33a99..d4b1151 100644 --- a/crates/kuncode-agent/src/tool/bash.rs +++ b/crates/kuncode-agent/src/tool/bash.rs @@ -1,11 +1,22 @@ -use std::{process::Stdio, time::Duration}; +//! Executes approved shell commands with bounded output and descendant cleanup. + +use std::{ + io, + process::{ExitStatus, Stdio}, + time::Duration, +}; use async_trait::async_trait; use kuncode_core::completion::ToolDefinition; use kuncode_core::non_empty_vec::NonEmptyVec; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; -use tokio::{process::Command, time::timeout}; +use thiserror::Error; +use tokio::{ + io::{AsyncRead, AsyncReadExt}, + process::{Child, ChildStderr, ChildStdout, Command}, + time::timeout, +}; use crate::{ permission::{ @@ -67,6 +78,57 @@ impl Bash { pub fn workspace(&self) -> &Workspace { &self.workspace } + + async fn run_command(&self, cmd: String, command_timeout: Duration) -> ToolOutput { + let mut command = Command::new("bash"); + command + .arg("-lc") + .arg(&cmd) + .current_dir(self.workspace.root()) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + + let output = match capture_command(command, command_timeout).await { + Ok(output) => output, + Err(error) => { + let kind = match error { + CommandExecutionError::Execution(_) => "execution", + CommandExecutionError::Timeout { .. } => "timeout", + }; + return ToolOutput::failure(kind, error.to_string()); + } + }; + + let (stdout, stdout_truncated) = output_text("stdout", &output.stdout); + let (stderr, stderr_truncated) = output_text("stderr", &output.stderr); + let truncated = stdout_truncated || stderr_truncated; + let ok = output.status.success(); + let exit_code = output.status.code(); + + ToolOutput { + ok, + data: Some(BashOutput { + cmd, + exit_code, + stdout, + stderr, + }), + error: if ok { + None + } else { + Some(ToolErrorPayload { + kind: "non_zero_exit".into(), + message: match exit_code { + Some(code) => format!("command exited with status {code}"), + None => "command terminated by signal".to_string(), + }, + }) + }, + truncated, + } + } } #[async_trait] @@ -110,80 +172,202 @@ impl TypedTool for Bash { } async fn run_prepared(&self, prepared: BashArgs, _ctx: &ToolContext) -> ToolOutput { - let cmd = prepared.cmd; + self.run_command(prepared.cmd, COMMAND_TIMEOUT).await + } +} - let mut command = Command::new("bash"); - command - .arg("-lc") - .arg(&cmd) - .current_dir(self.workspace.root()) - .stdin(Stdio::null()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .kill_on_drop(true); +#[derive(Debug, Error)] +enum CommandExecutionError { + #[error("failed to run command: {0}")] + Execution(#[source] io::Error), + #[error("command exceeded {seconds} seconds")] + Timeout { seconds: u64 }, +} - let output = match timeout(COMMAND_TIMEOUT, command.output()).await { - Ok(Ok(output)) => output, - Ok(Err(err)) => { - return ToolOutput::failure("execution", format!("failed to run command: {err}")); - } - Err(_) => { - return ToolOutput::failure( - "timeout", - format!("command exceeded {} seconds", COMMAND_TIMEOUT.as_secs()), - ); - } - }; +#[derive(Debug)] +struct CapturedCommand { + status: ExitStatus, + stdout: CapturedStream, + stderr: CapturedStream, +} - let (stdout, stdout_truncated) = output_text("stdout", &output.stdout); - let (stderr, stderr_truncated) = output_text("stderr", &output.stderr); - let truncated = stdout_truncated || stderr_truncated; - let ok = output.status.success(); - let exit_code = output.status.code(); +#[derive(Debug)] +struct CapturedStream { + prefix: Vec, + total_bytes: u64, +} - ToolOutput { - ok, - data: Some(BashOutput { - cmd, - exit_code, - stdout, - stderr, - }), - error: if ok { - None - } else { - Some(ToolErrorPayload { - kind: "non_zero_exit".into(), - message: match exit_code { - Some(code) => format!("command exited with status {code}"), - None => "command terminated by signal".to_string(), - }, - }) - }, - truncated, +impl CapturedStream { + fn truncated(&self) -> bool { + self.total_bytes > self.prefix.len() as u64 + } +} + +struct ManagedChild { + child: Child, + cleanup_armed: bool, + #[cfg(unix)] + process_group: Option, +} + +impl ManagedChild { + fn new(child: Child) -> Self { + #[cfg(unix)] + let process_group = child.id().and_then(|id| libc::pid_t::try_from(id).ok()); + + Self { + child, + cleanup_armed: true, + #[cfg(unix)] + process_group, } } + + fn take_stdout(&mut self) -> Option { + self.child.stdout.take() + } + + fn take_stderr(&mut self) -> Option { + self.child.stderr.take() + } + + async fn wait(&mut self) -> io::Result { + self.child.wait().await + } + + fn signal_termination(&mut self) { + #[cfg(unix)] + if let Some(process_group) = self.process_group { + // SAFETY: `process_group` is the positive PID assigned by the OS to + // the child and configured as its PGID before spawn. `killpg` only + // reads these scalar arguments and targets that isolated group. + let _ = unsafe { libc::killpg(process_group, libc::SIGKILL) }; + } + + // Keep this fallback on Unix too: if the group signal races process + // setup or fails, Tokio can still terminate the direct child. + let _ = self.child.start_kill(); + } + + async fn terminate_and_reap(&mut self) { + self.signal_termination(); + let _ = self.child.wait().await; + self.cleanup_armed = false; + } + + fn disarm(&mut self) { + self.cleanup_armed = false; + } } -/// Decodes a captured stream, capping it at `OUTPUT_LIMIT_BYTES`. Bash output -/// may not be valid UTF-8, so decoding is intentionally lossy (`from_utf8_lossy`). +impl Drop for ManagedChild { + fn drop(&mut self) { + if self.cleanup_armed { + self.signal_termination(); + } + } +} + +async fn capture_command( + mut command: Command, + command_timeout: Duration, +) -> Result { + // A separate process group lets cancellation and timeout include shell + // pipelines, background jobs, and grandchildren rather than only `bash`. + #[cfg(unix)] + command.process_group(0); + + let child = command.spawn().map_err(CommandExecutionError::Execution)?; + let mut child = ManagedChild::new(child); + let Some(stdout) = child.take_stdout() else { + child.terminate_and_reap().await; + return Err(CommandExecutionError::Execution(io::Error::other( + "stdout pipe was not captured", + ))); + }; + let Some(stderr) = child.take_stderr() else { + child.terminate_and_reap().await; + return Err(CommandExecutionError::Execution(io::Error::other( + "stderr pipe was not captured", + ))); + }; + + let capture = async { + // Leave the direct child unreaped until both pipes reach EOF. Its PID + // therefore cannot be reused while cleanup still addresses the PGID. + let (stdout, stderr) = tokio::try_join!(capture_stream(stdout), capture_stream(stderr))?; + let status = child.wait().await?; + Ok::<_, io::Error>(CapturedCommand { + status, + stdout, + stderr, + }) + }; + + match timeout(command_timeout, capture).await { + Ok(Ok(output)) => { + child.disarm(); + Ok(output) + } + Ok(Err(error)) => { + child.terminate_and_reap().await; + Err(CommandExecutionError::Execution(error)) + } + Err(_) => { + child.terminate_and_reap().await; + Err(CommandExecutionError::Timeout { + seconds: command_timeout.as_secs(), + }) + } + } +} + +async fn capture_stream(mut reader: R) -> io::Result +where + R: AsyncRead + Unpin, +{ + let mut prefix = Vec::with_capacity(OUTPUT_LIMIT_BYTES); + let mut total_bytes = 0_u64; + let mut chunk = [0_u8; 8 * 1024]; + + loop { + let read = reader.read(&mut chunk).await?; + if read == 0 { + break; + } + total_bytes = total_bytes.saturating_add(read as u64); + let retained = (OUTPUT_LIMIT_BYTES - prefix.len()).min(read); + prefix.extend_from_slice(&chunk[..retained]); + } + + Ok(CapturedStream { + prefix, + total_bytes, + }) +} + +/// Decodes a stream whose retained prefix is capped at `OUTPUT_LIMIT_BYTES`. +/// Bash output may not be valid UTF-8, so decoding is intentionally lossy +/// (`from_utf8_lossy`). /// /// When the cap trips, a visible marker is appended naming the stream and the /// byte scale, so the model knows it holds only the head of the stream and must /// not assume it saw everything. How to get the rest (filter, redirect, re-run) /// is left to the model — bash is a general shell. -fn output_text(stream: &str, bytes: &[u8]) -> (String, bool) { - if bytes.len() <= OUTPUT_LIMIT_BYTES { - return (String::from_utf8_lossy(bytes).into_owned(), false); +fn output_text(stream: &str, captured: &CapturedStream) -> (String, bool) { + if !captured.truncated() { + return ( + String::from_utf8_lossy(&captured.prefix).into_owned(), + false, + ); } - // Slicing a byte slice at an arbitrary index never splits a `char` (that is - // a `str` concern); `from_utf8_lossy` turns any partial trailing sequence - // into U+FFFD, so the result is always valid UTF-8. - let mut text = String::from_utf8_lossy(&bytes[..OUTPUT_LIMIT_BYTES]).into_owned(); + // The retained byte prefix may end inside a code point; `from_utf8_lossy` + // turns that partial trailing sequence into U+FFFD. + let mut text = String::from_utf8_lossy(&captured.prefix).into_owned(); text.push_str(&format!( "\n…⟨kuncode: {stream} truncated — showed first {OUTPUT_LIMIT_BYTES} of {total} bytes⟩", - total = bytes.len(), + total = captured.total_bytes, )); (text, true) } @@ -334,10 +518,15 @@ fn is_dynamic_shell_command(command: &str) -> bool { #[cfg(test)] mod tests { - use std::sync::Arc; + use std::{sync::Arc, time::Duration}; + + use tokio::io::AsyncReadExt; - use super::{Bash, simple_command_chain}; + use super::{ + Bash, CapturedStream, OUTPUT_LIMIT_BYTES, capture_stream, output_text, simple_command_chain, + }; use crate::{ + test_support::TestDir, tool::{Tool, ToolContext, execute_for_test}, workspace::Workspace, }; @@ -424,6 +613,102 @@ mod tests { assert!(stdout.contains("stdout truncated")); } + #[tokio::test] + async fn stream_capture_retains_only_the_bounded_prefix() { + let total_bytes = (OUTPUT_LIMIT_BYTES as u64) * 50; + let input = tokio::io::repeat(b'x').take(total_bytes); + + let captured = capture_stream(input) + .await + .expect("in-memory stream should be readable"); + + assert_eq!(captured.prefix.len(), OUTPUT_LIMIT_BYTES); + assert_eq!(captured.total_bytes, total_bytes); + assert!(captured.prefix.iter().all(|byte| *byte == b'x')); + } + + #[cfg(unix)] + #[tokio::test] + async fn drains_large_stdout_and_stderr_concurrently() { + let bash = bash().await; + let out = bash + .run_command( + "(head -c 2000000 /dev/zero | tr '\\000' o) & \ + (head -c 2000000 /dev/zero | tr '\\000' e >&2) & wait" + .to_string(), + super::COMMAND_TIMEOUT, + ) + .await; + + assert!(out.ok); + assert!(out.truncated); + let data = out.data.expect("data present"); + assert!(data.stdout.starts_with(&"o".repeat(OUTPUT_LIMIT_BYTES))); + assert!(data.stdout.contains("of 2000000 bytes")); + assert!(data.stderr.starts_with(&"e".repeat(OUTPUT_LIMIT_BYTES))); + assert!(data.stderr.contains("of 2000000 bytes")); + } + + #[test] + fn output_decoding_remains_lossy_without_rewriting_controls() { + let captured = CapturedStream { + prefix: vec![b'a', 0xff, 0x1b], + total_bytes: 3, + }; + + assert_eq!(output_text("stdout", &captured), ("a�\u{1b}".into(), false)); + } + + #[cfg(unix)] + #[tokio::test] + async fn timeout_terminates_descendants() { + let tmp = TestDir::new(); + let bash = Bash::new(tmp.workspace().await); + let out = bash + .run_command( + "(printf ready > started; \ + while [ ! -e release ]; do sleep 0.01; done; \ + printf survived > descendant-survived) & wait" + .to_string(), + Duration::from_secs(1), + ) + .await; + + assert!(!out.ok); + assert_eq!(out.error.expect("error payload").kind.as_str(), "timeout"); + assert!(tmp.path().join("started").exists()); + release_and_assert_descendant_stopped(&tmp).await; + } + + #[cfg(unix)] + #[tokio::test] + async fn cancellation_terminates_descendants() { + let tmp = TestDir::new(); + let bash = Arc::new(Bash::new(tmp.workspace().await)); + let task = tokio::spawn({ + let bash = Arc::clone(&bash); + async move { + bash.run_command( + "(printf ready > started; \ + while [ ! -e release ]; do sleep 0.01; done; \ + printf survived > descendant-survived) & wait" + .to_string(), + Duration::from_secs(30), + ) + .await + } + }); + + wait_for_path(&tmp.path().join("started")).await; + task.abort(); + assert!( + task.await + .expect_err("aborted execution should be cancelled") + .is_cancelled() + ); + release_and_assert_descendant_stopped(&tmp).await; + } + #[tokio::test] async fn runs_commands_from_workspace_root() { let workspace = Workspace::new(std::env::current_dir().expect("current directory exists")) @@ -478,4 +763,26 @@ mod tests { assert_eq!(simple_command_chain(command), None, "{command}"); } } + + #[cfg(unix)] + async fn wait_for_path(path: &std::path::Path) { + for _ in 0..200 { + if path.exists() { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("{} should have been created", path.display()); + } + + #[cfg(unix)] + async fn release_and_assert_descendant_stopped(tmp: &TestDir) { + std::fs::write(tmp.path().join("release"), b"go").expect("release gate should be created"); + let survivor = tmp.path().join("descendant-survived"); + for _ in 0..100 { + assert!(!survivor.exists(), "descendant survived process-group kill"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert!(!survivor.exists(), "descendant survived process-group kill"); + } } diff --git a/crates/kuncode-agent/src/tool/filesystem/glob.rs b/crates/kuncode-agent/src/tool/filesystem/glob.rs index 842bf1d..06b1fe5 100644 --- a/crates/kuncode-agent/src/tool/filesystem/glob.rs +++ b/crates/kuncode-agent/src/tool/filesystem/glob.rs @@ -623,7 +623,8 @@ mod tests { ); } - #[cfg(unix)] + // Darwin rejects arbitrary non-UTF-8 path bytes before the tool can observe them. + #[cfg(target_os = "linux")] #[tokio::test] async fn names_that_are_not_utf8_are_counted_as_unconsidered() { use std::{ffi::OsStr, os::unix::ffi::OsStrExt}; diff --git a/crates/kuncode-agent/src/tool/filesystem/ls.rs b/crates/kuncode-agent/src/tool/filesystem/ls.rs index df804d1..4cf4562 100644 --- a/crates/kuncode-agent/src/tool/filesystem/ls.rs +++ b/crates/kuncode-agent/src/tool/filesystem/ls.rs @@ -1231,7 +1231,8 @@ mod tests { assert_eq!(paths, ["weird", "weird/name.rs", "weird\\name.rs"]); } - #[cfg(unix)] + // Darwin rejects arbitrary non-UTF-8 path bytes before the tool can observe them. + #[cfg(target_os = "linux")] #[tokio::test] async fn names_that_are_not_utf8_are_counted_instead_of_listed() { use std::{ffi::OsStr, os::unix::ffi::OsStrExt}; diff --git a/crates/kuncode-agent/src/tool/filesystem/read_file.rs b/crates/kuncode-agent/src/tool/filesystem/read_file.rs index 556e54b..0646ac0 100644 --- a/crates/kuncode-agent/src/tool/filesystem/read_file.rs +++ b/crates/kuncode-agent/src/tool/filesystem/read_file.rs @@ -1,6 +1,6 @@ //! The `read_file` tool: read a UTF-8 workspace file with line pagination. -use std::path::PathBuf; +use std::{io, path::PathBuf}; use async_trait::async_trait; use kuncode_core::completion::ToolDefinition; @@ -9,7 +9,7 @@ use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use tokio::{ fs::OpenOptions, - io::{AsyncBufReadExt, BufReader}, + io::{AsyncBufRead, AsyncBufReadExt, BufReader}, }; use super::helpers::{ @@ -21,7 +21,7 @@ use crate::{ }, tool::{ FileStamp, PreparationContext, PreparedInvocationState, ToolContext, ToolError, ToolOutput, - TypedPreparation, TypedTool, definition_for, output::truncate_utf8, + TypedPreparation, TypedTool, definition_for, }, workspace::Workspace, }; @@ -188,13 +188,13 @@ impl TypedTool for ReadFile { .as_ref() .map(FileStamp::from_metadata) .unwrap_or_default(); - let mut lines = BufReader::new(file).lines(); + let mut lines = BufReader::new(file); // Skip the lines before `start_line` without keeping them. Cost is // proportional to `start_line`, not file size; nothing past the // requested window is read. for _ in 0..(start_line - 1) { - match lines.next_line().await { + match read_bounded_line(&mut lines, 0).await { Ok(Some(_)) => {} // `start_line` is past EOF: there is simply nothing to return. Ok(None) => break, @@ -219,7 +219,7 @@ impl TypedTool for ReadFile { // A read error while peeking is a real failure (e.g. invalid // UTF-8 on the next line), not EOF — surface it like every other // read instead of reporting a false end-of-file via `has_more`. - has_more = match lines.next_line().await { + has_more = match read_bounded_line(&mut lines, 0).await { Ok(Some(_)) => true, Ok(None) => false, Err(err) => return io_error("read", &resolved, err, &self.workspace), @@ -227,14 +227,15 @@ impl TypedTool for ReadFile { break; } - let raw = match lines.next_line().await { + let raw = match read_bounded_line(&mut lines, MAX_LINE_BYTES).await { Ok(Some(line)) => line, Ok(None) => break, Err(err) => return io_error("read", &resolved, err, &self.workspace), }; - let raw_bytes = raw.len(); - let (mut line, line_truncated) = truncate_utf8(&raw, MAX_LINE_BYTES); + let raw_bytes = raw.total_bytes; + let mut line = raw.text; + let line_truncated = raw_bytes > line.len(); // Honor the total byte budget, but always return at least one line // so a single over-long line still yields its (capped) prefix. @@ -297,6 +298,144 @@ impl TypedTool for ReadFile { } } +#[derive(Debug)] +struct BoundedLine { + text: String, + total_bytes: usize, +} + +// Reads and validates one UTF-8 line while retaining only its bounded prefix. +// The discarded tail is still drained and validated so it cannot hide invalid +// UTF-8 or leave the reader in the middle of a line. +async fn read_bounded_line( + reader: &mut R, + retain_limit: usize, +) -> io::Result> +where + R: AsyncBufRead + Unpin, +{ + let retain_limit = retain_limit.min(MAX_LINE_BYTES); + let mut retained = Vec::with_capacity(retain_limit); + let mut validator = Utf8Validator::default(); + let mut total_bytes = 0usize; + let mut last_byte = None; + let mut terminated = false; + + loop { + let available = reader.fill_buf().await?; + if available.is_empty() { + break; + } + + let newline = available.iter().position(|byte| *byte == b'\n'); + let content_end = newline.unwrap_or(available.len()); + let content = &available[..content_end]; + + validator.push(content)?; + total_bytes = total_bytes.checked_add(content.len()).ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidData, "line length exceeds usize") + })?; + if let Some(byte) = content.last() { + last_byte = Some(*byte); + } + + let remaining = retain_limit.saturating_sub(retained.len()); + retained.extend_from_slice(&content[..content.len().min(remaining)]); + + let consumed = content_end + usize::from(newline.is_some()); + terminated = newline.is_some(); + reader.consume(consumed); + if terminated { + break; + } + } + + if total_bytes == 0 && !terminated { + return Ok(None); + } + validator.finish()?; + + // Match `AsyncBufReadExt::lines`: strip a carriage return only when it is + // immediately before a newline, after validating it as part of the input. + if terminated && last_byte == Some(b'\r') { + let unstripped_bytes = total_bytes; + total_bytes -= 1; + if retained.len() == unstripped_bytes { + retained.pop(); + } + } + + // A bounded prefix may end midway through an otherwise valid code point. + // Back up to the last complete boundary, matching `truncate_utf8` semantics. + let valid_prefix_len = match std::str::from_utf8(&retained) { + Ok(_) => retained.len(), + Err(error) if error.error_len().is_none() => error.valid_up_to(), + Err(_) => return Err(invalid_utf8_error()), + }; + retained.truncate(valid_prefix_len); + let text = String::from_utf8(retained).map_err(|_| invalid_utf8_error())?; + + Ok(Some(BoundedLine { text, total_bytes })) +} + +#[derive(Debug, Default)] +struct Utf8Validator { + pending: Vec, +} + +impl Utf8Validator { + fn push(&mut self, mut bytes: &[u8]) -> io::Result<()> { + if !self.pending.is_empty() { + let sequence_len = utf8_sequence_len(self.pending[0]).ok_or_else(invalid_utf8_error)?; + if self.pending.len() >= sequence_len { + return Err(invalid_utf8_error()); + } + let take = (sequence_len - self.pending.len()).min(bytes.len()); + self.pending.extend_from_slice(&bytes[..take]); + if self.pending.len() < sequence_len { + return Ok(()); + } + std::str::from_utf8(&self.pending).map_err(|_| invalid_utf8_error())?; + self.pending.clear(); + bytes = &bytes[take..]; + } + + if let Err(error) = std::str::from_utf8(bytes) { + if error.error_len().is_some() { + return Err(invalid_utf8_error()); + } + self.pending + .extend_from_slice(&bytes[error.valid_up_to()..]); + } + Ok(()) + } + + fn finish(self) -> io::Result<()> { + if self.pending.is_empty() { + Ok(()) + } else { + Err(invalid_utf8_error()) + } + } +} + +fn utf8_sequence_len(first_byte: u8) -> Option { + match first_byte { + 0x00..=0x7f => Some(1), + 0xc2..=0xdf => Some(2), + 0xe0..=0xef => Some(3), + 0xf0..=0xf4 => Some(4), + _ => None, + } +} + +fn invalid_utf8_error() -> io::Error { + io::Error::new( + io::ErrorKind::InvalidData, + "stream did not contain valid UTF-8", + ) +} + /// Inline marker appended to a line whose tail was dropped to fit /// `MAX_LINE_BYTES`. Deliberately explicit: the elided tail is neither in the /// returned content nor reachable via `next_line` (which advances by whole @@ -311,7 +450,9 @@ fn line_truncated_marker(elided_bytes: usize) -> String { mod tests { use std::{fs, sync::Arc}; - use super::{MAX_LINE_BYTES, ReadFile}; + use tokio::io::BufReader; + + use super::{MAX_LINE_BYTES, ReadFile, read_bounded_line}; use crate::test_support::TestDir; use crate::tool::{ToolContext, ToolOutput, execute_for_test}; @@ -371,6 +512,18 @@ mod tests { assert!(data["next_line"].is_null()); } + #[tokio::test] + async fn read_file_preserves_carriage_return_at_unterminated_eof() { + let tmp = TestDir::new(); + fs::write(tmp.path().join("notes.txt"), "tail\r").expect("file should be written"); + let tool = ReadFile::new(tmp.workspace().await); + + let output = call(tool, serde_json::json!({ "path": "notes.txt" })).await; + + assert!(output.ok); + assert_eq!(output.data.expect("data present")["content"], "tail\r"); + } + #[tokio::test] async fn read_file_start_past_end_returns_empty() { let tmp = TestDir::new(); @@ -394,7 +547,7 @@ mod tests { #[tokio::test] async fn read_file_truncates_an_overlong_line() { let tmp = TestDir::new(); - let long_line = "x".repeat(MAX_LINE_BYTES + 1_000); + let long_line = "x".repeat(4 * 1024 * 1024); fs::write(tmp.path().join("min.js"), &long_line).expect("file should be written"); let tool = ReadFile::new(tmp.workspace().await); @@ -415,6 +568,88 @@ mod tests { assert_eq!(data["truncated_lines"], serde_json::json!([1])); } + #[tokio::test] + async fn bounded_line_reader_retains_only_the_requested_prefix() { + let input = vec![b'x'; 4 * 1024 * 1024]; + // A deliberately small transport buffer forces the line and its UTF-8 + // validation state across many reads. + let mut reader = BufReader::with_capacity(257, input.as_slice()); + + let line = read_bounded_line(&mut reader, MAX_LINE_BYTES) + .await + .expect("line should be read") + .expect("line should be present"); + + assert_eq!(line.total_bytes, input.len()); + assert_eq!(line.text.len(), MAX_LINE_BYTES); + } + + #[tokio::test] + async fn bounded_line_reader_validates_code_points_across_buffers() { + let input = "你".repeat(MAX_LINE_BYTES); + // Five-byte buffers split successive three-byte code points at + // different offsets instead of accidentally preserving alignment. + let mut reader = BufReader::with_capacity(5, input.as_bytes()); + + let line = read_bounded_line(&mut reader, MAX_LINE_BYTES) + .await + .expect("line should be read") + .expect("line should be present"); + + assert_eq!(line.total_bytes, input.len()); + assert!(line.text.len() <= MAX_LINE_BYTES); + assert!(line.text.chars().all(|character| character == '你')); + } + + #[tokio::test] + async fn read_file_skips_an_overlong_line_without_retaining_it() { + let tmp = TestDir::new(); + let long_line = "x".repeat(4 * 1024 * 1024); + fs::write( + tmp.path().join("generated.txt"), + format!("{long_line}\r\nselected\r\ntrailing"), + ) + .expect("file should be written"); + let tool = ReadFile::new(tmp.workspace().await); + + let output = call( + tool, + serde_json::json!({ + "path": "generated.txt", + "start_line": 2, + "limit": 1 + }), + ) + .await; + + assert!(output.ok); + let data = output.data.expect("data present"); + assert_eq!(data["content"], "selected"); + assert_eq!(data["start_line"], 2); + assert_eq!(data["returned_lines"], 1); + assert_eq!(data["has_more"], true); + assert_eq!(data["next_line"], 3); + } + + #[tokio::test] + async fn read_file_rejects_invalid_utf8_in_a_discarded_line_tail() { + let tmp = TestDir::new(); + let mut body = vec![b'x'; 4 * 1024 * 1024]; + body.extend_from_slice(&[0xff, b'\n']); + body.extend_from_slice(b"selected\n"); + fs::write(tmp.path().join("mixed.bin"), body).expect("file should be written"); + let tool = ReadFile::new(tmp.workspace().await); + + let output = call( + tool, + serde_json::json!({ "path": "mixed.bin", "start_line": 2 }), + ) + .await; + + assert!(!output.ok); + assert_eq!(output.error.expect("error present").kind.as_str(), "read"); + } + #[tokio::test] async fn read_file_truncates_a_multibyte_line_on_a_char_boundary() { let tmp = TestDir::new(); diff --git a/crates/kuncode-agent/src/tool/web_fetch.rs b/crates/kuncode-agent/src/tool/web_fetch.rs index 0d4af40..6a80ea5 100644 --- a/crates/kuncode-agent/src/tool/web_fetch.rs +++ b/crates/kuncode-agent/src/tool/web_fetch.rs @@ -5,7 +5,8 @@ //! //! - **Which address may be dialed.** The private `address` module vets the //! resolved address, not the hostname, keeping the tool off cloud metadata and -//! private networks. +//! private networks. This client's system proxy discovery is disabled so every +//! target still passes through that resolver. //! - **Which origin was authorized.** A call is approved for one //! [`CanonicalOrigin`], so a redirect that leaves it is reported back instead //! of followed — the model can ask again for the new origin, and the user @@ -170,7 +171,8 @@ impl WebFetch { .connect_timeout(CONNECT_TIMEOUT) .timeout(REQUEST_TIMEOUT) .default_headers(headers) - .dns_resolver(address::GuardedResolver::from_environment()) + .no_proxy() + .dns_resolver(address::GuardedResolver::new()) .redirect(same_origin_redirects()) .build()?; Ok(Self { diff --git a/crates/kuncode-agent/src/tool/web_fetch/address.rs b/crates/kuncode-agent/src/tool/web_fetch/address.rs index d878752..2c45aa7 100644 --- a/crates/kuncode-agent/src/tool/web_fetch/address.rs +++ b/crates/kuncode-agent/src/tool/web_fetch/address.rs @@ -9,23 +9,9 @@ //! than to the hostname, so a public name that resolves (or re-resolves) inward //! is refused too. -use std::collections::BTreeSet; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use reqwest::dns::{Addrs, Name, Resolve, Resolving}; -use url::Url; - -/// Environment variables reqwest reads to discover a proxy. Their hosts are -/// exempt from [`is_blocked`]: the user configured them, and a corporate proxy -/// legitimately lives on a private address. -const PROXY_ENVIRONMENT_VARIABLES: [&str; 6] = [ - "ALL_PROXY", - "all_proxy", - "HTTPS_PROXY", - "https_proxy", - "HTTP_PROXY", - "http_proxy", -]; /// Reports whether `ip` names an address `web_fetch` must never dial. /// @@ -98,41 +84,24 @@ fn embedded_ipv4(address: Ipv6Addr) -> Option { /// hold under a redirect chain or a DNS record that answers publicly once and /// privately next: reqwest dials exactly the addresses returned here, so there /// is no second lookup to rebind. -/// -/// A proxy moves the boundary. Its hostname is exempt (see -/// [`PROXY_ENVIRONMENT_VARIABLES`]), and a proxied request resolves the *target* -/// on the proxy's side, so a hostname target is no longer vetted here at all — -/// only an IP-literal one is, and `web_fetch` checks those before it asks for -/// approval. -pub(super) struct GuardedResolver { - trusted_hosts: BTreeSet, -} +pub(super) struct GuardedResolver; impl GuardedResolver { - /// Builds a resolver that exempts the proxies configured for this process. - pub(super) fn from_environment() -> Self { - Self { - trusted_hosts: PROXY_ENVIRONMENT_VARIABLES - .iter() - .filter_map(|name| std::env::var(name).ok()) - .filter_map(|value| proxy_host(&value)) - .collect(), - } + /// Builds the resolver used by the direct-only `web_fetch` client. + pub(super) fn new() -> Self { + Self } } impl Resolve for GuardedResolver { fn resolve(&self, name: Name) -> Resolving { let host = name.as_str().to_ascii_lowercase(); - let trusted = self.trusted_hosts.contains(&host); Box::pin(async move { // Port 0: reqwest substitutes the URL's port, or the scheme default. let resolved = tokio::net::lookup_host((host.as_str(), 0)) .await? .collect::>(); - if !trusted - && let Some(blocked) = resolved.iter().find(|address| is_blocked(address.ip())) - { + if let Some(blocked) = resolved.iter().find(|address| is_blocked(address.ip())) { // One blocked answer fails the whole lookup instead of being // filtered out, so a record that mixes a public address with an // internal one cannot get the internal one dialed on a retry. @@ -147,19 +116,6 @@ impl Resolve for GuardedResolver { } } -/// Extracts the host from a proxy environment value, accepting the bare -/// `host:port` form those variables are also written in. -fn proxy_host(value: &str) -> Option { - let value = value.trim(); - if value.is_empty() { - return None; - } - let parsed = Url::parse(value) - .or_else(|_| Url::parse(&format!("http://{value}"))) - .ok()?; - parsed.host_str().map(str::to_ascii_lowercase) -} - #[cfg(test)] mod tests { use super::*; @@ -210,14 +166,4 @@ mod tests { assert!(!blocked(address), "{address} should be reachable"); } } - - #[test] - fn proxy_values_yield_a_host_in_both_spellings() { - assert_eq!( - proxy_host("http://Proxy.Corp.Internal:8080"), - Some("proxy.corp.internal".to_string()) - ); - assert_eq!(proxy_host("127.0.0.1:7890"), Some("127.0.0.1".to_string())); - assert_eq!(proxy_host(" "), None); - } } diff --git a/crates/kuncode-agent/src/tool/web_fetch/tests.rs b/crates/kuncode-agent/src/tool/web_fetch/tests.rs index 708d9a1..4940ea9 100644 --- a/crates/kuncode-agent/src/tool/web_fetch/tests.rs +++ b/crates/kuncode-agent/src/tool/web_fetch/tests.rs @@ -156,6 +156,92 @@ async fn fetch_with(arguments: serde_json::Value) -> ToolOutput { .expect("no harness-level error") } +#[test] +fn system_proxy_environment_is_ignored() { + const CHILD_ENV: &str = "KUNCODE_WEB_FETCH_NO_PROXY_TEST_CHILD"; + const TEST_NAME: &str = "tool::web_fetch::tests::system_proxy_environment_is_ignored"; + + if std::env::var_os(CHILD_ENV).is_some() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime builds"); + let output = runtime.block_on(fetch("http://kuncode-proxy-test.invalid/")); + assert!(!output.ok, "the reserved hostname must not reach a proxy"); + return; + } + + use std::io::{ErrorKind, Read as _, Write as _}; + use std::process::Command; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::thread; + use std::time::Duration; + + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("proxy port binds"); + listener + .set_nonblocking(true) + .expect("proxy listener becomes non-blocking"); + let proxy_url = format!( + "http://{}", + listener.local_addr().expect("proxy has an address") + ); + let contacted = Arc::new(AtomicBool::new(false)); + let stop = Arc::new(AtomicBool::new(false)); + let server = { + let contacted = contacted.clone(); + let stop = stop.clone(); + thread::spawn(move || { + while !stop.load(Ordering::Acquire) { + match listener.accept() { + Ok((mut stream, _)) => { + contacted.store(true, Ordering::Release); + stream + .set_read_timeout(Some(Duration::from_secs(1))) + .expect("proxy stream timeout is configured"); + let mut request = [0; 1024]; + let _ = stream.read(&mut request); + let _ = stream.write_all(&response("200 OK", "text/plain", "proxied")); + return; + } + Err(error) if error.kind() == ErrorKind::WouldBlock => { + thread::sleep(Duration::from_millis(5)); + } + Err(error) => panic!("proxy listener failed: {error}"), + } + } + }) + }; + + // Environment variables are process-global, so a child process proves the + // client contract without racing the rest of this test binary. + let result = Command::new(std::env::current_exe().expect("test executable exists")) + .args(["--exact", TEST_NAME, "--nocapture"]) + .env(CHILD_ENV, "1") + .env("ALL_PROXY", &proxy_url) + .env("all_proxy", &proxy_url) + .env("HTTP_PROXY", &proxy_url) + .env("http_proxy", &proxy_url) + .env("HTTPS_PROXY", &proxy_url) + .env("https_proxy", &proxy_url) + .env_remove("NO_PROXY") + .env_remove("no_proxy") + .output() + .expect("isolated test process starts"); + stop.store(true, Ordering::Release); + server.join().expect("proxy listener stops cleanly"); + + assert!( + result.status.success(), + "isolated test failed:\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&result.stdout), + String::from_utf8_lossy(&result.stderr) + ); + assert!( + !contacted.load(Ordering::Acquire), + "web_fetch must not contact a process-configured proxy" + ); +} + #[tokio::test] async fn plain_text_comes_back_verbatim() { let server = diff --git a/crates/kuncode-cli/Cargo.toml b/crates/kuncode-cli/Cargo.toml index 26f5c49..16bb4dc 100644 --- a/crates/kuncode-cli/Cargo.toml +++ b/crates/kuncode-cli/Cargo.toml @@ -19,7 +19,6 @@ serde_json = { workspace = true } tracing = { workspace = true } tracing-appender = { workspace = true } tracing-subscriber = { workspace = true } -dotenvy = { workspace = true } ratatui = { workspace = true } crossterm = { workspace = true } futures-util = { workspace = true } diff --git a/crates/kuncode-cli/src/main.rs b/crates/kuncode-cli/src/main.rs index 9eb65d1..8a4d429 100644 --- a/crates/kuncode-cli/src/main.rs +++ b/crates/kuncode-cli/src/main.rs @@ -49,7 +49,6 @@ pub(crate) struct Cli { #[tokio::main] async fn main() -> Result<(), Box> { - dotenvy::dotenv().ok(); // Keep the non-blocking writer alive until every async task has stopped so // shutdown flushes the final turn/error records before the process exits. let _logging_guard = logging::init(std::env::current_dir().ok().as_deref());