From 537872e311cd46947b792dabdda992a323da9f48 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CJackyZhang8=E2=80=9D?= Date: Tue, 21 Jul 2026 18:48:07 +0800 Subject: [PATCH] =?UTF-8?q?=20=20=E4=B8=BB=E8=A6=81=E6=94=B9=E5=8A=A8?= =?UTF-8?q?=EF=BC=9A=20=20=20-=20web=5Ffetch=20=E5=A4=8D=E7=94=A8=E7=BB=9F?= =?UTF-8?q?=E4=B8=80=20SSRF=20=E9=98=B2=E6=8A=A4=EF=BC=8C=E9=98=BB?= =?UTF-8?q?=E6=AD=A2=E7=A7=81=E7=BD=91=E3=80=81=20=20=20=20=20loopback?= =?UTF-8?q?=E3=80=81=E4=BA=91=E5=85=83=E6=95=B0=E6=8D=AE=E5=9C=B0=E5=9D=80?= =?UTF-8?q?=E5=8F=8A=E5=8D=B1=E9=99=A9=E9=87=8D=E5=AE=9A=E5=90=91=EF=BC=9B?= =?UTF-8?q?=E9=BB=98=E8=AE=A4=E7=A6=81=E7=94=A8=E6=97=A0=E6=B3=95=20=20=20?= =?UTF-8?q?=20=20=E5=AE=89=E5=85=A8=E6=A3=80=E6=9F=A5=E9=87=8D=E5=AE=9A?= =?UTF-8?q?=E5=90=91=E7=9A=84=20CDP=20fallback=E3=80=82=20=20=20-=20Gatewa?= =?UTF-8?q?y=20=E4=B8=8A=E4=BC=A0=E4=BD=BF=E7=94=A8=E7=9C=9F=E5=AE=9E?= =?UTF-8?q?=E8=B7=AF=E5=BE=84=E8=A7=A3=E6=9E=90=EF=BC=8C=E9=98=BB=E6=AD=A2?= =?UTF-8?q?=E7=AC=A6=E5=8F=B7=E9=93=BE=E6=8E=A5=E9=80=83=E9=80=B8=20=20=20?= =?UTF-8?q?=20=20workspace=E3=80=82=20=20=20-=20=E4=BF=AE=E5=A4=8D=20Provi?= =?UTF-8?q?derPool=20=E8=B7=B3=E8=BF=87=E6=97=A0=E6=95=88=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=E5=90=8E=E7=9A=84=E5=81=A5=E5=BA=B7=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E7=B4=A2=E5=BC=95=E9=94=99=E4=BD=8D=E3=80=82=20=20=20-=20WeCom?= =?UTF-8?q?=20=E6=97=A5=E5=BF=97=E4=B8=8D=E5=86=8D=E8=BE=93=E5=87=BA=20tok?= =?UTF-8?q?en=E3=80=81AES=20key=E3=80=81=E7=AD=BE=E5=90=8D=E5=92=8C?= =?UTF-8?q?=E5=8A=A0=E5=AF=86=E8=B4=9F=E8=BD=BD=E3=80=82=20=20=20-=20Sessi?= =?UTF-8?q?on=20=E4=BF=9D=E5=AD=98=E6=94=B9=E4=B8=BA=E4=B8=B4=E6=97=B6?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E3=80=81fsync=E3=80=81=E5=8E=9F=E5=AD=90?= =?UTF-8?q?=E6=9B=BF=E6=8D=A2=EF=BC=8C=E5=B9=B6=E4=BF=9D=E7=95=99=20=20=20?= =?UTF-8?q?=20=20=E6=96=87=E4=BB=B6=E6=9D=83=E9=99=90=E3=80=82=20=20=20-?= =?UTF-8?q?=20Ghost=20=E9=85=8D=E7=BD=AE=E5=88=A0=E9=99=A4=E4=BC=9A?= =?UTF-8?q?=E8=A7=A6=E5=8F=91=E9=87=8D=E6=96=B0=E5=8A=A0=E8=BD=BD=EF=BC=8C?= =?UTF-8?q?=E5=8A=A0=E8=BD=BD=E5=A4=B1=E8=B4=A5=E5=90=8E=E4=B8=8B=E4=B8=AA?= =?UTF-8?q?=E5=91=A8=E6=9C=9F=20=20=20=20=20=E7=BB=A7=E7=BB=AD=E9=87=8D?= =?UTF-8?q?=E8=AF=95=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 1 + bin/blockcell/src/commands/gateway/files.rs | 52 ++++++- crates/channels/src/wecom/webhook.rs | 98 ++++++++++--- crates/providers/src/pool.rs | 36 ++++- crates/scheduler/src/ghost.rs | 148 ++++++++++++------- crates/storage/Cargo.toml | 3 + crates/storage/src/session.rs | 128 ++++++++++++++++- crates/tools/src/html_to_md.rs | 3 +- crates/tools/src/http_request.rs | 121 +--------------- crates/tools/src/lib.rs | 1 + crates/tools/src/ssrf.rs | 152 ++++++++++++++++++++ crates/tools/src/web.rs | 29 +++- 12 files changed, 573 insertions(+), 199 deletions(-) create mode 100644 crates/tools/src/ssrf.rs diff --git a/Cargo.lock b/Cargo.lock index 574dd939..b57c0c97 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -654,6 +654,7 @@ dependencies = [ "tokio", "tracing", "uuid", + "windows-sys 0.61.2", ] [[package]] diff --git a/bin/blockcell/src/commands/gateway/files.rs b/bin/blockcell/src/commands/gateway/files.rs index 57774745..eb369448 100644 --- a/bin/blockcell/src/commands/gateway/files.rs +++ b/bin/blockcell/src/commands/gateway/files.rs @@ -28,6 +28,19 @@ fn base64_upload_within_decoded_limit(content: &str, limit: usize) -> bool { decoded_upper_bound.saturating_sub(padding) <= limit } +fn resolve_workspace_upload_target( + workspace: &std::path::Path, + relative: &std::path::Path, +) -> Result { + let resolved_workspace = blockcell_core::path_policy::resolve_for_policy(workspace); + let resolved_target = + blockcell_core::path_policy::resolve_for_policy(&workspace.join(relative)); + if !resolved_target.starts_with(&resolved_workspace) { + return Err("Access denied: upload path escapes workspace".to_string()); + } + Ok(resolved_target) +} + async fn reject_if_file_too_large(path: &std::path::Path, limit: u64) -> Option { match tokio::fs::metadata(path).await { Ok(meta) if !file_size_within_limit(meta.len(), limit) => Some( @@ -551,7 +564,16 @@ pub(super) async fn handle_files_upload( } let workspace = state.paths.for_agent(&agent_id).workspace(); - let target = workspace.join(&rel); + let target = match resolve_workspace_upload_target(&workspace, &rel) { + Ok(target) => target, + Err(error) => { + return ( + StatusCode::FORBIDDEN, + Json(serde_json::json!({ "error": error })), + ) + .into_response() + } + }; let path_echo = rel.to_string_lossy().to_string(); let content = content.to_string(); let encoding = encoding.to_string(); @@ -597,6 +619,34 @@ pub(super) async fn handle_files_upload( mod tests { use super::*; + #[cfg(unix)] + #[test] + fn gateway_upload_rejects_symlink_escape() { + use std::os::unix::fs::symlink; + + let root = std::env::temp_dir().join(format!( + "blockcell-gateway-upload-test-{}", + uuid::Uuid::new_v4().simple() + )); + let workspace = root.join("workspace"); + let outside = root.join("outside"); + let real = workspace.join("real"); + std::fs::create_dir_all(&outside).expect("create outside dir"); + std::fs::create_dir_all(&real).expect("create real dir"); + symlink(&outside, workspace.join("link")).expect("create symlink"); + + let escaped = + resolve_workspace_upload_target(&workspace, std::path::Path::new("link/new.txt")); + assert!(escaped.is_err(), "symlink escape must be rejected"); + + let allowed = + resolve_workspace_upload_target(&workspace, std::path::Path::new("real/new.txt")) + .expect("real workspace target should be allowed"); + assert!(allowed.starts_with(std::fs::canonicalize(&workspace).unwrap())); + + let _ = std::fs::remove_dir_all(&root); + } + #[test] fn file_size_limit_rejects_values_above_limit() { assert!(file_size_within_limit( diff --git a/crates/channels/src/wecom/webhook.rs b/crates/channels/src/wecom/webhook.rs index 0b2b54f4..600edd5b 100644 --- a/crates/channels/src/wecom/webhook.rs +++ b/crates/channels/src/wecom/webhook.rs @@ -1,5 +1,35 @@ use super::*; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct WecomVerificationDiagnostics { + timestamp_present: bool, + nonce_len: usize, + signature_present: bool, + encrypted_payload_len: usize, + token_configured: bool, + aes_key_configured: bool, +} + +impl WecomVerificationDiagnostics { + fn new( + callback_token: &str, + encoding_aes_key: &str, + timestamp: &str, + nonce: &str, + signature: &str, + encrypted_payload: &str, + ) -> Self { + Self { + timestamp_present: !timestamp.is_empty(), + nonce_len: nonce.len(), + signature_present: !signature.is_empty(), + encrypted_payload_len: encrypted_payload.len(), + token_configured: !callback_token.is_empty(), + aes_key_configured: !encoding_aes_key.is_empty(), + } + } +} + pub(crate) fn build_mixed_summary(mixed: &LongConnMixed) -> String { let parts: Vec = mixed .items @@ -338,21 +368,25 @@ pub async fn process_webhook( let echostr_enc_owned = percent_decode(echostr_raw); let echostr_enc = echostr_enc_owned.as_str(); - // ── 原始数据诊断日志（INFO级别，方便复制调试）────────────────────── + let diagnostics = WecomVerificationDiagnostics::new( + &wecom_cfg.callback_token, + &wecom_cfg.encoding_aes_key, + timestamp, + nonce, + msg_signature, + echostr_enc, + ); tracing::info!( - token = %wecom_cfg.callback_token, - timestamp = %timestamp, - nonce = %nonce, - msg_signature= %msg_signature, - echostr = %echostr_enc, - echostr_len = echostr_enc.len(), - encoding_aes_key = %wecom_cfg.encoding_aes_key, - encoding_aes_key_len = wecom_cfg.encoding_aes_key.len(), - "WeCom GET 原始参数" + timestamp_present = diagnostics.timestamp_present, + nonce_len = diagnostics.nonce_len, + signature_present = diagnostics.signature_present, + encrypted_payload_len = diagnostics.encrypted_payload_len, + token_configured = diagnostics.token_configured, + aes_key_configured = diagnostics.aes_key_configured, + "WeCom GET verification parameters received" ); if !wecom_cfg.callback_token.is_empty() { - // 计算签名并打印，方便对比 let mut parts = [ wecom_cfg.callback_token.as_str(), timestamp, @@ -362,20 +396,10 @@ pub async fn process_webhook( parts.sort_unstable(); let combined = parts.join(""); let computed = sha1_hex(combined.as_bytes()); - tracing::info!( - computed_signature = %computed, - expected_signature = %msg_signature, - sort_input = %combined, - "WeCom GET 签名计算" - ); // 4-param signature: sort(token, timestamp, nonce, msg_encrypt) if computed != msg_signature { - tracing::warn!( - computed = %computed, - expected = %msg_signature, - "WeCom webhook: GET 签名不匹配" - ); + tracing::warn!("WeCom webhook: GET signature verification failed"); return (403, "Forbidden: invalid signature".to_string()); } } @@ -659,3 +683,33 @@ pub async fn process_webhook( (200, "success".to_string()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn wecom_verification_diagnostics_redact_secrets() { + let diagnostics = WecomVerificationDiagnostics::new( + "callback-secret", + "aes-secret", + "1700000000", + "nonce-secret", + "signature-secret", + "encrypted-secret", + ); + let rendered = format!("{diagnostics:?}"); + + for secret in [ + "callback-secret", + "aes-secret", + "nonce-secret", + "signature-secret", + "encrypted-secret", + ] { + assert!(!rendered.contains(secret), "diagnostics leaked {secret}"); + } + assert!(rendered.contains("token_configured: true")); + assert!(rendered.contains("aes_key_configured: true")); + } +} diff --git a/crates/providers/src/pool.rs b/crates/providers/src/pool.rs index 3611496c..38c36022 100644 --- a/crates/providers/src/pool.rs +++ b/crates/providers/src/pool.rs @@ -209,6 +209,7 @@ impl ProviderPool { idx, model = %model, provider = %provider_name, weight, priority, "ProviderPool: entry loaded" ); + let built_idx = entries.len(); entries.push(BuiltEntry { model, provider_name, @@ -216,7 +217,7 @@ impl ProviderPool { priority, provider: Arc::from(p), }); - health_map.insert(idx, EntryHealth::Healthy); + health_map.insert(built_idx, EntryHealth::Healthy); } Err(e) => { warn!(idx, model = %model, error = %e, "ProviderPool: failed to build entry, skipping"); @@ -722,6 +723,39 @@ mod tests { assert_eq!(status[0].weight, 2); } + #[test] + fn pool_skips_invalid_entry_without_health_index_gap() { + let mut config = Config::default(); + config.agents.defaults.model_pool = vec![ + blockcell_core::config::ModelEntry { + model: "missing/model".to_string(), + provider: "provider-that-does-not-exist".to_string(), + weight: 1, + priority: 1, + input_price: None, + output_price: None, + temperature: None, + tool_call_mode: blockcell_core::config::ToolCallMode::Native, + }, + blockcell_core::config::ModelEntry { + model: "ollama/qwen3.6".to_string(), + provider: "ollama".to_string(), + weight: 1, + priority: 2, + input_price: None, + output_price: None, + temperature: None, + tool_call_mode: blockcell_core::config::ToolCallMode::Native, + }, + ]; + + let pool = ProviderPool::from_config(&config).expect("valid second entry should build"); + let status = pool.status_summary(); + assert_eq!(status.len(), 1); + assert_eq!(status[0].health, "healthy"); + assert_eq!(pool.acquire().map(|(idx, _)| idx), Some(0)); + } + #[test] fn test_report_transient_fails_then_cooling() { let mut config = Config::default(); diff --git a/crates/scheduler/src/ghost.rs b/crates/scheduler/src/ghost.rs index 3e733e86..3dc8e451 100644 --- a/crates/scheduler/src/ghost.rs +++ b/crates/scheduler/src/ghost.rs @@ -73,6 +73,23 @@ impl SyncTracker { } } +#[derive(Debug, Default)] +struct GhostConfigReloadTracker { + last_mtime: Option, + loaded_once: bool, +} + +impl GhostConfigReloadTracker { + fn should_reload(&self, current_mtime: Option) -> bool { + !self.loaded_once || current_mtime != self.last_mtime + } + + fn record_success(&mut self, current_mtime: Option) { + self.loaded_once = true; + self.last_mtime = current_mtime; + } +} + pub struct GhostMaintenanceService { config: GhostMaintenanceServiceConfig, #[allow(dead_code)] @@ -246,8 +263,7 @@ impl GhostMaintenanceService { let config_mtime = |path: &std::path::Path| -> Option { std::fs::metadata(path).ok().and_then(|m| m.modified().ok()) }; - let mut last_config_mtime: Option = None; - let mut loaded_once = false; + let mut reload_tracker = GhostConfigReloadTracker::default(); let mut consecutive_failures = 0u32; const MAX_CONSECUTIVE_FAILURES: u32 = 3; @@ -257,58 +273,58 @@ impl GhostMaintenanceService { _ = check_interval.tick() => { // Hot-reload config（按 mtime 短路，避免每分钟重复读+解析） let current_mtime = config_mtime(&config_file); - let config_changed = match (current_mtime, last_config_mtime) { - (Some(current), Some(last)) => current != last, - (Some(_), None) => true, - (None, _) => false, - }; - if !loaded_once || config_changed { - if let Ok(new_config) = Config::load_or_default(&config_paths) { - let new_ghost = GhostMaintenanceServiceConfig::from_config(&new_config); - - // Check if relevant fields changed - let schedule_changed = new_ghost.schedule != self.config.schedule; - let changed = new_ghost.enabled != self.config.enabled || - schedule_changed || - new_ghost.model != self.config.model || - new_ghost.max_syncs_per_day != self.config.max_syncs_per_day || - new_ghost.auto_social != self.config.auto_social; - - if changed { - info!("👻 Ghost config updated via hot-reload"); - self.config = new_ghost; - - // Re-parse schedule if changed - if schedule_changed { - schedule = match Self::parse_cron_schedule(&self.config.schedule) { - Ok(s) => s, - Err(e) => { - let normalized = Self::normalize_cron_schedule(&self.config.schedule); - error!( - error = %e, - schedule = %self.config.schedule, - normalized_schedule = %normalized, - "Ghost: invalid cron schedule, falling back to every 4 hours" - ); - "0 0 */4 * * *".parse::().unwrap() + if reload_tracker.should_reload(current_mtime) { + match Config::load_or_default(&config_paths) { + Ok(new_config) => { + let new_ghost = GhostMaintenanceServiceConfig::from_config(&new_config); + + // Check if relevant fields changed + let schedule_changed = new_ghost.schedule != self.config.schedule; + let changed = new_ghost.enabled != self.config.enabled || + schedule_changed || + new_ghost.model != self.config.model || + new_ghost.max_syncs_per_day != self.config.max_syncs_per_day || + new_ghost.auto_social != self.config.auto_social; + + if changed { + info!("👻 Ghost config updated via hot-reload"); + self.config = new_ghost; + + // Re-parse schedule if changed + if schedule_changed { + schedule = match Self::parse_cron_schedule(&self.config.schedule) { + Ok(s) => s, + Err(e) => { + let normalized = Self::normalize_cron_schedule(&self.config.schedule); + error!( + error = %e, + schedule = %self.config.schedule, + normalized_schedule = %normalized, + "Ghost: invalid cron schedule, falling back to every 4 hours" + ); + "0 0 */4 * * *".parse::().unwrap() + } + }; + // 修复：schedule 变更后重置 next_scheduled， + // 避免旧的 last_run 去重逻辑阻止新 schedule 的首次执行。 + next_scheduled = schedule.upcoming(Utc).next(); } - }; - // 修复：schedule 变更后重置 next_scheduled， - // 避免旧的 last_run 去重逻辑阻止新 schedule 的首次执行。 - next_scheduled = schedule.upcoming(Utc).next(); - } - if !self.config.enabled { - info!("👻 GhostMaintenanceService disabled via config"); - } else { - info!("👻 GhostMaintenanceService enabled/updated via config: {}", self.config.schedule); + if !self.config.enabled { + info!("👻 GhostMaintenanceService disabled via config"); + } else { + info!("👻 GhostMaintenanceService enabled/updated via config: {}", self.config.schedule); + } + } + + // load_or_default may write defaults and change the mtime. + reload_tracker.record_success(config_mtime(&config_file)); + } + Err(error) => { + warn!(error = %error, "Ghost config reload failed; will retry on next tick"); } } } - loaded_once = true; - // 重新读取 mtime：load_or_default 可能写入了默认字段。 - last_config_mtime = config_mtime(&config_file); - } if !self.config.enabled { continue; @@ -350,6 +366,40 @@ impl GhostMaintenanceService { mod tests { use super::*; + #[test] + fn ghost_config_reload_detects_deleted_file() { + let previous = std::time::UNIX_EPOCH + std::time::Duration::from_secs(10); + let tracker = GhostConfigReloadTracker { + last_mtime: Some(previous), + loaded_once: true, + }; + + assert!(tracker.should_reload(None)); + } + + #[test] + fn ghost_config_reload_failure_remains_retryable() { + let current = std::time::UNIX_EPOCH + std::time::Duration::from_secs(20); + let tracker = GhostConfigReloadTracker::default(); + + assert!(tracker.should_reload(Some(current))); + assert!(tracker.should_reload(Some(current))); + assert!(!tracker.loaded_once); + assert_eq!(tracker.last_mtime, None); + } + + #[test] + fn ghost_config_reload_success_records_latest_mtime() { + let current = std::time::UNIX_EPOCH + std::time::Duration::from_secs(30); + let mut tracker = GhostConfigReloadTracker::default(); + + tracker.record_success(Some(current)); + + assert!(tracker.loaded_once); + assert_eq!(tracker.last_mtime, Some(current)); + assert!(!tracker.should_reload(Some(current))); + } + #[test] fn test_sync_tracker() { let mut tracker = SyncTracker::new(); diff --git a/crates/storage/Cargo.toml b/crates/storage/Cargo.toml index b263629b..72c3ec22 100644 --- a/crates/storage/Cargo.toml +++ b/crates/storage/Cargo.toml @@ -20,3 +20,6 @@ rabitq_rs = { workspace = true } [dev-dependencies] tempfile = "3.8" + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = ["Win32_Storage_FileSystem"] } diff --git a/crates/storage/src/session.rs b/crates/storage/src/session.rs index 594a9f32..e2425d39 100644 --- a/crates/storage/src/session.rs +++ b/crates/storage/src/session.rs @@ -3,9 +3,91 @@ use blockcell_core::{session_file_stem, session_id_from_file_stem, Paths, Result use serde::{Deserialize, Serialize}; use serde_json::Value; use std::fs::{File, OpenOptions}; -use std::io::{BufRead, BufReader, Write}; +use std::io::{BufRead, BufReader, BufWriter, Write}; +use std::sync::atomic::{AtomicU64, Ordering}; use tracing::debug; +static SESSION_WRITE_COUNTER: AtomicU64 = AtomicU64::new(0); + +fn session_temp_path(path: &std::path::Path) -> std::path::PathBuf { + let file_name = path + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or("session.jsonl"); + let counter = SESSION_WRITE_COUNTER.fetch_add(1, Ordering::Relaxed); + path.with_file_name(format!( + ".{file_name}.tmp.{}.{}", + std::process::id(), + counter + )) +} + +#[cfg(windows)] +fn replace_session_file( + temp_path: &std::path::Path, + path: &std::path::Path, +) -> std::io::Result<()> { + use std::os::windows::ffi::OsStrExt; + use windows_sys::Win32::Storage::FileSystem::{ + MoveFileExW, MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, + }; + + let temp_wide: Vec = temp_path.as_os_str().encode_wide().chain(Some(0)).collect(); + let path_wide: Vec = path.as_os_str().encode_wide().chain(Some(0)).collect(); + let replaced = unsafe { + MoveFileExW( + temp_wide.as_ptr(), + path_wide.as_ptr(), + MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH, + ) + }; + if replaced == 0 { + Err(std::io::Error::last_os_error()) + } else { + Ok(()) + } +} + +#[cfg(not(windows))] +fn replace_session_file( + temp_path: &std::path::Path, + path: &std::path::Path, +) -> std::io::Result<()> { + std::fs::rename(temp_path, path) +} + +fn write_session_bytes_atomically(path: &std::path::Path, bytes: &[u8]) -> std::io::Result<()> { + let temp_path = session_temp_path(path); + let result = (|| { + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + let file = options.open(&temp_path)?; + if let Ok(metadata) = std::fs::metadata(path) { + std::fs::set_permissions(&temp_path, metadata.permissions())?; + } + + let mut writer = BufWriter::new(file); + writer.write_all(bytes)?; + writer.flush()?; + writer.get_ref().sync_all()?; + drop(writer); + + replace_session_file(&temp_path, path)?; + + #[cfg(unix)] + if let Some(parent) = path.parent() { + std::fs::File::open(parent)?.sync_all()?; + } + + Ok(()) + })(); + + if result.is_err() { + let _ = std::fs::remove_file(&temp_path); + } + result +} + #[derive(Debug, Serialize, Deserialize)] #[serde(tag = "_type")] enum SessionLine { @@ -216,19 +298,19 @@ impl SessionStore { messages: &[ChatMessage], metadata: &Value, ) -> Result<()> { - let mut file = File::create(path)?; - let metadata_line = SessionLine::Metadata { created_at: created_at.to_string(), updated_at: updated_at.to_string(), metadata: metadata.clone(), }; - writeln!(file, "{}", serde_json::to_string(&metadata_line)?)?; + let mut content = Vec::new(); + writeln!(content, "{}", serde_json::to_string(&metadata_line)?)?; for msg in messages { - writeln!(file, "{}", serde_json::to_string(msg)?)?; + writeln!(content, "{}", serde_json::to_string(msg)?)?; } + write_session_bytes_atomically(path, &content)?; Ok(()) } @@ -406,4 +488,40 @@ mod tests { .expect("load metadata after save"); assert_eq!(loaded["skill_state"]["last_skill"], "deep_analysis"); } + + #[test] + fn session_atomic_replacement_writes_complete_file_without_temp_residue() { + let dir = TempDir::new().expect("temp dir"); + let path = dir.path().join("session.jsonl"); + std::fs::write(&path, "old").expect("write old session"); + + write_session_bytes_atomically(&path, b"metadata\nmessage\n") + .expect("atomic replacement should succeed"); + + assert_eq!(std::fs::read(&path).unwrap(), b"metadata\nmessage\n"); + let entries: Vec = std::fs::read_dir(dir.path()) + .unwrap() + .map(|entry| entry.unwrap().file_name().to_string_lossy().to_string()) + .collect(); + assert_eq!(entries, vec!["session.jsonl"]); + } + + #[cfg(unix)] + #[test] + fn session_atomic_replacement_preserves_existing_permissions() { + use std::os::unix::fs::PermissionsExt; + + let dir = TempDir::new().expect("temp dir"); + let path = dir.path().join("session.jsonl"); + std::fs::write(&path, "old").expect("write old session"); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)) + .expect("set restrictive permissions"); + + write_session_bytes_atomically(&path, b"new").expect("atomic replacement should succeed"); + + assert_eq!( + std::fs::metadata(&path).unwrap().permissions().mode() & 0o777, + 0o600 + ); + } } diff --git a/crates/tools/src/html_to_md.rs b/crates/tools/src/html_to_md.rs index bd8cedf0..6dc2be7b 100644 --- a/crates/tools/src/html_to_md.rs +++ b/crates/tools/src/html_to_md.rs @@ -55,8 +55,9 @@ impl MarkdownMeta { /// 3. If response is `text/html` → convert locally via `htmd` /// 4. Otherwise → return raw text pub async fn fetch_as_markdown(url: &str, max_chars: usize) -> Result<(String, MarkdownMeta)> { + crate::ssrf::ensure_url_allowed(url).await?; let client = Client::builder() - .redirect(reqwest::redirect::Policy::limited(10)) + .redirect(crate::ssrf::redirect_policy(true)) .timeout(std::time::Duration::from_secs(30)) .build() .map_err(|e| Error::Tool(format!("Failed to create HTTP client: {}", e)))?; diff --git a/crates/tools/src/http_request.rs b/crates/tools/src/http_request.rs index 3f0d0b2c..2db8fbe2 100644 --- a/crates/tools/src/http_request.rs +++ b/crates/tools/src/http_request.rs @@ -2,106 +2,12 @@ use async_trait::async_trait; use blockcell_core::{Error, Result}; use reqwest::Client; use serde_json::{json, Value}; -use std::net::{IpAddr, ToSocketAddrs}; +use crate::ssrf::{ensure_url_allowed, redirect_policy}; +#[cfg(test)] +use crate::ssrf::{host_is_blocked, is_blocked_ip}; use crate::{Tool, ToolContext, ToolSchema}; -/// Set this env var to `1`/`true` to allow requests to private/internal -/// addresses (disables the SSRF guard). Off by default. -const SSRF_ALLOW_ENV: &str = "BLOCKCELL_HTTP_ALLOW_PRIVATE"; - -fn private_network_allowed() -> bool { - std::env::var(SSRF_ALLOW_ENV) - .map(|v| v == "1" || v.eq_ignore_ascii_case("true")) - .unwrap_or(false) -} - -/// Whether an IP must be refused to mitigate SSRF (loopback, private, -/// link-local incl. cloud metadata 169.254.169.254, CGNAT, ULA, ...). -fn is_blocked_ip(ip: &IpAddr) -> bool { - match ip { - IpAddr::V4(v4) => { - let o = v4.octets(); - v4.is_loopback() - || v4.is_private() - || v4.is_link_local() - || v4.is_unspecified() - || v4.is_broadcast() - || o[0] == 0 - || (o[0] == 100 && (64..128).contains(&o[1])) // CGNAT 100.64.0.0/10 - } - IpAddr::V6(v6) => { - v6.is_loopback() - || v6.is_unspecified() - || (v6.segments()[0] & 0xfe00) == 0xfc00 // unique-local fc00::/7 - || (v6.segments()[0] & 0xffc0) == 0xfe80 // link-local fe80::/10 - || v6 - .to_ipv4() - .map(|v4| is_blocked_ip(&IpAddr::V4(v4))) - .unwrap_or(false) - } - } -} - -/// Synchronous host check used by the redirect policy: resolves the host and -/// reports whether it points at a blocked address. DNS failures return -/// `false` so reqwest surfaces the underlying connection error. -fn host_is_blocked(host: &str) -> bool { - if let Ok(ip) = host.parse::() { - return is_blocked_ip(&ip); - } - match (host, 0u16).to_socket_addrs() { - Ok(addrs) => addrs.into_iter().any(|a| is_blocked_ip(&a.ip())), - Err(_) => false, - } -} - -fn ssrf_denied(host: &str) -> Error { - Error::PermissionDenied(format!( - "Refusing to request private/internal address ({host}). \ - Set {SSRF_ALLOW_ENV}=1 to override." - )) -} - -/// Pre-flight SSRF check for the initial URL (async DNS resolution). -async fn ensure_url_allowed(url: &str) -> Result<()> { - if private_network_allowed() { - return Ok(()); - } - let parsed = - reqwest::Url::parse(url).map_err(|e| Error::Validation(format!("Invalid URL: {}", e)))?; - let host = parsed - .host_str() - .ok_or_else(|| Error::Validation("URL has no host".to_string()))?; - - if let Ok(ip) = host.parse::() { - return if is_blocked_ip(&ip) { - Err(ssrf_denied(host)) - } else { - Ok(()) - }; - } - - let port = parsed.port_or_known_default().unwrap_or(80); - let addrs = tokio::net::lookup_host((host, port)) - .await - .map_err(|e| Error::Tool(format!("DNS resolution failed for {}: {}", host, e)))?; - let mut any = false; - for addr in addrs { - any = true; - if is_blocked_ip(&addr.ip()) { - return Err(ssrf_denied(host)); - } - } - if !any { - return Err(Error::Tool(format!( - "DNS resolution returned no addresses for {}", - host - ))); - } - Ok(()) -} - fn parse_string_map(input: &str) -> Option> { let trimmed = input.trim(); if trimmed.is_empty() { @@ -332,27 +238,8 @@ impl Tool for HttpRequestTool { ensure_url_allowed(url).await?; // Build client - let redirect_policy = if follow_redirects { - let allow_private = private_network_allowed(); - reqwest::redirect::Policy::custom(move |attempt| { - if attempt.previous().len() >= 10 { - return attempt.error("too many redirects"); - } - if !allow_private { - if let Some(host) = attempt.url().host_str() { - if host_is_blocked(host) { - return attempt.error("redirect to private/internal address blocked"); - } - } - } - attempt.follow() - }) - } else { - reqwest::redirect::Policy::none() - }; - let client = Client::builder() - .redirect(redirect_policy) + .redirect(redirect_policy(follow_redirects)) .timeout(std::time::Duration::from_secs(timeout_secs)) .build() .map_err(|e| Error::Tool(format!("Failed to create HTTP client: {}", e)))?; diff --git a/crates/tools/src/lib.rs b/crates/tools/src/lib.rs index dd957482..1ace77aa 100644 --- a/crates/tools/src/lib.rs +++ b/crates/tools/src/lib.rs @@ -41,6 +41,7 @@ pub mod session_search; pub mod skill_manage; pub mod skills; pub mod spawn; +pub(crate) mod ssrf; pub mod stream_subscribe; pub mod system_info; pub mod tasks; diff --git a/crates/tools/src/ssrf.rs b/crates/tools/src/ssrf.rs new file mode 100644 index 00000000..df3e1f82 --- /dev/null +++ b/crates/tools/src/ssrf.rs @@ -0,0 +1,152 @@ +use blockcell_core::{Error, Result}; +use std::net::{IpAddr, ToSocketAddrs}; + +/// Set this env var to `1`/`true` to allow requests to private/internal +/// addresses (disables the SSRF guard). Off by default. +pub(crate) const SSRF_ALLOW_ENV: &str = "BLOCKCELL_HTTP_ALLOW_PRIVATE"; + +pub(crate) fn private_network_allowed() -> bool { + std::env::var(SSRF_ALLOW_ENV) + .map(|v| v == "1" || v.eq_ignore_ascii_case("true")) + .unwrap_or(false) +} + +pub(crate) fn is_blocked_ip(ip: &IpAddr) -> bool { + match ip { + IpAddr::V4(v4) => { + let o = v4.octets(); + v4.is_loopback() + || v4.is_private() + || v4.is_link_local() + || v4.is_unspecified() + || v4.is_broadcast() + || o[0] == 0 + || (o[0] == 100 && (64..128).contains(&o[1])) + } + IpAddr::V6(v6) => { + v6.is_loopback() + || v6.is_unspecified() + || (v6.segments()[0] & 0xfe00) == 0xfc00 + || (v6.segments()[0] & 0xffc0) == 0xfe80 + || v6 + .to_ipv4() + .map(|v4| is_blocked_ip(&IpAddr::V4(v4))) + .unwrap_or(false) + } + } +} + +pub(crate) fn host_is_blocked(host: &str) -> bool { + let host = host + .strip_prefix('[') + .and_then(|value| value.strip_suffix(']')) + .unwrap_or(host); + if let Ok(ip) = host.parse::() { + return is_blocked_ip(&ip); + } + match (host, 0u16).to_socket_addrs() { + Ok(addrs) => addrs.into_iter().any(|a| is_blocked_ip(&a.ip())), + Err(_) => false, + } +} + +fn ssrf_denied(host: &str) -> Error { + Error::PermissionDenied(format!( + "Refusing to request private/internal address ({host}). Set {SSRF_ALLOW_ENV}=1 to override." + )) +} + +pub(crate) async fn ensure_url_allowed(url: &str) -> Result<()> { + if private_network_allowed() { + return Ok(()); + } + + let parsed = + reqwest::Url::parse(url).map_err(|e| Error::Validation(format!("Invalid URL: {e}")))?; + if !matches!(parsed.scheme(), "http" | "https") { + return Err(Error::Validation("URL must use http or https".to_string())); + } + let host = parsed + .host_str() + .ok_or_else(|| Error::Validation("URL has no host".to_string()))?; + let host = host + .strip_prefix('[') + .and_then(|value| value.strip_suffix(']')) + .unwrap_or(host); + + if let Ok(ip) = host.parse::() { + return if is_blocked_ip(&ip) { + Err(ssrf_denied(host)) + } else { + Ok(()) + }; + } + + let port = parsed.port_or_known_default().unwrap_or(80); + let addrs = tokio::net::lookup_host((host, port)) + .await + .map_err(|e| Error::Tool(format!("DNS resolution failed for {host}: {e}")))?; + let mut any = false; + for addr in addrs { + any = true; + if is_blocked_ip(&addr.ip()) { + return Err(ssrf_denied(host)); + } + } + if !any { + return Err(Error::Tool(format!( + "DNS resolution returned no addresses for {host}" + ))); + } + Ok(()) +} + +pub(crate) fn redirect_policy(follow: bool) -> reqwest::redirect::Policy { + if !follow { + return reqwest::redirect::Policy::none(); + } + + let allow_private = private_network_allowed(); + reqwest::redirect::Policy::custom(move |attempt| { + if attempt.previous().len() >= 10 { + return attempt.error("too many redirects"); + } + if !allow_private { + if let Some(host) = attempt.url().host_str() { + if host_is_blocked(host) { + return attempt.error("redirect to private/internal address blocked"); + } + } + } + attempt.follow() + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ssrf_ip_classification_blocks_internal_ranges() { + for ip in [ + "127.0.0.1", + "169.254.169.254", + "10.0.0.1", + "100.64.0.1", + "::1", + "fc00::1", + "fe80::1", + ] { + assert!(is_blocked_ip(&ip.parse().unwrap()), "{ip} must be blocked"); + } + assert!(!is_blocked_ip(&"8.8.8.8".parse().unwrap())); + } + + #[tokio::test] + async fn ssrf_initial_url_rejects_non_http_schemes() { + let error = ensure_url_allowed("file:///etc/passwd") + .await + .expect_err("file URLs must be rejected"); + assert!(error.to_string().contains("http or https")); + } +} diff --git a/crates/tools/src/web.rs b/crates/tools/src/web.rs index a4ba47e9..9a60bdaa 100644 --- a/crates/tools/src/web.rs +++ b/crates/tools/src/web.rs @@ -348,6 +348,7 @@ impl Tool for WebFetchTool { async fn execute(&self, ctx: ToolContext, params: Value) -> Result { let url = params["url"].as_str().unwrap(); + crate::ssrf::ensure_url_allowed(url).await?; let extract_mode = params .get("extractMode") .and_then(|v| v.as_str()) @@ -390,7 +391,7 @@ async fn fetch_markdown( let is_challenge = html_looks_like_challenge(&content) || (content.trim().len() < 500 && meta.status == 200); - if is_challenge { + if is_challenge && crate::ssrf::private_network_allowed() { if let Some(ws) = workspace { tracing::debug!(url, "web_fetch: JS challenge detected, trying CDP fallback"); match fetch_via_cdp(url, max_chars, ws).await { @@ -398,6 +399,11 @@ async fn fetch_markdown( Err(e) => tracing::warn!(error = %e, url, "CDP fetch fallback failed"), } } + } else if is_challenge { + tracing::debug!( + url, + "web_fetch: skipping CDP fallback while private-network SSRF protection is enabled" + ); } let truncated = content.len() >= max_chars; @@ -522,7 +528,7 @@ async fn fetch_via_cdp(url: &str, max_chars: usize, workspace: &std::path::Path) /// Fetch and extract plain text (strip all formatting). async fn fetch_text(url: &str, max_chars: usize) -> Result { let client = Client::builder() - .redirect(reqwest::redirect::Policy::limited(10)) + .redirect(crate::ssrf::redirect_policy(true)) .timeout(std::time::Duration::from_secs(30)) .build() .map_err(|e| Error::Tool(format!("Failed to create HTTP client: {}", e)))?; @@ -581,7 +587,7 @@ async fn fetch_text(url: &str, max_chars: usize) -> Result { /// Fetch raw response body without conversion. async fn fetch_raw(url: &str, max_chars: usize) -> Result { let client = Client::builder() - .redirect(reqwest::redirect::Policy::limited(10)) + .redirect(crate::ssrf::redirect_policy(true)) .timeout(std::time::Duration::from_secs(30)) .build() .map_err(|e| Error::Tool(format!("Failed to create HTTP client: {}", e)))?; @@ -671,6 +677,23 @@ fn extract_text_from_html(html: &str) -> String { #[cfg(test)] mod tests { use super::*; + + #[tokio::test] + async fn web_fetch_ssrf_guard_rejects_private_targets() { + for url in [ + "http://127.0.0.1:18790/v1/config", + "http://169.254.169.254/latest/meta-data/", + "http://[::1]/", + ] { + let error = crate::ssrf::ensure_url_allowed(url) + .await + .expect_err("private target must be rejected"); + assert!( + error.to_string().contains("private/internal"), + "unexpected error for {url}: {error}" + ); + } + } use serde_json::json; #[test]