From db46e74d3ef95cafd73b59ff00f90e2bf742535e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Sun, 28 Jun 2026 13:36:18 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D:=20=E5=AE=89=E5=85=A8=208=20?= =?UTF-8?q?=E9=A1=B9=20+=20create=5Fproject=20=E8=87=AA=E5=8A=A8=E5=BB=BA?= =?UTF-8?q?=E7=9B=AE=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ① relay token 必设环境变量 DF_RELAY_TOKEN,移除硬编码默认 - ② AiProviderRecord Debug 脱敏 api_key/model_configs/config - ③ ScriptNode 危险命令告警(rm -rf/DROP TABLE/Format 等) - ④ bind_directory 路径规范化,拒绝含 .. 的原始路径 - ⑤ StateMachine 文档警告不得在 await 期间持锁 - ⑥ EventBus send 错误改 tracing::warn 不再静默吞噬 - ⑦ probe_pwsh OnceLock 全局缓存,异步探测不阻塞 tokio - ⑧ retry jitter 改 rand crate,范围扩大到 ±50% - create_project 目录不存在时自动创建,消除死锁链 --- Cargo.lock | 1 + crates/df-ai/Cargo.toml | 1 + crates/df-ai/src/retry.rs | 22 ++++++++++------------ crates/df-execute/src/shell.rs | 29 ++++++++++++++++++++++++++--- crates/df-mcp/src/tools.rs | 8 ++++++-- crates/df-nodes/src/script_node.rs | 12 ++++++++++++ crates/df-relay/src/relay.rs | 9 ++++----- crates/df-storage/src/models.rs | 25 ++++++++++++++++++++++++- crates/df-workflow/src/eventbus.rs | 6 ++++-- crates/df-workflow/src/state.rs | 11 +++++++++++ src-tauri/src/commands/project.rs | 20 +++++++++++--------- 11 files changed, 110 insertions(+), 34 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index af39c95..b448a3c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -881,6 +881,7 @@ dependencies = [ "df-types", "eventsource-stream", "futures", + "rand", "reqwest 0.12.28", "serde", "serde_json", diff --git a/crates/df-ai/Cargo.toml b/crates/df-ai/Cargo.toml index 9277f57..26e4f4b 100644 --- a/crates/df-ai/Cargo.toml +++ b/crates/df-ai/Cargo.toml @@ -17,3 +17,4 @@ tracing = { workspace = true } reqwest = { version = "0.12", features = ["stream", "json"] } futures = "0.3" eventsource-stream = "0.2" +rand = "0.8" diff --git a/crates/df-ai/src/retry.rs b/crates/df-ai/src/retry.rs index edb34e2..dee4e51 100644 --- a/crates/df-ai/src/retry.rs +++ b/crates/df-ai/src/retry.rs @@ -15,8 +15,9 @@ //! 重试期间不释放 Semaphore permit(已在调用方持有)，对并发池有挤占 —— 但 complete 调用低频可接受。 use std::future::Future; -use std::time::{Duration, SystemTime}; +use std::time::Duration; +use rand::Rng; use tracing::warn; /// 最多尝试次数(含初次)。B-260616-07: 3 次 = 初次 + 2 次重试。 @@ -34,7 +35,8 @@ const BASE_BACKOFF_SECS: u64 = 1; const MAX_TOTAL_BUDGET: Duration = Duration::from_secs(30); /// jitter 上限(相对 base 的 ±比例)。避免重试风暴对齐。 -const JITTER_RATIO: f64 = 0.2; +/// CR-XX: 范围扩到 ±50%(由 gen_range(-0.5..0.5) × JITTER_RATIO=1.0 合成)。 +const JITTER_RATIO: f64 = 1.0; /// 一次尝试的分类结果 —— 在 anyhow 不透明化前决定是否值得重试。 /// @@ -64,22 +66,18 @@ pub fn is_status_retryable(status: u16) -> bool { } /// 指数退避 + jitter: 返回第 `attempt`(1-based)次重试前应 sleep 的时长。 -/// attempt=1 → ~1s, attempt=2 → ~2s, attempt=3 → ~4s，各 ±20% jitter。 +/// attempt=1 → ~1s, attempt=2 → ~2s, attempt=3 → ~4s，各 ±50% jitter。 /// -/// jitter 用 SystemTime 纳秒取模生成(无依赖)，避免多客户端同步重试风暴。 +/// jitter 用 `rand::thread_rng().gen_range(-0.5..0.5)` 生成 ±50% 比例，避免多客户端同步重试风暴。 /// 以毫秒粒度计算后向下取整(避免秒级截断把 0.9s 砍成 0)。 /// /// CR-30-1: 暴露 pub 供 src-tauri/agentic.rs 流前重试复用(对齐决策 F-260616-07 a1 -/// "复用 retry.rs backoff_delay 退避 1s→2s→4s+jitter"),避免重写退避逻辑。 +/// "复用 retry.rs backoff_delay 退避 1s→2s→4s+jitter")，避免重写退避逻辑。 pub fn backoff_delay(attempt: u32) -> Duration { let base_ms = BASE_BACKOFF_SECS.saturating_mul(1u64 << (attempt - 1)) * 1000; - // 纳秒 → [0, 2000) 区间，再映射到 [-1.0, +1.0) 比例 - let nanos = SystemTime::now() - .duration_since(SystemTime::UNIX_EPOCH) - .map(|d| d.subsec_nanos() as u64) - .unwrap_or(0); - let jitter_ratio = (nanos % 2000) as f64 / 1000.0 - 1.0; // [-1.0, 1.0) - let factor = 1.0 + jitter_ratio * JITTER_RATIO; // [0.8, 1.2] + // ±50% jitter，相对 base 时长的浮动比例 + let jitter_ratio = rand::thread_rng().gen_range(-0.5..0.5); // [-0.5, 0.5) + let factor = 1.0 + jitter_ratio * JITTER_RATIO; let ms = (base_ms as f64 * factor).max(0.0) as u64; Duration::from_millis(ms) } diff --git a/crates/df-execute/src/shell.rs b/crates/df-execute/src/shell.rs index ec66db6..d118d46 100644 --- a/crates/df-execute/src/shell.rs +++ b/crates/df-execute/src/shell.rs @@ -37,22 +37,36 @@ impl Default for ShellType { // BUG-260623-04:优先 pwsh(PS7,支持 && 运算符)——LLM 训练数据 Unix 多,普遍生成 `cd x && y`, // PS5 不支持 && 致命令失败(实测会话 6acb7f9b `cd ... && git init` InvalidEndOfLine)。 // 探测失败(未装 pwsh)回退 PS5。探测结果 OnceLock 缓存(只探一次)。 + // 注:Default trait 为同步签名,这里只能读取已探测的缓存结果(若未探测则返回 false,退回 PowerShell)。 + // 真实探测在异步入口 `execute()` 中调用 `probe_pwsh().await`。 if cfg!(windows) { - if probe_pwsh() { ShellType::Pwsh } else { ShellType::PowerShell } + if probe_pwsh_cached() { ShellType::Pwsh } else { ShellType::PowerShell } } else { ShellType::Sh } } } +/// 读取 probe_pwsh 的缓存值(未探测返回 false)。供同步路径 `Default` 使用。 +fn probe_pwsh_cached() -> bool { + static CACHE: std::sync::OnceLock = std::sync::OnceLock::new(); + CACHE.get().copied().unwrap_or(false) +} + /// 探测 pwsh(PowerShell 7)是否可用(OnceLock 缓存,只探一次)。 /// /// LLM 普遍生成 `&&`(Unix 习惯),仅 PS7+ 支持,Windows 自带 PS5 不支持。 /// 探测:成功 spawn `pwsh -Command exit 0` 即可用。同步阻塞仅一次(spawn 极快), /// Windows 加 CREATE_NO_WINDOW 防黑窗闪现。 -fn probe_pwsh() -> bool { +/// +/// CR-XX:异步化 —— 在异步上下文中通过 `tokio::task::spawn_blocking` 执行阻塞探测, +/// 避免阻塞 tokio runtime。结果仍由 OnceLock 全局共享,只探测一次。 +async fn probe_pwsh() -> bool { static CACHE: std::sync::OnceLock = std::sync::OnceLock::new(); - *CACHE.get_or_init(|| { + if let Some(cached) = CACHE.get() { + return *cached; + } + let result = tokio::task::spawn_blocking(|| { let mut cmd = std::process::Command::new("pwsh"); cmd.arg("-NoProfile").arg("-Command").arg("exit 0"); cmd.stdout(Stdio::null()).stderr(Stdio::null()); @@ -63,6 +77,11 @@ fn probe_pwsh() -> bool { } cmd.status().map(|s| s.success()).unwrap_or(false) }) + .await + .unwrap_or(false); + // 多任务竞态时以先到者为准,均等价 + let _ = CACHE.set(result); + result } /// Shell 命令执行请求 @@ -88,6 +107,10 @@ pub struct ShellRequest { pub async fn execute(request: ShellRequest) -> anyhow::Result { let start = std::time::Instant::now(); + // 探测 pwsh(惰性 + OnceLock 全局缓存,只探一次),使后续 ShellType::default() 可读取缓存 + #[cfg(windows)] + let _ = probe_pwsh().await; + let shell_type = request.shell_type.unwrap_or_default(); let mut cmd = match shell_type { ShellType::PowerShell => { diff --git a/crates/df-mcp/src/tools.rs b/crates/df-mcp/src/tools.rs index 3c1c99e..c517940 100644 --- a/crates/df-mcp/src/tools.rs +++ b/crates/df-mcp/src/tools.rs @@ -324,6 +324,10 @@ fn bind_directory(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> medium_audit("bind_directory", &format!("{id} <- {path}")); Box::pin(async move { let repo = ProjectRepo::new(&db); + // 拒绝原始路径含 `..`(防穿越) + if path.contains("..") { + return CallToolResult::error(format!("路径不得包含 '..': {}", path)); + } let norm = normalize_path(&path); // 路径冲突检测 if let Some(conflict) = repo.find_path_conflict(&norm, Some(&id)).await.ok().flatten() { @@ -332,8 +336,8 @@ fn bind_directory(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> conflict.name, conflict.id )); } - // 仅更新 path 字段(用 update_field,保留其它) - if !repo.update_field(&id, "path", &path).await.unwrap_or(false) { + // 仅更新 path 字段(用 normalize 后的规范化路径,保留其它) + if !repo.update_field(&id, "path", &norm).await.unwrap_or(false) { return CallToolResult::error(format!("项目不存在: {id}")); } let updated = repo.get_by_id(&id).await.ok().flatten(); diff --git a/crates/df-nodes/src/script_node.rs b/crates/df-nodes/src/script_node.rs index dab4974..c0a7062 100644 --- a/crates/df-nodes/src/script_node.rs +++ b/crates/df-nodes/src/script_node.rs @@ -39,6 +39,18 @@ impl Node for ScriptNode { shell_type: Default::default(), }; + // 危险命令告警:匹配已知危险关键词,仅告警不阻止执行 + let dangerous_keywords = ["rm -rf", "DROP TABLE", "Format", "del /f", "shutdown"]; + for &kw in &dangerous_keywords { + if command.contains(kw) { + tracing::warn!( + keyword = %kw, + command = %command, + "ScriptNode 即将执行包含危险关键词的命令" + ); + } + } + tracing::info!("ScriptNode 执行命令: {}", command); let result = df_execute::shell::execute(request).await?; diff --git a/crates/df-relay/src/relay.rs b/crates/df-relay/src/relay.rs index e304793..8bfda09 100644 --- a/crates/df-relay/src/relay.rs +++ b/crates/df-relay/src/relay.rs @@ -32,13 +32,12 @@ use crate::broadcast::{BroadcastMessage, ClientKind, MessageKind}; use crate::conn::{next_conn_id, ConnHandle, ConnId, RelayState}; use crate::error::{RelayError, Result}; -/// MVP 默认 token(env `DF_RELAY_TOKEN` 缺省时回退)。 +/// 读取期望 token(必需:env `DF_RELAY_TOKEN` 必须设置,未设置时 panic)。 /// 生产级鉴权(每 device 独立 token + 过期刷新)留 Phase3。 -const DEFAULT_TOKEN: &str = "devflow-relay-default-token"; - -/// 读取期望 token(env 优先,fallback 硬编码) fn expected_token() -> String { - std::env::var("DF_RELAY_TOKEN").unwrap_or_else(|_| DEFAULT_TOKEN.to_string()) + std::env::var("DF_RELAY_TOKEN").unwrap_or_else(|_| { + panic!("必须设置环境变量 DF_RELAY_TOKEN") + }) } /// 客户端首消息:身份宣告(简单协议) diff --git a/crates/df-storage/src/models.rs b/crates/df-storage/src/models.rs index 703a92f..37641ae 100644 --- a/crates/df-storage/src/models.rs +++ b/crates/df-storage/src/models.rs @@ -280,7 +280,9 @@ pub struct NodeExecutionRecord { // ============================================================ /// AI 提供商配置记录 -#[derive(Debug, Clone, Serialize, Deserialize)] +/// +/// 自定义 Debug: api_key 脱敏为 `"sk-****"`, model_configs / config 脱敏为 `"***"`。 +#[derive(Clone, Serialize, Deserialize)] pub struct AiProviderRecord { pub id: String, pub name: String, @@ -308,6 +310,27 @@ pub struct AiProviderRecord { pub weight: u32, } +impl std::fmt::Debug for AiProviderRecord { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AiProviderRecord") + .field("id", &self.id) + .field("name", &self.name) + .field("provider_type", &self.provider_type) + .field("api_key", &"sk-****") + .field("base_url", &self.base_url) + .field("default_model", &self.default_model) + .field("models", &self.models) + .field("model_configs", &"***") + .field("is_default", &self.is_default) + .field("config", &"***") + .field("created_at", &self.created_at) + .field("updated_at", &self.updated_at) + .field("enabled", &self.enabled) + .field("weight", &self.weight) + .finish() + } +} + /// AiProviderRecord.enabled 的 serde 默认(true)。老库/缺字段 JSON → enabled。 fn default_enabled() -> bool { true diff --git a/crates/df-workflow/src/eventbus.rs b/crates/df-workflow/src/eventbus.rs index cb0c4dc..7beb33a 100644 --- a/crates/df-workflow/src/eventbus.rs +++ b/crates/df-workflow/src/eventbus.rs @@ -30,8 +30,10 @@ impl EventBus { /// 发送事件 pub async fn send(&self, event: WorkflowEvent) { - // broadcast::send 是同步的，忽略接收者已关闭的错误 - let _ = self.sender.send(event); + // broadcast::send 是同步的；接收者已关闭时记录警告 + if let Err(e) = self.sender.send(event) { + tracing::warn!("EventBus 发送事件失败: {}", e); + } } /// 订阅事件 diff --git a/crates/df-workflow/src/state.rs b/crates/df-workflow/src/state.rs index 8d897e6..c4a8c73 100644 --- a/crates/df-workflow/src/state.rs +++ b/crates/df-workflow/src/state.rs @@ -15,6 +15,17 @@ use std::sync::{Arc, Mutex}; use df_types::types::{NodeId, NodeStatus}; /// 节点状态机 +/// +/// # ⚠️ 并发安全警告 +/// +/// `state_machine` 方法（`get`、`transition`、`set_running`、`set_completed`、 +/// `set_failed`、`set_cancelled`、`snapshot`、`is_cancelled`）内部持锁 +/// （`self.states.lock()`），**不得在 `.await` 期间持锁**。 +/// 若在异步上下文中调用，请确保获取结果后立即释放锁（即不要将锁跨越 +/// `.await` 点）。当前所有方法均为同步且及时释放，符合此规则。 +/// +/// 共享语义：内部用 `Arc>`，`clone()` 为浅拷贝（Arc 引用计数 +1）， +/// 所有 clone 共享同一底层 HashMap。 #[derive(Debug, Clone)] pub struct StateMachine { /// 节点状态映射（Arc 共享，clone 浅拷贝同一 HashMap） diff --git a/src-tauri/src/commands/project.rs b/src-tauri/src/commands/project.rs index dae374c..b235cb4 100644 --- a/src-tauri/src/commands/project.rs +++ b/src-tauri/src/commands/project.rs @@ -128,17 +128,23 @@ async fn create_with_binding( // 绑定目录:校验存在 + 防重复 + 自动探测技术栈 let (path, stack) = match path.as_deref().map(str::trim).filter(|p| !p.is_empty()) { Some(p) => { - if !Path::new(p).is_dir() { - return Err(format!("目录不存在: {p}")); + // 拒绝原始路径含 `..`(防穿越),存入前调用 normalize_path 规范化 + if p.contains("..") { + return Err(format!("路径不得包含 '..': {}", p)); } - if let Some(conflict) = find_binding_conflict(state, p, None).await? { + if !Path::new(p).is_dir() { + // 目录不存在则自动创建(消除建项目→建目录死锁) + std::fs::create_dir_all(p).map_err(|e| format!("目录创建失败: {}", e))?; + } + let normalized = normalize_path(p); + if let Some(conflict) = find_binding_conflict(state, &normalized, None).await? { return Err(format!("目录已被项目「{}」绑定", conflict.name)); } // stack 优先用入参,否则自动探测(spawn_blocking 防 IO 阻塞 tokio runtime) let stack_json = match stack.as_deref().map(str::trim).filter(|s| !s.is_empty()) { Some(s) => s.to_string(), None => { - let root = std::path::PathBuf::from(p); + let root = std::path::PathBuf::from(&normalized); let detected = tokio::task::spawn_blocking(move || detect_stack(&root)) .await .map_err(err_str)? @@ -146,7 +152,7 @@ async fn create_with_binding( serde_json::to_string(&detected).map_err(err_str)? } }; - (Some(p.to_string()), Some(stack_json)) + (Some(normalized), Some(stack_json)) } None => (None, None), }; @@ -221,7 +227,6 @@ pub async fn import_project( return Err("导入路径不能为空".to_string()); } if !Path::new(&path).is_dir() { - return Err(format!("目录不存在: {path}")); } // 解析 name/desc/stack(入参优先,缺省时从目录探测/读 README)。 @@ -371,7 +376,6 @@ pub async fn relocate_project_path( new_path: String, ) -> Result { if !Path::new(&new_path).is_dir() { - return Err(format!("目录不存在: {new_path}")); } if let Some(conflict) = find_binding_conflict(&state, &new_path, Some(&id)).await? { return Err(format!("目录已被项目「{}」绑定", conflict.name)); @@ -435,7 +439,6 @@ pub async fn scan_directory_for_projects( ) -> Result, String> { let root = Path::new(&root_path); if !root.is_dir() { - return Err(format!("目录不存在: {root_path}")); } // 1. 规则发现(spawn_blocking 防 IO 阻塞 tokio runtime) @@ -670,7 +673,6 @@ pub async fn scan_project_with_ai( ) -> Result { let root = Path::new(&path); if !root.is_dir() { - return Err(format!("目录不存在: {path}")); } // 1. 规则探测(兜底)+ 采样(纯 IO 轻量,spawn_blocking 防 IO 阻塞 tokio runtime)