//! 项目基础设施配置 Repo:ProjectServiceRepo(project_services 表,知识图谱 Phase 3 V31) //! //! 记录项目依赖的基础设施(数据库/缓存/MQ/API 等),为 AI 执行任务时提供基础设施上下文: //! "这项目用了什么数据库、Redis 在哪、有没有 MQ"。对标设计 //! docs/02-架构设计/专项设计/项目知识图谱与任务队列系统-2026-06-26.md §2.3。 //! //! 关键设计(对标 IdeaEvalRepo 走宏 + 专用方法模式): //! - **走 `impl_repo!` 宏**:表有 `updated_at`(基础设施配置可改,非审计不可篡改), //! 故宏生成的 insert/get_by_id/list_all/query/update_field/update_full/delete 全可用, //! 白名单已在 settings.rs 登记(allowed_columns_for("project_services") 返回 Some)。 //! - `service_type` 白名单(mysql/postgresql/sqlite/redis/mongodb/mq/api/other)应用层校验, //! 非进 DB 约束(对标 task_links.link_type / project_events.event_type,保留扩展性)。 //! - **敏感凭证审查(D10)**:insert/update_full 时检测 config_json / endpoint / remark 中是否 //! 含 password/secret/token/api_key 子串,命中即拒(防密码/密钥误存进表)。审查在应用层 //! 非 DB 约束(SQLite CHECK 对 JSON 内容无法表达),best-effort 兜底,非加密级安全。 //! - 专用 `list_by_project(project_id)` / `list_by_project_env(project_id, environment)` //! 覆盖最高频查询(命中 idx_project_services_project)。 use std::sync::Arc; use rusqlite::{params, Connection, OptionalExtension, Row}; use tokio::sync::Mutex; use df_types::error::Result; use crate::db::Database; use crate::models::ProjectServiceRecord; use super::impl_repo; use super::{now_millis_str, storage_err, validate_column_name}; // ============================================================ // service_type 白名单 + 校验 // ============================================================ /// `service_type` 白名单:基础设施类型枚举(对标设计 §2.3)。 /// /// 防拼写漂移 / 非法值进库。新增类型(如 elasticsearch / kafka)在此数组追加 + 设计文档 /// §2.3 同步更新,无需 DB 迁移(应用层校验,保留扩展性)。 pub const PROJECT_SERVICE_TYPES: &[&str] = &[ "mysql", "postgresql", "sqlite", "redis", "mongodb", "mq", "api", "other", ]; /// 校验 `service_type` 在白名单内,否则返回 Err(对标 task_links.validate_link_type)。 pub fn validate_service_type(service_type: &str) -> Result<()> { if PROJECT_SERVICE_TYPES.contains(&service_type) { Ok(()) } else { Err(df_types::error::Error::Storage(format!( "非法 service_type: {service_type},合法值: {:?}", PROJECT_SERVICE_TYPES ))) } } // ============================================================ // 敏感凭证审查(D10 安全边界) // ============================================================ /// 敏感字段名子串黑名单(小写匹配)。config_json / endpoint / remark 含任一即视为误存凭证。 /// /// D10 边界:本表不存密码/密钥/Token,凭证走环境变量。此审查是 best-effort 兜底——检测常见 /// 凭证字段名子串拒绝,防用户/AI 误把 `{"password": "xxx"}` 灌进 config_json。非加密级安全, /// 非穷尽(如把密码放在非标准字段名仍可能漏检),核心防御仍依赖调用方纪律 + 环境变量约定。 const SENSITIVE_TOKENS: &[&str] = &["password", "passwd", "secret", "api_key", "apikey", "token"]; /// 审查给定文本(可空)是否含敏感字段名子串(大小写不敏感),命中即返回 Err(D10 边界)。 /// /// 对 config_json / endpoint / remark 三字段调用。endpoint 含 "token=" query 参数(如 /// `?token=abc`)或 config_json 含 `"password":` 都会被拦截。空/None 文本直接放行。 fn reject_sensitive(text: Option<&str>, field_name: &str) -> Result<()> { let Some(t) = text else { return Ok(()) }; if t.is_empty() { return Ok(()); } let lower = t.to_ascii_lowercase(); for tok in SENSITIVE_TOKENS { if lower.contains(tok) { return Err(df_types::error::Error::Storage(format!( "project_services.{field_name} 含敏感凭证字段「{tok}」,违反 D10 安全边界:\ 不存敏感凭证,密码/密钥走环境变量。请改用环境变量引用" ))); } } Ok(()) } /// 对一条 ProjectServiceRecord 做应用层前置校验(service_type 白名单 + 凭证审查)。 /// insert / update_full 共用,保证两条写入路径校验一致。 fn validate_record(rec: &ProjectServiceRecord) -> Result<()> { validate_service_type(&rec.service_type)?; reject_sensitive(rec.config_json.as_deref(), "config_json")?; reject_sensitive(rec.endpoint.as_deref(), "endpoint")?; reject_sensitive(rec.remark.as_deref(), "remark")?; Ok(()) } // ============================================================ // from_row 辅助函数 // ============================================================ fn project_service_from_row(row: &Row<'_>) -> std::result::Result { Ok(ProjectServiceRecord { id: row.get("id")?, project_id: row.get("project_id")?, name: row.get("name")?, service_type: row.get("service_type")?, endpoint: row.get("endpoint")?, config_json: row.get("config_json")?, environment: row.get("environment")?, remark: row.get("remark")?, created_at: row.get("created_at")?, updated_at: row.get("updated_at")?, }) } // ============================================================ // Repo 实现 // ============================================================ impl_repo!( /// 项目基础设施配置表 CRUD(project_services,V31,可改非审计)。 /// /// 走 `impl_repo!` 宏:表有 `updated_at`,标准 CRUD 全可用,白名单在 settings.rs 登记。 /// 注:宏生成的 `update_field` 经 settings 白名单(不含 service_type)收口——service_type /// 变更须走 `update_full`(过应用层校验 + 凭证审查),不可旁路 update_field 改类型。 ProjectServiceRepo, ProjectServiceRecord, "project_services", from_row => |row| project_service_from_row(row), insert => |conn, rec| { conn.execute( "INSERT INTO project_services (id, project_id, name, service_type, endpoint, config_json, environment, remark, created_at, updated_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)", params![ rec.id, rec.project_id, rec.name, rec.service_type, rec.endpoint, rec.config_json, rec.environment, rec.remark, rec.created_at, rec.updated_at ], ) }, update => |conn, rec| { conn.execute( "UPDATE project_services SET project_id = ?1, name = ?2, service_type = ?3, endpoint = ?4, config_json = ?5, environment = ?6, remark = ?7, updated_at = ?8 WHERE id = ?9", params![ rec.project_id, rec.name, rec.service_type, rec.endpoint, rec.config_json, rec.environment, rec.remark, rec.updated_at, rec.id ], ) } ); impl ProjectServiceRepo { /// 插入一条基础设施配置(应用层校验 service_type 白名单 + 凭证审查 D10 后再落库)。 /// /// 校验 fail-fast 在 spawn_blocking 前(非法值不进 DB 层),与 TaskLinkRepo::create_link /// 同款。`created_at` / `updated_at` 由本方法内部取当前毫秒时间戳覆盖(调用方传入的不可信), /// 保证「写入时机即落库时机」(对标 ProjectEventRepo::insert)。 pub async fn insert_validated(&self, record: ProjectServiceRecord) -> Result { // 应用层校验先行(fail-fast,非法值/凭证不进 DB 层) validate_record(&record)?; let conn = self.conn.clone(); let now = now_millis_str(); let mut rec = record; rec.created_at = now.clone(); rec.updated_at = now; tokio::task::spawn_blocking(move || { let guard = conn.blocking_lock(); let id = rec.id.clone(); let conn = &*guard; let r = &rec; conn.execute( "INSERT INTO project_services (id, project_id, name, service_type, endpoint, config_json, environment, remark, created_at, updated_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)", params![ r.id, r.project_id, r.name, r.service_type, r.endpoint, r.config_json, r.environment, r.remark, r.created_at, r.updated_at ], ) .map_err(storage_err)?; Ok(id) }) .await .map_err(storage_err)? } /// 整体更新(应用层校验 service_type + 凭证审查 D10 后再落库,保留 id 与 created_at)。 /// /// 对标宏生成的 `update_full`,但在落库前过 `validate_record`(宏无业务前置逻辑)。 /// `updated_at` 由本方法内部取当前毫秒时间戳覆盖。 pub async fn update_full_validated(&self, record: &ProjectServiceRecord) -> Result { validate_record(record)?; let conn = self.conn.clone(); let mut rec = record.clone(); rec.updated_at = now_millis_str(); tokio::task::spawn_blocking(move || { let guard = conn.blocking_lock(); let conn = &*guard; let r = &rec; let affected = conn.execute( "UPDATE project_services SET project_id = ?1, name = ?2, service_type = ?3, endpoint = ?4, config_json = ?5, environment = ?6, remark = ?7, updated_at = ?8 WHERE id = ?9", params![ r.project_id, r.name, r.service_type, r.endpoint, r.config_json, r.environment, r.remark, r.updated_at, r.id ], ) .map_err(storage_err)?; Ok(affected > 0) }) .await .map_err(storage_err)? } /// 按项目查询全部基础设施配置(命中 idx_project_services_project,created_at DESC)。 /// /// AI 项目上下文注入 / Dashboard 项目基础设施视图用:取某项目的全部服务配置。 pub async fn list_by_project( &self, project_id: &str, ) -> Result> { let conn = self.conn.clone(); let project_id = project_id.to_owned(); tokio::task::spawn_blocking(move || { let guard = conn.blocking_lock(); let mut stmt = guard .prepare( "SELECT id, project_id, name, service_type, endpoint, config_json, environment, remark, created_at, updated_at \ FROM project_services WHERE project_id = ?1 ORDER BY created_at DESC", ) .map_err(storage_err)?; let rows = stmt .query_map(params![project_id], project_service_from_row) .map_err(storage_err)?; let mut results = Vec::new(); for r in rows { results.push(r.map_err(storage_err)?); } Ok(results) }) .await .map_err(storage_err)? } /// 按项目 + 环境查询基础设施配置(命中 idx_project_services_project,environment 等值过滤)。 /// /// AI 执行任务时按当前环境取基础设施:production 部署任务只取 production 服务, /// 开发调试取 development 服务。环境隔离避免误连生产库。 pub async fn list_by_project_env( &self, project_id: &str, environment: &str, ) -> Result> { let conn = self.conn.clone(); let project_id = project_id.to_owned(); let environment = environment.to_owned(); tokio::task::spawn_blocking(move || { let guard = conn.blocking_lock(); let mut stmt = guard .prepare( "SELECT id, project_id, name, service_type, endpoint, config_json, environment, remark, created_at, updated_at \ FROM project_services WHERE project_id = ?1 AND environment = ?2 ORDER BY created_at DESC", ) .map_err(storage_err)?; let rows = stmt .query_map(params![project_id, environment], project_service_from_row) .map_err(storage_err)?; let mut results = Vec::new(); for r in rows { results.push(r.map_err(storage_err)?); } Ok(results) }) .await .map_err(storage_err)? } } // ============================================================ // 单元测试 — ProjectServiceRepo CRUD + 校验 + 环境过滤(内存 DB, // 对标 project_event_repo 测试:先建占位 project 满足 FK,再插服务) // ============================================================ #[cfg(test)] mod tests { use super::*; use crate::crud::ProjectRepo; use crate::models::ProjectRecord; use df_types::types::ProjectStatus; /// 构造内存 DB + 占位 project(project_services.project_id FK 要求 projects 存在)。 async fn setup() -> (crate::db::Database, ProjectServiceRepo) { let db = crate::db::Database::open_in_memory() .await .expect("open_in_memory"); let repo = ProjectServiceRepo::new(&db); let project_repo = ProjectRepo::new(&db); project_repo .insert(ProjectRecord { id: "proj-1".to_string(), name: "proj-1".to_string(), description: String::new(), status: ProjectStatus::Planning, idea_id: None, path: None, stack: None, created_at: "1700000000000".to_string(), updated_at: "1700000000000".to_string(), }) .await .unwrap(); (db, repo) } /// 构造一条 ProjectServiceRecord fixture(全字段,环境可指定)。 fn srec(id: &str, service_type: &str, environment: &str) -> ProjectServiceRecord { ProjectServiceRecord { id: id.to_string(), project_id: "proj-1".to_string(), name: format!("svc-{id}"), service_type: service_type.to_string(), endpoint: Some("localhost:3306".to_string()), config_json: Some(r#"{"pool_size":10}"#.to_string()), environment: environment.to_string(), remark: Some("主库".to_string()), created_at: "1700000000000".to_string(), updated_at: "1700000000000".to_string(), } } #[tokio::test] async fn validate_service_type_rejects_unknown() { // 白名单内全部通过 assert!(validate_service_type("mysql").is_ok()); assert!(validate_service_type("postgresql").is_ok()); assert!(validate_service_type("sqlite").is_ok()); assert!(validate_service_type("redis").is_ok()); assert!(validate_service_type("mongodb").is_ok()); assert!(validate_service_type("mq").is_ok()); assert!(validate_service_type("api").is_ok()); assert!(validate_service_type("other").is_ok()); // 非法值拒绝 assert!(validate_service_type("elasticsearch").is_err()); assert!(validate_service_type("").is_err()); } #[tokio::test] async fn insert_validated_and_read_back_full_fields() { let (_db, repo) = setup().await; let id = repo .insert_validated(srec("s1", "mysql", "development")) .await .unwrap(); assert_eq!(id, "s1"); let got = repo.get_by_id("s1").await.unwrap().expect("应存在"); // 全字段回读校验 assert_eq!(got.id, "s1"); assert_eq!(got.project_id, "proj-1"); assert_eq!(got.name, "svc-s1"); assert_eq!(got.service_type, "mysql"); assert_eq!(got.endpoint.as_deref(), Some("localhost:3306")); assert_eq!(got.config_json.as_deref(), Some(r#"{"pool_size":10}"#)); assert_eq!(got.environment, "development"); assert_eq!(got.remark.as_deref(), Some("主库")); // created_at / updated_at 由存储层覆盖为当前毫秒 assert!(!got.created_at.is_empty()); assert!(!got.updated_at.is_empty()); } #[tokio::test] async fn insert_rejects_invalid_service_type() { let (_db, repo) = setup().await; let err = repo .insert_validated(srec("s1", "elasticsearch", "development")) .await; assert!(err.is_err(), "非法 service_type 应被拒绝"); // 未入库 assert!(repo.get_by_id("s1").await.unwrap().is_none()); } #[tokio::test] async fn insert_rejects_sensitive_in_config_json() { // D10 边界:config_json 含 password 应被拒绝 let (_db, repo) = setup().await; let mut rec = srec("s1", "mysql", "development"); rec.config_json = Some(r#"{"password":"secret123"}"#.to_string()); let err = repo.insert_validated(rec).await; assert!(err.is_err(), "config_json 含 password 应被 D10 拒绝"); assert!(repo.get_by_id("s1").await.unwrap().is_none()); } #[tokio::test] async fn insert_rejects_sensitive_in_endpoint() { // endpoint query 参数带 token= 应被拒绝 let (_db, repo) = setup().await; let mut rec = srec("s1", "api", "development"); rec.endpoint = Some("https://api.example.com?token=abc123".to_string()); let err = repo.insert_validated(rec).await; assert!(err.is_err(), "endpoint 含 token 应被 D10 拒绝"); } #[tokio::test] async fn insert_rejects_sensitive_in_remark() { let (_db, repo) = setup().await; let mut rec = srec("s1", "redis", "development"); rec.remark = Some("API_KEY=sk-xxxx".to_string()); let err = repo.insert_validated(rec).await; assert!(err.is_err(), "remark 含 api_key 应被 D10 拒绝"); } #[tokio::test] async fn update_full_validated_rejects_sensitive() { // 先插一条合法的,update 时灌入凭证应被拒 let (_db, repo) = setup().await; repo.insert_validated(srec("s1", "mysql", "development")) .await .unwrap(); let mut bad = srec("s1", "mysql", "development"); bad.config_json = Some(r#"{"password":"new"}"#.to_string()); let err = repo.update_full_validated(&bad).await; assert!(err.is_err(), "update 灌凭证应被 D10 拒绝"); // 原记录未变(config_json 仍是 pool_size) let got = repo.get_by_id("s1").await.unwrap().unwrap(); assert_eq!(got.config_json.as_deref(), Some(r#"{"pool_size":10}"#)); } #[tokio::test] async fn update_full_validated_changes_fields() { let (_db, repo) = setup().await; repo.insert_validated(srec("s1", "mysql", "development")) .await .unwrap(); let mut updated = srec("s1", "redis", "production"); updated.name = "缓存".to_string(); let hit = repo.update_full_validated(&updated).await.unwrap(); assert!(hit, "应命中更新"); let got = repo.get_by_id("s1").await.unwrap().unwrap(); assert_eq!(got.name, "缓存"); assert_eq!(got.service_type, "redis"); assert_eq!(got.environment, "production"); } #[tokio::test] async fn list_by_project_returns_all() { let (_db, repo) = setup().await; repo.insert_validated(srec("s1", "mysql", "development")) .await .unwrap(); repo.insert_validated(srec("s2", "redis", "development")) .await .unwrap(); repo.insert_validated(srec("s3", "mq", "production")) .await .unwrap(); // 按项目查返回全部 3 条(含所有环境) let got = repo.list_by_project("proj-1").await.unwrap(); assert_eq!(got.len(), 3); // 不存在的项目返回空 let empty = repo.list_by_project("no-such").await.unwrap(); assert!(empty.is_empty()); } #[tokio::test] async fn list_by_project_env_filters_environment() { let (_db, repo) = setup().await; repo.insert_validated(srec("s1", "mysql", "development")) .await .unwrap(); repo.insert_validated(srec("s2", "redis", "development")) .await .unwrap(); repo.insert_validated(srec("s3", "mysql", "production")) .await .unwrap(); // 只取 development:2 条(s1/s2) let dev = repo .list_by_project_env("proj-1", "development") .await .unwrap(); assert_eq!(dev.len(), 2); for s in &dev { assert_eq!(s.environment, "development"); } // 只取 production:1 条(s3) let prod = repo .list_by_project_env("proj-1", "production") .await .unwrap(); assert_eq!(prod.len(), 1); assert_eq!(prod[0].id, "s3"); // 不存在的环境返回空 let staging = repo .list_by_project_env("proj-1", "staging") .await .unwrap(); assert!(staging.is_empty()); } #[tokio::test] async fn delete_removes_record() { let (_db, repo) = setup().await; repo.insert_validated(srec("s1", "mysql", "development")) .await .unwrap(); assert!(repo.delete("s1").await.unwrap()); assert!(repo.get_by_id("s1").await.unwrap().is_none()); // 再删返回 false assert!(!repo.delete("s1").await.unwrap()); } /// None 字段(config_json/endpoint/remark)也能正常插入与查询。 #[tokio::test] async fn insert_with_null_optional_fields() { let (_db, repo) = setup().await; let rec = ProjectServiceRecord { id: "s1".to_string(), project_id: "proj-1".to_string(), name: "bare-svc".to_string(), service_type: "api".to_string(), endpoint: None, config_json: None, environment: "development".to_string(), remark: None, created_at: "1700000000000".to_string(), updated_at: "1700000000000".to_string(), }; repo.insert_validated(rec).await.unwrap(); let got = repo.get_by_id("s1").await.unwrap().unwrap(); assert!(got.endpoint.is_none()); assert!(got.config_json.is_none()); assert!(got.remark.is_none()); } }