Compare commits

...

1 Commits

Author SHA1 Message Date
db46e74d3e 修复: 安全 8 项 + create_project 自动建目录
- ① 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 目录不存在时自动创建,消除死锁链
2026-06-28 13:36:18 +08:00
11 changed files with 110 additions and 34 deletions

1
Cargo.lock generated
View File

@@ -881,6 +881,7 @@ dependencies = [
"df-types",
"eventsource-stream",
"futures",
"rand",
"reqwest 0.12.28",
"serde",
"serde_json",

View File

@@ -17,3 +17,4 @@ tracing = { workspace = true }
reqwest = { version = "0.12", features = ["stream", "json"] }
futures = "0.3"
eventsource-stream = "0.2"
rand = "0.8"

View File

@@ -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)
}

View File

@@ -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<bool> = 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<bool> = 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<ShellResult> {
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 => {

View File

@@ -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();

View File

@@ -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?;

View File

@@ -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")
})
}
/// 客户端首消息:身份宣告(简单协议)

View File

@@ -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

View File

@@ -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);
}
}
/// 订阅事件

View File

@@ -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<Mutex<HashMap>>``clone()` 为浅拷贝Arc 引用计数 +1
/// 所有 clone 共享同一底层 HashMap。
#[derive(Debug, Clone)]
pub struct StateMachine {
/// 节点状态映射Arc 共享clone 浅拷贝同一 HashMap

View File

@@ -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<ProjectRecord, String> {
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<Vec<ScannedProjectItem>, 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<AiScanResult, String> {
let root = Path::new(&path);
if !root.is_dir() {
return Err(format!("目录不存在: {path}"));
}
// 1. 规则探测(兜底)+ 采样(纯 IO 轻量,spawn_blocking 防 IO 阻塞 tokio runtime)