From fdc718b18e4bc2d50076ba30d1f05c3e4abe2b1f Mon Sep 17 00:00:00 2001 From: KiriAky 107 Date: Tue, 8 Sep 2026 21:22:20 +0800 Subject: [PATCH] =?UTF-8?q?feat(extensions):=20=E5=8E=9F=E5=AD=90=E5=88=87?= =?UTF-8?q?=E6=8D=A2=E5=B9=B6=E6=81=A2=E5=A4=8D=E5=8C=85=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=E7=BB=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../OpenNexus生产化实施进度-2026-09-08.md | 10 + frontend/src-tauri/src/extension_store.rs | 175 +++++++++- .../src-tauri/src/extension_transaction.rs | 298 ++++++++++++++++++ frontend/src-tauri/src/lib.rs | 3 + 4 files changed, 481 insertions(+), 5 deletions(-) create mode 100644 frontend/src-tauri/src/extension_transaction.rs diff --git a/docs/development/OpenNexus生产化实施进度-2026-09-08.md b/docs/development/OpenNexus生产化实施进度-2026-09-08.md index 7356e3a..4151a39 100644 --- a/docs/development/OpenNexus生产化实施进度-2026-09-08.md +++ b/docs/development/OpenNexus生产化实施进度-2026-09-08.md @@ -207,3 +207,13 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写 - Authority 使用系统随机密钥，销毁时清零；到期边界、不同 Authority、全部失效后的旧令牌均拒绝。环境凭据使用 scope 声明，调用方不得向声明填入实际秘密。输入和编码大小有上限，入口穿越、错误环境名和 HTTP 社区来源拒绝。 - 测试逐字段改成另一份仍合法的声明，确认认证码不再匹配，避免把输入格式拒绝误认为绑定测试通过；另测到期、密钥轮换、路径及体积边界。21 项扩展回归与全目标 Clippy -D warnings 通过，日志 `.build/extension-permit-tests.log`。 - 这是许可认证原语，尚未接入安装确认、在线信任、锁定事件、运行沙箱或 broker。它不证明用户已同意，也不授予执行能力；完整 C/D 验收仍未通过。 + + +## 增量：包与配置活动指针的组事务 + +- extension_transaction 在单一 SQLite 事务中保存旧状态、整组新包/配置指针和 checking 日志；期望 revision 不匹配、重复 slot、已有待检查事务或操作 ID 被不同请求重用均拒绝。健康结果提交后才清除 pending；失败整体恢复旧元组，首次安装失败移除新指针。 +- 暂存库 schema 3 接入，升级 schema 1/2 前保留对应备份。switch_prepared 重新检查签名归档和准备目录，并按 Vault、来源、命名空间和包 ID 推导 slot，拒绝跨 Vault 指针。打开库时未完成 checking 自动回滚，不恢复任何执行许可。 +- 三个切换边界分别注入错误并重开 20 次，两个包与各自配置始终匹配；另测 CAS、busy、操作重放、失败首次安装、健康成功后重开保留、目录损坏拒绝及 Vault 绑定。24 项扩展回归和全目标 Clippy -D warnings 通过。 +- 这仍是安装事务的存储层：配置只要求对象，完整类型 schema/秘密剥离、依赖计划和用户确认复核、在线撤回、停旧实例、沙箱内迁移、真实健康探测及 UI 尚待编排。不能把注入错误重开测试当成 D-03 断电/磁盘满完整矩阵，也不能开启 extensions capability。 + +- 本轮 Rust desktop 全量回归 78 项通过，2 个既有特殊入口 ignored；实际 Sync 子进程中断恢复由父测试执行通过。全量日志 `.build/extension-transaction-rust-full.log`。 diff --git a/frontend/src-tauri/src/extension_store.rs b/frontend/src-tauri/src/extension_store.rs index f6e026e..fb58658 100644 --- a/frontend/src-tauri/src/extension_store.rs +++ b/frontend/src-tauri/src/extension_store.rs @@ -236,14 +236,17 @@ impl ExtensionStore { ordinary(&sidecar)?; } } - let db = Connection::open(database)?; + let mut db = Connection::open(database)?; db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;")?; let version: i64 = db.query_row("PRAGMA user_version", [], |r| r.get(0))?; - if version > 2 { + if version > 3 { return Err(HostError::new("EXTENSION_SCHEMA_INCOMPATIBLE")); } - if version == 1 { - let backup = root.join(format!("extensions.schema1.{}.sqlite3", Uuid::new_v4())); + if (1..3).contains(&version) { + let backup = root.join(format!( + "extensions.schema{version}.{}.sqlite3", + Uuid::new_v4() + )); db.execute("VACUUM INTO ?1", [backup.to_string_lossy().as_ref()])?; OpenOptions::new().write(true).open(backup)?.sync_all()?; } @@ -251,13 +254,88 @@ impl ExtensionStore { CREATE TABLE IF NOT EXISTS versions (package_key TEXT PRIMARY KEY,source TEXT NOT NULL,namespace TEXT NOT NULL,package_id TEXT NOT NULL,version TEXT NOT NULL,fingerprint TEXT NOT NULL,release TEXT NOT NULL,manifest TEXT NOT NULL,inventory TEXT NOT NULL,archive_hash TEXT NOT NULL,size INTEGER NOT NULL,signer BLOB NOT NULL,state TEXT NOT NULL); CREATE TABLE IF NOT EXISTS stage_operations (id TEXT PRIMARY KEY,fingerprint TEXT NOT NULL,receipt TEXT NOT NULL); CREATE TABLE IF NOT EXISTS prepared_packages (package_key TEXT PRIMARY KEY REFERENCES versions(package_key),directory TEXT NOT NULL UNIQUE,tree_sha256 TEXT NOT NULL); - PRAGMA user_version=2; COMMIT;")?; + CREATE TABLE IF NOT EXISTS extension_active(slot TEXT PRIMARY KEY,target TEXT NOT NULL,revision TEXT NOT NULL,pending_operation TEXT); + CREATE TABLE IF NOT EXISTS extension_transactions(id TEXT PRIMARY KEY,fingerprint TEXT NOT NULL,before_state TEXT NOT NULL,after_state TEXT NOT NULL,state TEXT NOT NULL); + PRAGMA user_version=3; COMMIT;")?; + crate::extension_transaction::recover(&mut db)?; Ok(Self { root, db, _lock: lock, }) } + /// Atomically selects a prepared group after installer policy checks. This + /// method does not stop processes, validate configuration schemas or issue permits. + pub fn switch_prepared( + &mut self, + operation: &str, + vault_id: &str, + changes: &[crate::extension_transaction::Change], + ) -> Result { + use cap_fs_ext::DirExt; + let vault = Uuid::parse_str(vault_id) + .map_err(|_| HostError::new("VAULT_INVALID"))? + .to_string(); + if vault != vault_id || changes.is_empty() || changes.len() > 200 { + return Err(HostError::new("EXTENSION_TRANSACTION_INVALID")); + } + let root = cap_std::fs::Dir::open_ambient_dir(&self.root, cap_std::ambient_authority())?; + let prepared = root.open_dir_nofollow("prepared")?; + for change in changes { + let (source, json, key, directory, tree): (String,String,Vec,String,String) = self.db.query_row( + "SELECT v.source,v.release,v.signer,p.directory,p.tree_sha256 FROM versions v JOIN prepared_packages p ON p.package_key=v.package_key WHERE v.package_key=?1", + [&change.target.package_key], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?)))?; + let release: Release = serde_json::from_str(&json) + .map_err(|_| HostError::new("EXTENSION_STORE_CORRUPT"))?; + let public: [u8; 32] = key + .try_into() + .map_err(|_| HostError::new("EXTENSION_STORE_CORRUPT"))?; + let (inventory, _) = release.verify_package( + &public, + &release.key_id, + &release.namespace, + false, + false, + &self.archive(&change.target.package_key)?, + )?; + let slot = hash( + &serde_json::to_vec(&(&vault, &source, &release.namespace, &release.package_id)) + .unwrap(), + ); + if change.target.slot != slot + || change.target.directory != directory + || change.target.tree_sha256 != tree + || Uuid::parse_str(&directory) + .map(|v| v.to_string()) + .ok() + .as_ref() + != Some(&directory) + { + return Err(HostError::new("EXTENSION_INSTALL_CONFLICT")); + } + if crate::extension_unpack::verify_tree( + &prepared.open_dir_nofollow(&directory)?, + &inventory, + )? != tree + { + return Err(HostError::new("EXTENSION_STORE_CORRUPT")); + } + } + crate::extension_transaction::switch(&mut self.db, operation, changes) + } + pub fn finish_installation( + &mut self, + operation: &str, + healthy: bool, + ) -> Result { + crate::extension_transaction::finish(&mut self.db, operation, healthy) + } + pub fn active_installation( + &self, + slot: &str, + ) -> Result> { + crate::extension_transaction::active(&self.db, slot) + } /// Prepare a verified staged package. The caller supplies current signer/revocation /// policy; persisted preparation does not bypass that policy on replay. pub fn prepare( @@ -544,6 +622,93 @@ mod tests { } } #[test] + fn prepared_switch_rechecks_content_vault_binding_and_recovers_on_open() { + use crate::extension_transaction::{Change, Target}; + let temp = tempfile::tempdir().unwrap(); + let (release, archive, key) = fixture(); + let mut store = ExtensionStore::open(temp.path()).unwrap(); + let staged = store + .stage(request( + &Uuid::new_v4().to_string(), + &release, + &archive, + &key, + )) + .unwrap(); + let prepared = store + .prepare( + &staged.package_key, + Signer { + public_key: &key, + key_id: "test-key", + namespace: "examples", + revoked: false, + }, + false, + ) + .unwrap(); + let vault = Uuid::new_v4().to_string(); + let slot = hash( + &serde_json::to_vec(&( + &vault, + "https://catalog.example/", + &release.namespace, + &release.package_id, + )) + .unwrap(), + ); + let changes = vec![Change { + target: Target { + slot: slot.clone(), + package_key: staged.package_key, + directory: prepared.directory.clone(), + tree_sha256: prepared.tree_sha256, + configuration: serde_json::json!({"review":true}), + }, + expected_revision: None, + }]; + assert!(store + .switch_prepared( + &Uuid::new_v4().to_string(), + &Uuid::new_v4().to_string(), + &changes + ) + .is_err()); + let operation = Uuid::new_v4().to_string(); + store.switch_prepared(&operation, &vault, &changes).unwrap(); + assert!(store + .active_installation(&slot) + .unwrap() + .unwrap() + .pending_operation + .is_some()); + drop(store); + let mut store = ExtensionStore::open(temp.path()).unwrap(); + assert!(store.active_installation(&slot).unwrap().is_none()); + assert_eq!( + store.finish_installation(&operation, true).unwrap().state, + "rolled_back" + ); + let operation = Uuid::new_v4().to_string(); + store.switch_prepared(&operation, &vault, &changes).unwrap(); + store.finish_installation(&operation, true).unwrap(); + drop(store); + let mut store = ExtensionStore::open(temp.path()).unwrap(); + assert_eq!( + store.active_installation(&slot).unwrap().unwrap().target, + changes[0].target + ); + fs::write( + temp.path() + .join("prepared") + .join(&prepared.directory) + .join("persona.json"), + b"bad", + ) + .unwrap(); + assert!(store.switch_prepared(&operation, &vault, &changes).is_err()); + } + #[test] fn schema_one_upgrade_preserves_versions_and_creates_readable_backup() { let temp = tempfile::tempdir().unwrap(); let (release, archive, key) = fixture(); diff --git a/frontend/src-tauri/src/extension_transaction.rs b/frontend/src-tauri/src/extension_transaction.rs new file mode 100644 index 0000000..78ac7c7 --- /dev/null +++ b/frontend/src-tauri/src/extension_transaction.rs @@ -0,0 +1,298 @@ +//! Atomic package/configuration pointers. A pointer is never a runtime permission. +use crate::workspace::{hash, HostError, Result}; +use rusqlite::{params, Connection, OptionalExtension}; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeSet; + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct Target { + pub slot: String, + pub package_key: String, + pub directory: String, + pub tree_sha256: String, + pub configuration: serde_json::Value, +} +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub struct Change { + pub target: Target, + pub expected_revision: Option, +} +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub struct Active { + pub target: Target, + pub revision: String, + pub pending_operation: Option, +} +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct Receipt { + pub operation_id: String, + pub state: String, +} +pub fn schema(db: &Connection) -> Result<()> { + db.execute_batch("CREATE TABLE IF NOT EXISTS extension_active(slot TEXT PRIMARY KEY,target TEXT NOT NULL,revision TEXT NOT NULL,pending_operation TEXT); + CREATE TABLE IF NOT EXISTS extension_transactions(id TEXT PRIMARY KEY,fingerprint TEXT NOT NULL,before_state TEXT NOT NULL,after_state TEXT NOT NULL,state TEXT NOT NULL);")?; + Ok(()) +} +pub fn active(db: &Connection, slot: &str) -> Result> { + let row: Option<(String, String, Option)> = db + .query_row( + "SELECT target,revision,pending_operation FROM extension_active WHERE slot=?1", + [slot], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), + ) + .optional()?; + row.map(|(target, revision, pending_operation)| { + Ok(Active { + target: serde_json::from_str(&target) + .map_err(|_| HostError::new("EXTENSION_STORE_CORRUPT"))?, + revision, + pending_operation, + }) + }) + .transpose() +} +fn encoded(value: &T) -> Result { + serde_json::to_string(value).map_err(|_| HostError::new("EXTENSION_TRANSACTION_INVALID")) +} +pub fn switch(db: &mut Connection, operation: &str, changes: &[Change]) -> Result { + switch_inner(db, operation, changes, |_| Ok(())) +} +fn switch_inner( + db: &mut Connection, + operation: &str, + changes: &[Change], + mut checkpoint: impl FnMut(&str) -> Result<()>, +) -> Result { + if uuid::Uuid::parse_str(operation).is_err() || changes.is_empty() || changes.len() > 200 { + return Err(HostError::new("EXTENSION_TRANSACTION_INVALID")); + } + let serialized = encoded(&changes)?; + if serialized.len() > 4 * 1024 * 1024 { + return Err(HostError::new("EXTENSION_TRANSACTION_INVALID")); + } + let fingerprint = hash(serialized.as_bytes()); + let transaction = db.transaction()?; + let prior: Option<(String, String)> = transaction + .query_row( + "SELECT fingerprint,state FROM extension_transactions WHERE id=?1", + [operation], + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .optional()?; + if let Some((previous, state)) = prior { + if previous != fingerprint { + return Err(HostError::new("OPERATION_REUSED")); + } + return Ok(Receipt { + operation_id: operation.into(), + state, + }); + } + let mut slots = BTreeSet::new(); + let mut before = Vec::new(); + for change in changes { + let target = &change.target; + for digest in [&target.slot, &target.package_key, &target.tree_sha256] { + if digest.len() != 64 + || !digest + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) + { + return Err(HostError::new("EXTENSION_TRANSACTION_INVALID")); + } + } + if uuid::Uuid::parse_str(&target.directory) + .map(|v| v.to_string()) + .ok() + .as_ref() + != Some(&target.directory) + || !target.configuration.is_object() + || !slots.insert(&target.slot) + { + return Err(HostError::new("EXTENSION_TRANSACTION_INVALID")); + } + let previous = active(&transaction, &target.slot)?; + if previous + .as_ref() + .is_some_and(|p| p.pending_operation.is_some()) + { + return Err(HostError::new("EXTENSION_TRANSACTION_BUSY")); + } + if previous.as_ref().map(|p| &p.revision) != change.expected_revision.as_ref() { + return Err(HostError::new("EXTENSION_INSTALL_CONFLICT")); + } + before.push(previous); + } + transaction.execute( + "INSERT INTO extension_transactions VALUES (?1,?2,?3,?4,'checking')", + params![operation, fingerprint, encoded(&before)?, serialized], + )?; + checkpoint("journal_recorded")?; + for change in changes { + let target = encoded(&change.target)?; + transaction.execute("INSERT INTO extension_active VALUES (?1,?2,?3,?4) ON CONFLICT(slot) DO UPDATE SET target=excluded.target,revision=excluded.revision,pending_operation=excluded.pending_operation", + params![change.target.slot,target,hash(target.as_bytes()),operation])?; + checkpoint("pointer_recorded")?; + } + transaction.commit()?; + checkpoint("switch_committed")?; + Ok(Receipt { + operation_id: operation.into(), + state: "checking".into(), + }) +} + +/// `healthy` must come from the Host's matching package/config health probe. +/// Recovery calls this with false; it never reissues any execution permits. +pub fn finish(db: &mut Connection, operation: &str, healthy: bool) -> Result { + let tx = db.transaction()?; + let (before, after, state): (String, String, String) = tx.query_row( + "SELECT before_state,after_state,state FROM extension_transactions WHERE id=?1", + [operation], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), + )?; + if state != "checking" { + return Ok(Receipt { + operation_id: operation.into(), + state, + }); + } + let changes: Vec = + serde_json::from_str(&after).map_err(|_| HostError::new("EXTENSION_STORE_CORRUPT"))?; + let previous: Vec> = + serde_json::from_str(&before).map_err(|_| HostError::new("EXTENSION_STORE_CORRUPT"))?; + if previous.len() != changes.len() { + return Err(HostError::new("EXTENSION_STORE_CORRUPT")); + } + for (change, old) in changes.iter().zip(previous) { + let current = active(&tx, &change.target.slot)? + .ok_or_else(|| HostError::new("EXTENSION_STORE_CORRUPT"))?; + if current.pending_operation.as_deref() != Some(operation) + || current.target != change.target + { + return Err(HostError::new("EXTENSION_STORE_CORRUPT")); + } + if healthy { + tx.execute( + "UPDATE extension_active SET pending_operation=NULL WHERE slot=?1", + [&change.target.slot], + )?; + } else if let Some(old) = old { + tx.execute("UPDATE extension_active SET target=?2,revision=?3,pending_operation=NULL WHERE slot=?1", + params![change.target.slot,encoded(&old.target)?,old.revision])?; + } else { + tx.execute( + "DELETE FROM extension_active WHERE slot=?1", + [&change.target.slot], + )?; + } + } + let state = if healthy { "complete" } else { "rolled_back" }; + tx.execute( + "UPDATE extension_transactions SET state=?2 WHERE id=?1", + params![operation, state], + )?; + tx.commit()?; + Ok(Receipt { + operation_id: operation.into(), + state: state.into(), + }) +} +pub fn recover(db: &mut Connection) -> Result { + let ids: Vec = db + .prepare("SELECT id FROM extension_transactions WHERE state='checking' ORDER BY id")? + .query_map([], |r| r.get(0))? + .collect::>()?; + for id in &ids { + finish(db, id, false)?; + } + Ok(ids.len()) +} + +#[cfg(test)] +mod tests { + use super::*; + fn open(path: &std::path::Path) -> Connection { + let db = Connection::open(path).unwrap(); + db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;") + .unwrap(); + schema(&db).unwrap(); + db + } + fn change(slot: char, version: u8, previous: Option) -> Change { + Change { + target: Target { + slot: slot.to_string().repeat(64), + package_key: format!("{version:x}").repeat(64), + directory: uuid::Uuid::new_v4().to_string(), + tree_sha256: "a".repeat(64), + configuration: serde_json::json!({"version": version}), + }, + expected_revision: previous, + } + } + #[test] + fn group_switch_crashes_recover_matching_packages_and_configuration() { + for boundary in ["journal_recorded", "pointer_recorded", "switch_committed"] { + for _ in 0..20 { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("state.sqlite3"); + let mut db = open(&path); + let old = vec![change('a', 1, None), change('b', 1, None)]; + let id = uuid::Uuid::new_v4().to_string(); + switch(&mut db, &id, &old).unwrap(); + finish(&mut db, &id, true).unwrap(); + let next: Vec<_> = old + .iter() + .map(|c| { + change( + c.target.slot.chars().next().unwrap(), + 2, + Some(active(&db, &c.target.slot).unwrap().unwrap().revision), + ) + }) + .collect(); + let update = uuid::Uuid::new_v4().to_string(); + assert!( + switch_inner(&mut db, &update, &next, |at| if at == boundary { + Err(HostError::new("INJECTED")) + } else { + Ok(()) + }) + .is_err() + ); + drop(db); + let mut db = open(&path); + recover(&mut db).unwrap(); + for original in &old { + let current = active(&db, &original.target.slot).unwrap().unwrap(); + assert_eq!(current.target, original.target); + assert!(current.pending_operation.is_none()); + } + assert_eq!(recover(&mut db).unwrap(), 0); + } + } + } + #[test] + fn cas_busy_replay_and_failed_first_install() { + let mut db = Connection::open_in_memory().unwrap(); + schema(&db).unwrap(); + let c = vec![change('a', 1, None)]; + let id = uuid::Uuid::new_v4().to_string(); + switch(&mut db, &id, &c).unwrap(); + assert!(switch(&mut db, &uuid::Uuid::new_v4().to_string(), &c).is_err()); + assert_eq!(switch(&mut db, &id, &c).unwrap().state, "checking"); + finish(&mut db, &id, false).unwrap(); + assert!(active(&db, &c[0].target.slot).unwrap().is_none()); + assert_eq!(switch(&mut db, &id, &c).unwrap().state, "rolled_back"); + assert_eq!(finish(&mut db, &id, true).unwrap().state, "rolled_back"); + let id = uuid::Uuid::new_v4().to_string(); + switch(&mut db, &id, &c).unwrap(); + finish(&mut db, &id, true).unwrap(); + assert!(switch(&mut db, &uuid::Uuid::new_v4().to_string(), &c).is_err()); + let mut different = c.clone(); + different[0].target.configuration = serde_json::json!({"changed":true}); + assert!(switch(&mut db, &id, &different).is_err()); + } +} diff --git a/frontend/src-tauri/src/lib.rs b/frontend/src-tauri/src/lib.rs index 087ca31..d5622b2 100644 --- a/frontend/src-tauri/src/lib.rs +++ b/frontend/src-tauri/src/lib.rs @@ -42,3 +42,6 @@ pub mod extension_unpack; #[cfg(feature = "desktop")] pub mod extension_permit; + +#[cfg(feature = "desktop")] +pub mod extension_transaction;