admin/CC-Switch
1
0
Fork 0
mirror of https://github.com/farion1231/cc-switch.git synced 2026-08-04 19:45:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(database): enforce canonical restore trust boundaries

Browse Source
This commit is contained in:
SaladDay
2026-08-01 16:01:00 +00:00
parent 89961dff28
commit 28530ff641
8 changed files with 1064 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/database/backup.rs
+328 -17
View File
@@ -2,7 +2,7 @@
//!
//! 提供 SQL 导出/导入和二进制快照备份功能。
use super::schema::CanonicalStage;
use super::schema::{CanonicalStage, MigrationRunContext};
use super::{lock_conn, Database, SCHEMA_VERSION};
use crate::config::get_app_config_dir;
use crate::error::AppError;
@@ -208,12 +208,19 @@ enum IntegerDomain {
InputTokenSemantics,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RealDomain {
NotReal,
FiniteUnitInterval,
}
#[derive(Debug, Clone, Copy)]
struct RestoreColumnSpec {
name: &'static str,
storage: StorageKind,
nullable: bool,
integer_domain: IntegerDomain,
real_domain: RealDomain,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -244,6 +251,7 @@ macro_rules! text_col {
storage: StorageKind::Text,
nullable: false,
integer_domain: IntegerDomain::Unrestricted,
real_domain: RealDomain::NotReal,
}
};
}
@@ -255,6 +263,7 @@ macro_rules! nullable_text_col {
storage: StorageKind::Text,
nullable: true,
integer_domain: IntegerDomain::Unrestricted,
real_domain: RealDomain::NotReal,
}
};
}
@@ -266,6 +275,7 @@ macro_rules! integer_col {
storage: StorageKind::Integer,
nullable: false,
integer_domain: IntegerDomain::$domain,
real_domain: RealDomain::NotReal,
}
};
}
@@ -277,17 +287,19 @@ macro_rules! nullable_integer_col {
storage: StorageKind::Integer,
nullable: true,
integer_domain: IntegerDomain::$domain,
real_domain: RealDomain::NotReal,
}
};
}
macro_rules! real_col {
($name:literal) => {
($name:literal, $domain:ident) => {
RestoreColumnSpec {
name: $name,
storage: StorageKind::Real,
nullable: false,
integer_domain: IntegerDomain::Unrestricted,
real_domain: RealDomain::$domain,
}
};
}
@@ -398,7 +410,7 @@ const PROXY_CONFIG_RESTORE_COLUMNS: &[RestoreColumnSpec] = &[
integer_col!("circuit_failure_threshold", NonNegativeI32),
integer_col!("circuit_success_threshold", NonNegativeI32),
integer_col!("circuit_timeout_seconds", NonNegativeI32),
real_col!("circuit_error_rate_threshold"),
real_col!("circuit_error_rate_threshold", FiniteUnitInterval),
integer_col!("circuit_min_requests", NonNegativeI32),
text_col!("default_cost_multiplier"),
text_col!("pricing_model_source"),
@@ -761,6 +773,19 @@ fn same_open_file_identity(opened: &File, current: &File) -> std::io::Result<boo
Ok(windows_file_identity(opened)? == windows_file_identity(current)?)
}
#[cfg(windows)]
fn metadata_is_reparse_point(metadata: &Metadata) -> bool {
use std::os::windows::fs::MetadataExt;
use windows_sys::Win32::Storage::FileSystem::FILE_ATTRIBUTE_REPARSE_POINT;
metadata.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT != 0
}
#[cfg(not(windows))]
fn metadata_is_reparse_point(_metadata: &Metadata) -> bool {
false
}
#[cfg(all(not(unix), not(windows)))]
fn same_open_file_identity(opened: &File, current: &File) -> std::io::Result<bool> {
let _ = (opened, current);
@@ -836,11 +861,89 @@ fn open_backup_child(
#[cfg(windows)]
fn open_backup_child(
_directory: &File,
directory_path: &Path,
directory: &File,
_directory_path: &Path,
filename: &std::ffi::OsStr,
) -> std::io::Result<File> {
open_nofollow(&directory_path.join(filename))
use std::os::windows::ffi::OsStrExt;
use std::os::windows::io::{AsRawHandle, FromRawHandle};
use windows_sys::Wdk::Foundation::OBJECT_ATTRIBUTES;
use windows_sys::Wdk::Storage::FileSystem::{
NtCreateFile, FILE_NON_DIRECTORY_FILE, FILE_OPEN, FILE_OPEN_REPARSE_POINT,
FILE_SYNCHRONOUS_IO_NONALERT,
};
use windows_sys::Win32::Foundation::{
CloseHandle, RtlNtStatusToDosError, INVALID_HANDLE_VALUE, OBJ_CASE_INSENSITIVE,
UNICODE_STRING,
};
use windows_sys::Win32::Storage::FileSystem::{
FILE_ATTRIBUTE_NORMAL, FILE_READ_ATTRIBUTES, FILE_READ_DATA, FILE_SHARE_READ,
FILE_SHARE_WRITE, SYNCHRONIZE,
};
use windows_sys::Win32::System::IO::IO_STATUS_BLOCK;
let mut wide = filename.encode_wide().collect::<Vec<_>>();
if wide.contains(&0) {
return Err(std::io::Error::from(std::io::ErrorKind::InvalidInput));
}
let byte_length = wide
.len()
.checked_mul(std::mem::size_of::<u16>())
.and_then(|length| u16::try_from(length).ok())
.ok_or_else(|| std::io::Error::from(std::io::ErrorKind::InvalidInput))?;
let object_name = UNICODE_STRING {
Length: byte_length,
MaximumLength: byte_length,
Buffer: wide.as_mut_ptr(),
};
let object_attributes = OBJECT_ATTRIBUTES {
Length: std::mem::size_of::<OBJECT_ATTRIBUTES>() as u32,
RootDirectory: directory.as_raw_handle(),
ObjectName: std::ptr::addr_of!(object_name),
Attributes: OBJ_CASE_INSENSITIVE,
SecurityDescriptor: std::ptr::null(),
SecurityQualityOfService: std::ptr::null(),
};
let mut io_status = IO_STATUS_BLOCK::default();
let mut handle = INVALID_HANDLE_VALUE;
// SAFETY: `directory` remains live for the call and is installed as
// RootDirectory; `object_name` points to `wide` for the same duration.
// NtCreateFile writes only the handle and IO status output buffers.
let status = unsafe {
NtCreateFile(
std::ptr::addr_of_mut!(handle),
FILE_READ_DATA | FILE_READ_ATTRIBUTES | SYNCHRONIZE,
std::ptr::addr_of!(object_attributes),
std::ptr::addr_of_mut!(io_status),
std::ptr::null(),
FILE_ATTRIBUTE_NORMAL,
FILE_SHARE_READ | FILE_SHARE_WRITE,
FILE_OPEN,
FILE_NON_DIRECTORY_FILE | FILE_OPEN_REPARSE_POINT | FILE_SYNCHRONOUS_IO_NONALERT,
std::ptr::null(),
0,
)
};
if status < 0 {
if handle != INVALID_HANDLE_VALUE && !handle.is_null() {
// SAFETY: a non-invalid handle written on the failure path is
// still owned by this function and must not leak.
unsafe {
CloseHandle(handle);
}
}
return Err(std::io::Error::from_raw_os_error(
unsafe { RtlNtStatusToDosError(status) } as i32,
));
}
if handle == INVALID_HANDLE_VALUE || handle.is_null() {
return Err(std::io::Error::other(
"NtCreateFile succeeded without returning a file handle",
));
}
// SAFETY: NtCreateFile returned a new owned file handle; this transfers its
// sole ownership to File.
Ok(unsafe { File::from_raw_handle(handle) })
}
#[cfg(all(not(unix), not(windows)))]
@@ -918,7 +1021,7 @@ fn verify_open_file_still_current(
fn validate_regular_file(path: &Path, max_bytes: u64) -> Result<Metadata, AppError> {
let metadata = fs::symlink_metadata(path).map_err(|error| AppError::io(path, error))?;
if !metadata.file_type().is_file() {
if !metadata.file_type().is_file() || metadata_is_reparse_point(&metadata) {
return Err(AppError::InvalidInput(format!(
"restore source must be a regular non-symlink file: {}",
path.display()
@@ -1009,6 +1112,9 @@ fn read_restore_file(path: &Path, max_bytes: u64) -> Result<Vec<u8>, AppError> {
/// the only path SQLite ever opens, so replacing any component of the original
/// path cannot redirect the bytes consumed by the parser.
fn snapshot_binary_restore_file(path: &Path, max_bytes: u64) -> Result<NamedTempFile, AppError> {
// File-level preflight rejects links/reparse points and oversized inputs
// before the authoritative directory-handle-relative open below.
let initial = validate_regular_file(path, max_bytes)?;
let directory_path = path.parent().ok_or_else(|| {
AppError::InvalidInput(format!(
"binary restore source has no parent directory: {}",
@@ -1027,7 +1133,18 @@ fn snapshot_binary_restore_file(path: &Path, max_bytes: u64) -> Result<NamedTemp
let opened = source
.metadata()
.map_err(|error| AppError::io(path, error))?;
if !opened.file_type().is_file() || opened.len() > max_bytes {
#[cfg(unix)]
let changed_before_open = !same_file_identity(&initial, &opened);
#[cfg(not(unix))]
let changed_before_open = {
let _shape_only = initial;
false
};
if !opened.file_type().is_file()
|| metadata_is_reparse_point(&opened)
|| opened.len() > max_bytes
|| changed_before_open
{
return Err(AppError::InvalidInput(format!(
"binary restore source must be a bounded regular file: {}",
path.display()
@@ -1073,6 +1190,8 @@ fn snapshot_binary_restore_file(path: &Path, max_bytes: u64) -> Result<NamedTemp
.map_err(|error| AppError::io(directory_path, error))?;
if !same_source
|| !same_directory
|| metadata_is_reparse_point(&completed)
|| metadata_is_reparse_point(&current_metadata)
|| opened.len() != copied
|| completed.len() != copied
|| current_metadata.len() != copied
@@ -1240,8 +1359,11 @@ impl UntrustedScratch {
self.connection
.set_db_config(DbConfig::SQLITE_DBCONFIG_ENABLE_TRIGGER, true)
.map_err(|error| AppError::Database(error.to_string()))?;
Database::create_tables_on_conn(&self.connection)?;
Database::apply_schema_migrations_on_conn(&self.connection)?;
Database::create_tables_on_conn(&self.connection, MigrationRunContext::UntrustedRestore)?;
Database::apply_schema_migrations_on_conn(
&self.connection,
MigrationRunContext::UntrustedRestore,
)?;
self.connection
.execute_batch("PRAGMA foreign_keys = ON;")
.map_err(|error| AppError::Database(error.to_string()))?;
@@ -1412,6 +1534,28 @@ fn validate_integer_domain(
}
}
fn validate_real_domain(
table: &str,
column: &RestoreColumnSpec,
value: &Value,
) -> Result<(), AppError> {
let Value::Real(value) = value else {
return Ok(());
};
let valid = match column.real_domain {
RealDomain::NotReal => false,
RealDomain::FiniteUnitInterval => value.is_finite() && (0.0..=1.0).contains(value),
};
if valid {
Ok(())
} else {
Err(AppError::InvalidInput(format!(
"restore row has out-of-domain real {value} at {table}.{} ({:?})",
column.name, column.real_domain
)))
}
}
fn text_value<'a>(table: &str, column: &str, value: &'a Value) -> Result<&'a str, AppError> {
match value {
Value::Text(value) => Ok(value),
@@ -1457,6 +1601,7 @@ fn validate_restore_row(spec: &RestoreTableSpec, values: &[Value]) -> Result<(),
for (column, value) in spec.columns.iter().zip(values) {
validate_storage(spec.name, column, value)?;
validate_integer_domain(spec.name, column, value)?;
validate_real_domain(spec.name, column, value)?;
}
match spec.validator {
RestoreRowValidator::OpaqueStorage => {}
@@ -1642,6 +1787,18 @@ fn assert_restore_policy_topology() -> Result<(), AppError> {
spec.name, column.name
)));
}
if column.storage == StorageKind::Real && column.real_domain == RealDomain::NotReal {
return Err(AppError::Database(format!(
"real restore column '{}.{}' has no finite domain",
spec.name, column.name
)));
}
if column.storage != StorageKind::Real && column.real_domain != RealDomain::NotReal {
return Err(AppError::Database(format!(
"non-real restore column '{}.{}' declares a real domain",
spec.name, column.name
)));
}
}
for parent in spec.parents {
if !seen.contains(parent) {
@@ -2527,9 +2684,9 @@ impl Database {
mod tests {
use super::{
assert_restore_policy_coverage, validate_canonical_behaviors, validate_regular_file,
validate_stage_rows, Database, RestoreFlavor, RestorePolicy, RestoreRowValidator,
StorageKind, UntrustedScratch, MAX_BINARY_RESTORE_BYTES, MAX_SCRATCH_BYTES,
MAX_SQL_IMPORT_BYTES, RESTORE_TABLE_SPECS, SCHEMA_VERSION,
validate_stage_rows, Database, MigrationRunContext, RestoreFlavor, RestorePolicy,
RestoreRowValidator, StorageKind, UntrustedScratch, MAX_BINARY_RESTORE_BYTES,
MAX_SCRATCH_BYTES, MAX_SQL_IMPORT_BYTES, RESTORE_TABLE_SPECS, SCHEMA_VERSION,
TEST_MAX_BACKUP_TRANSIENT_RETRIES, TEST_MAX_PAGE_COUNT, TEST_MAX_VM_STEPS,
};
use crate::error::AppError;
@@ -2666,11 +2823,16 @@ mod tests {
"input_token_semantics"
}
};
let real_domain = match column.real_domain {
super::RealDomain::NotReal => "not_real",
super::RealDomain::FiniteUnitInterval => "finite_unit_interval",
};
serde_json::json!([
column.name,
storage,
column.nullable,
integer_domain
integer_domain,
real_domain
])
}).collect::<Vec<_>>(),
"validator": validator,
@@ -2742,6 +2904,33 @@ mod tests {
source.snapshot_to_memory()
}
fn actual_v16_duplicate_endpoint_source() -> Result<Connection, AppError> {
let source = canonical_restore_source()?;
source.execute_batch(
"PRAGMA foreign_keys = OFF;
DROP TABLE provider_endpoints;
CREATE TABLE provider_endpoints (
id INTEGER PRIMARY KEY AUTOINCREMENT,
provider_id TEXT NOT NULL,
app_type TEXT NOT NULL,
url TEXT NOT NULL,
added_at INTEGER,
FOREIGN KEY (provider_id, app_type)
REFERENCES providers(id, app_type) ON DELETE CASCADE
);
INSERT INTO provider_endpoints
(id, provider_id, app_type, url, added_at)
VALUES
(1601, 'remote-provider', 'pi', 'https://duplicate-v16.invalid', 20),
(1602, 'remote-provider', 'pi', 'https://duplicate-v16.invalid', 10);
DROP TABLE pi_provider_projections;
DROP TABLE skill_deployments;
ALTER TABLE skills DROP COLUMN enabled_pi;
PRAGMA user_version = 16;",
)?;
Ok(source)
}
fn weak_ledger_source() -> Result<Connection, AppError> {
let source = canonical_restore_source()?;
source.execute_batch(
@@ -2997,6 +3186,9 @@ mod tests {
NonNegativeI32Domain,
Unsigned32Domain,
InputTokenSemanticsDomain,
NegativeCircuitThreshold,
OutOfRangeCircuitThreshold,
NonFiniteCircuitThreshold,
NegativeProviderMultiplier,
NegativeProviderMetaLimit,
InvalidProviderPricingSource,
@@ -3010,7 +3202,7 @@ mod tests {
}
impl InvalidRestoreCase {
const ALL: [Self; 22] = [
const ALL: [Self; 25] = [
Self::ProviderJson,
Self::McpTagsShape,
Self::ProfilePayloadShape,
@@ -3023,6 +3215,9 @@ mod tests {
Self::NonNegativeI32Domain,
Self::Unsigned32Domain,
Self::InputTokenSemanticsDomain,
Self::NegativeCircuitThreshold,
Self::OutOfRangeCircuitThreshold,
Self::NonFiniteCircuitThreshold,
Self::NegativeProviderMultiplier,
Self::NegativeProviderMetaLimit,
Self::InvalidProviderPricingSource,
@@ -3049,6 +3244,9 @@ mod tests {
Self::NonNegativeI32Domain => "non-negative-i32-domain",
Self::Unsigned32Domain => "unsigned-32-domain",
Self::InputTokenSemanticsDomain => "input-token-semantics-domain",
Self::NegativeCircuitThreshold => "negative-circuit-threshold",
Self::OutOfRangeCircuitThreshold => "out-of-range-circuit-threshold",
Self::NonFiniteCircuitThreshold => "non-finite-circuit-threshold",
Self::NegativeProviderMultiplier => "negative-provider-multiplier",
Self::NegativeProviderMetaLimit => "negative-provider-meta-limit",
Self::InvalidProviderPricingSource => "invalid-provider-pricing-source",
@@ -3169,6 +3367,27 @@ mod tests {
[],
)?;
}
InvalidRestoreCase::NegativeCircuitThreshold => {
source.execute(
"UPDATE proxy_config SET circuit_error_rate_threshold = -0.5
WHERE app_type = 'claude'",
[],
)?;
}
InvalidRestoreCase::OutOfRangeCircuitThreshold => {
source.execute(
"UPDATE proxy_config SET circuit_error_rate_threshold = 6.5
WHERE app_type = 'claude'",
[],
)?;
}
InvalidRestoreCase::NonFiniteCircuitThreshold => {
source.execute(
"UPDATE proxy_config SET circuit_error_rate_threshold = ?1
WHERE app_type = 'claude'",
[f64::INFINITY],
)?;
}
InvalidRestoreCase::NegativeProviderMultiplier => {
source.execute(
"UPDATE providers SET cost_multiplier = '-1'
@@ -3622,6 +3841,60 @@ mod tests {
Ok(())
}
#[test]
#[serial]
fn public_restore_entries_reject_v16_endpoint_repair_without_merging() -> Result<(), AppError> {
let test_home = tempfile::tempdir().map_err(|error| AppError::IoContext {
context: "create v16 duplicate endpoint restore home".to_string(),
source: error,
})?;
let _home_guard = TestHomeGuard::set(test_home.path());
for (entry_index, entry_point) in [RestoreEntryPoint::Sql, RestoreEntryPoint::Binary]
.into_iter()
.enumerate()
{
let source = actual_v16_duplicate_endpoint_source()?;
let target = Database::memory()?;
seed_live_restore_state(&target)?;
let before = logical_snapshot(&target)?;
let error = run_restore_entry(
&target,
&source,
entry_point,
&format!("duplicate-v16-{entry_index}.db"),
)
.expect_err("an untrusted v16 duplicate endpoint must not be repaired");
assert!(
matches!(error, AppError::InvalidInput(_)),
"migration repair rejection must remain structured: {error:?}"
);
assert!(
error.to_string().contains("migration repair is forbidden"),
"restore must fail at the untrusted migration boundary: {error}"
);
assert_eq!(
logical_snapshot(&target)?,
before,
"failed v16 repair changed live state via entry {entry_index}"
);
let duplicate_count: i64 = source.query_row(
"SELECT COUNT(*) FROM provider_endpoints
WHERE provider_id = 'remote-provider'
AND app_type = 'pi'
AND url = 'https://duplicate-v16.invalid'",
[],
|row| row.get(0),
)?;
assert_eq!(
duplicate_count, 2,
"untrusted migration must not merge the source rows"
);
}
Ok(())
}
#[test]
#[serial]
fn every_supported_user_version_has_a_public_migration_sentinel() -> Result<(), AppError> {
@@ -3991,6 +4264,41 @@ mod tests {
Ok(())
}
#[test]
#[serial]
fn repeated_live_database_init_is_rejected_until_primary_drops() -> Result<(), AppError> {
let test_home = tempfile::tempdir().map_err(|error| AppError::IoContext {
context: "create single-writer ownership home".to_string(),
source: error,
})?;
let _home_guard = TestHomeGuard::set(test_home.path());
let primary = Database::init()?;
let live_path = crate::config::get_app_config_dir().join("cc-switch.db");
assert_eq!(
Database::stored_user_version_exceeds_supported(&live_path)?,
None,
"the read-only version probe must coexist with the primary writer"
);
let duplicate = match Database::init() {
Ok(_) => {
return Err(AppError::Message(
"a second writable live Database unexpectedly initialized".to_string(),
));
}
Err(error) => error,
};
assert!(
matches!(duplicate, AppError::Conflict(_)),
"duplicate live init must be a structured conflict, got {duplicate:?}"
);
drop(primary);
let reopened = Database::init()?;
drop(reopened);
Ok(())
}
#[test]
#[serial]
fn restore_safety_backup_and_publish_hold_one_live_write_boundary() -> Result<(), AppError> {
@@ -4446,8 +4754,11 @@ mod tests {
let large_page_path = backup_dir.join("page-size-64k.db");
let large_page_source = Connection::open(&large_page_path)?;
large_page_source.execute_batch("PRAGMA page_size = 65536; VACUUM;")?;
Database::create_tables_on_conn(&large_page_source)?;
Database::apply_schema_migrations_on_conn(&large_page_source)?;
Database::create_tables_on_conn(&large_page_source, MigrationRunContext::LocalUpgrade)?;
Database::apply_schema_migrations_on_conn(
&large_page_source,
MigrationRunContext::LocalUpgrade,
)?;
large_page_source.execute(
"INSERT INTO providers (id, app_type, name, settings_config, meta)
VALUES ('large-page-source', 'pi', 'Large Page', '{}', '{}')",
src-tauri/src/database/backup_restore_certification.rs
+486 -8
View File
@@ -1,5 +1,8 @@
#![cfg(test)]
//! 前置工程 B:Canonical Restore 认证测试套件 v1(测试先行)
// 裁决方授权:认证文件允许豁免纯风格类 clippy lint(不含任何安全/正确性
// lint)。type_complexity 属阈值型风格检查,对测试专用契约文件无意义。
#![allow(clippy::type_complexity)]
//! 前置工程 B:Canonical Restore 认证测试套件 v2(测试先行)
//!
//! 规则与前置 A 完全一致(docs/pi-support-restructure-zh.md):实现方不得
//! 修改本文件;异议上报裁决;全绿是盲审前置条件而非充分条件。
@@ -29,13 +32,44 @@
//! 路径任何组件在检查后不得再经 symlink 重解析。**盲审重点核查项**。
//! O3 pre-B 认证通过后,裁决方重冻 infra 基线并将本套件红灯清单归零存档。
//!
//! ## 交接时的预期红绿
//! 应红 2:`certify_imported_sort_index_domain_is_enforced`(值域缺口:
//! RestoreColumnSpec 只查 storage/nullable,负数 sort_index 可发布,生产端
//! Option<usize> 读取即败——R4 finding)、
//! `certify_incremental_auto_vacuum_survives_restore`(canonical stage 未
//! 继承 INCREMENTAL,publish 后 live 文件退化为 NONE——R4 finding)。
//! 其余应绿。任何偏离(非清单红、应红变绿、编译失败)立即上报。
//! ## 首轮认证失败后的契约裁决(v2,2026-08-01)
//! 首轮组件盲审(止于 89961dff)确认两 High 两 Medium,均属契约真空,
//! 现裁决如下:
//! 裁决5【写入所有权边界,High】进程内对 live 数据库文件的**可写连接必须
//! 唯一**(AppState Database 主连接);对 live 路径的任何其他打开必须
//! 携带 SQLITE_OPEN_READ_ONLY(database/mod.rs 的版本预检连接是现存违
//! 例)。restore 在 Database mutex 内自 safety backup 前持锁至 publish
//! 完成——唯一可写连接使该 mutex 成为真实的全局写边界,堵死"第二连接
//! 在 safety backup 后提交、被 publish 覆盖"的窗口。结构红灯 R5;
//! **盲审重点**:全树审计 Connection::open* 调用点与其目标路径。
//! 裁决6【迁移语义分离,High】迁移链必须携带
//! `MigrationRunContext { LocalUpgrade, UntrustedRestore }`;一切修复/
//! 合并/去重步骤(含 v16→v17 的 endpoint 去重)在 UntrustedRestore 下
//! 必须整体 abort 并给结构化错误,LocalUpgrade 保留既有容错。结构红灯
//! R6(调用点级绑定:enum 形状/迁移入口签名/不可信入口传
//! UntrustedRestore/本地入口传 LocalUpgrade);行为义务 O4:用既有
//! user_version 哨兵机器构造 v16 重复 endpoint 输入,断言 import 整体
//! Err、零合并——**O4 交付并经裁决并入是首轮重认证的前置条件**,
//! 不得以结构绿替代语义覆盖。语义注记:该枚举表达"修复策略",
//! fresh/canonical bootstrap 映射 LocalUpgrade。
//! 裁决7【REAL 值域,Medium】浮点域列必须声明有限区间;
//! `circuit_error_rate_threshold` ∈ [0,1] 且有限。行为红灯 R7。
//! O2 扩展【Windows TOCTOU,Medium】Windows 二进制路径必须基于已验证目录
//! 句柄相对打开(参考 NtCreateFile RootDirectory 语义)或在打开后按句柄
//! 身份复核,且补文件级预检;**盲审重点**,Linux 宿主无法执行,逐行核查。
//! 裁决5 边界补记:重复 Database::init()、其他连接 ATTACH live 库均属
//! 进程内旁路,纳入全树审计(盲审);**进程外写者**(第二进程、外部
//! SQLite 客户端、直接替换文件)不受进程内边界约束,single-instance
//! 插件不是数据库锁——显式列为残余风险,由发布说明与盲审知悉。
//! 重认证预算:修复后 2 轮 fresh 双审;写入所有权或迁移语义同 invariant
//! 复发 → 停止并上报用户。
//!
//! ## 当前预期红绿(v2,基于 89961dff 之后的工作树)
//! 首轮实现已使 R1(sort_index 值域)与 R2(auto_vacuum)转绿。当前应红 3:
//! `certify_live_database_secondary_opens_are_read_only`(裁决5)、
//! `certify_untrusted_migration_context_is_threaded`(裁决6)、
//! `certify_imported_circuit_threshold_domain_is_enforced`(裁决7)。
//! 其余 7 项应绿。任何偏离(非清单红、应红变绿、编译失败)立即上报。
//!
//! ## 残余风险与收口
//! 沿用前置 A 的收口边界与残余清单(动态 SQL、trigger/view、Backup API、
@@ -607,3 +641,447 @@ fn certify_incremental_auto_vacuum_survives_restore() {
(canonical stage must inherit INCREMENTAL before its first page is written)"
);
}
// ---------------------------------------------------------------------------
// R5(应红,裁决5):live 库的次级连接必须只读——调用级检查
// ---------------------------------------------------------------------------
/// 每个函数收集连接打开调用:(调用名, 该调用实参中的 ident 集)。
#[derive(Default)]
struct OpenCallCollector {
calls: Vec<(String, BTreeSet<String>)>,
}
impl<'ast> Visit<'ast> for OpenCallCollector {
fn visit_expr_call(&mut self, node: &'ast syn::ExprCall) {
if let syn::Expr::Path(path) = node.func.as_ref() {
if let Some(segment) = path.path.segments.last() {
let name = segment.ident.to_string();
if matches!(name.as_str(), "open" | "open_with_flags" | "open_in_memory") {
let mut args = IdentProbe::default();
for arg in &node.args {
args.visit_expr(arg);
}
self.calls.push((name, args.found));
}
}
}
visit::visit_expr_call(self, node);
}
fn visit_expr_method_call(&mut self, node: &'ast syn::ExprMethodCall) {
let name = node.method.to_string();
if matches!(name.as_str(), "open" | "open_with_flags" | "open_in_memory") {
let mut args = IdentProbe::default();
for arg in &node.args {
args.visit_expr(arg);
}
self.calls.push((name, args.found));
}
visit::visit_expr_method_call(self, node);
}
}
/// 裁决5 的调用级规则。返回违规描述;空即合规。
fn live_open_violations(fn_name: &str, calls: &[(String, BTreeSet<String>)]) -> Vec<String> {
let mut violations = Vec::new();
let mut primary_opens = 0usize;
for (call, args) in calls {
match call.as_str() {
"open_in_memory" => {}
"open" => {
if fn_name == "init" {
primary_opens += 1;
} else {
violations.push(format!(
"{fn_name}: bare Connection::open is read-write; secondary \
connections must use open_with_flags + SQLITE_OPEN_READ_ONLY"
));
}
}
"open_with_flags" => {
if fn_name == "init" {
primary_opens += 1;
} else if !args.contains("SQLITE_OPEN_READ_ONLY") {
violations.push(format!(
"{fn_name}: open_with_flags without SQLITE_OPEN_READ_ONLY"
));
}
}
_ => {}
}
}
if fn_name == "init" && primary_opens > 1 {
violations.push(
"init: more than one primary open; the writable live connection must be unique"
.to_string(),
);
}
violations
}
fn collect_open_calls(items: &[Item], out: &mut Vec<(String, Vec<(String, BTreeSet<String>)>)>) {
for item in items {
match item {
Item::Fn(function) if !attrs_mark_test_only(&function.attrs) => {
let mut collector = OpenCallCollector::default();
collector.visit_block(&function.block);
out.push((function.sig.ident.to_string(), collector.calls));
}
Item::Impl(item_impl) if !attrs_mark_test_only(&item_impl.attrs) => {
for impl_item in &item_impl.items {
let ImplItem::Fn(function) = impl_item else {
continue;
};
if !attrs_mark_test_only(&function.attrs) {
let mut collector = OpenCallCollector::default();
collector.visit_block(&function.block);
out.push((function.sig.ident.to_string(), collector.calls));
}
}
}
Item::Mod(item_mod) if !attrs_mark_test_only(&item_mod.attrs) => {
if let Some((_, nested)) = &item_mod.content {
collect_open_calls(nested, out);
}
}
_ => {}
}
}
}
#[test]
fn certify_live_database_secondary_opens_are_read_only() {
// 负向 fixture 先证明检查器会响(修复后假绿是 R9 终审确认的 P0)。
let fixture = r#"
impl Database {
fn bad_bare(path: &Path) { let _ = Connection::open(path); }
fn bad_flags(path: &Path) {
let _ = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_WRITE);
}
fn good_flags(path: &Path) {
let _ = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY);
}
fn good_memory() { let _ = Connection::open_in_memory(); }
fn init(path: &Path) {
let _ = Connection::open(path);
let _ = Connection::open(path);
}
}
"#;
let parsed = syn::parse_file(fixture).expect("parse fixture");
let mut fixture_fns = Vec::new();
collect_open_calls(&parsed.items, &mut fixture_fns);
let violations_of = |name: &str, fns: &[(String, Vec<(String, BTreeSet<String>)>)]| {
fns.iter()
.filter(|(fn_name, _)| fn_name == name)
.flat_map(|(fn_name, calls)| live_open_violations(fn_name, calls))
.collect::<Vec<_>>()
};
assert!(!violations_of("bad_bare", &fixture_fns).is_empty());
assert!(!violations_of("bad_flags", &fixture_fns).is_empty());
assert!(violations_of("good_flags", &fixture_fns).is_empty());
assert!(violations_of("good_memory", &fixture_fns).is_empty());
assert!(
!violations_of("init", &fixture_fns).is_empty(),
"a second primary open inside init must be flagged"
);
// 真实检查:database/mod.rs 全部生产函数。
let source =
fs::read_to_string(source_root().join("database/mod.rs")).expect("read database/mod.rs");
let syntax = syn::parse_file(&source).expect("parse database/mod.rs");
let mut fns = Vec::new();
collect_open_calls(&syntax.items, &mut fns);
let mut violations = Vec::new();
for (name, calls) in &fns {
if name == "memory" {
continue; // 测试内存库工厂
}
violations.extend(live_open_violations(name, calls));
}
assert!(
violations.is_empty(),
"writable live-database connections must be unique to init: {violations:?}"
);
}
// ---------------------------------------------------------------------------
// R6(应红,裁决6):迁移语义分离——迁移调用实参级绑定
// ---------------------------------------------------------------------------
/// 收集对 `apply_schema_migrations_on_conn` 的每次调用及其实参 ident 集。
#[derive(Default)]
struct MigrationCallCollector {
calls: Vec<BTreeSet<String>>,
}
impl<'ast> Visit<'ast> for MigrationCallCollector {
fn visit_expr_call(&mut self, node: &'ast syn::ExprCall) {
if let syn::Expr::Path(path) = node.func.as_ref() {
if path
.path
.segments
.last()
.is_some_and(|segment| segment.ident == "apply_schema_migrations_on_conn")
{
let mut args = IdentProbe::default();
for arg in &node.args {
args.visit_expr(arg);
}
self.calls.push(args.found);
}
}
visit::visit_expr_call(self, node);
}
fn visit_expr_method_call(&mut self, node: &'ast syn::ExprMethodCall) {
if node.method == "apply_schema_migrations_on_conn" {
let mut args = IdentProbe::default();
for arg in &node.args {
args.visit_expr(arg);
}
self.calls.push(args.found);
}
visit::visit_expr_method_call(self, node);
}
}
#[test]
fn certify_untrusted_migration_context_is_threaded() {
// 函数体 ident 聚合会被迁移函数内部的 `match context` 分支假绿
// (R10 终审 P0),因此绑定到调用实参:
// 1) enum 精确形状;2) 确切入口 apply_schema_migrations_on_conn 的签名
// 携带 context;3) 不可信区域的迁移调用实参含 UntrustedRestore 且绝无
// LocalUpgrade;4) 本地区域的迁移调用实参含 LocalUpgrade 且绝无
// UntrustedRestore。v16 去重 fail-closed 的行为覆盖由义务 O4 交付
// (首轮重认证前置条件)。语义注记:枚举表达"修复策略",
// fresh/canonical bootstrap 映射 LocalUpgrade。
let schema = syn::parse_file(&schema_source()).expect("parse schema.rs");
let migration_source =
fs::read_to_string(source_root().join("database/migration.rs")).expect("read migration.rs");
let migration = syn::parse_file(&migration_source).expect("parse migration.rs");
// 1) enum 精确形状
fn find_enum<'a>(items: &'a [Item], name: &str) -> Option<&'a syn::ItemEnum> {
for item in items {
match item {
Item::Enum(item_enum) if item_enum.ident == name => return Some(item_enum),
Item::Mod(item_mod) => {
let nested = item_mod.content.as_ref().map(|(_, items)| items.as_slice());
if let Some(found) = nested.and_then(|items| find_enum(items, name)) {
return Some(found);
}
}
_ => {}
}
}
None
}
let context_enum = find_enum(&schema.items, "MigrationRunContext")
.or_else(|| find_enum(&migration.items, "MigrationRunContext"))
.expect("MigrationRunContext enum must exist in the migration chain");
let variants: Vec<String> = context_enum
.variants
.iter()
.map(|variant| variant.ident.to_string())
.collect();
assert_eq!(
variants,
vec!["LocalUpgrade".to_string(), "UntrustedRestore".to_string()],
"MigrationRunContext variants drifted from the adjudicated contract"
);
// 2) 确切入口签名携带 context
fn entry_takes_context(items: &[Item]) -> bool {
for item in items {
match item {
Item::Fn(function) if function.sig.ident == "apply_schema_migrations_on_conn" => {
let mut probe = IdentProbe::default();
for input in &function.sig.inputs {
if let syn::FnArg::Typed(typed) = input {
probe.visit_type(&typed.ty);
}
}
return probe.found.contains("MigrationRunContext");
}
Item::Impl(item_impl) => {
for impl_item in &item_impl.items {
let ImplItem::Fn(function) = impl_item else {
continue;
};
if function.sig.ident != "apply_schema_migrations_on_conn" {
continue;
}
let mut probe = IdentProbe::default();
for input in &function.sig.inputs {
if let syn::FnArg::Typed(typed) = input {
probe.visit_type(&typed.ty);
}
}
return probe.found.contains("MigrationRunContext");
}
}
Item::Mod(item_mod) => {
if let Some((_, nested)) = &item_mod.content {
if entry_takes_context(nested) {
return true;
}
}
}
_ => {}
}
}
false
}
assert!(
entry_takes_context(&schema.items) || entry_takes_context(&migration.items),
"apply_schema_migrations_on_conn must take MigrationRunContext in its signature"
);
// 3) 不可信区域:UntrustedScratch impl 内的迁移调用
let backup = syn::parse_file(&backup_source()).expect("parse backup.rs");
let mut untrusted_calls = MigrationCallCollector::default();
for item in &backup.items {
let Item::Impl(item_impl) = item else {
continue;
};
let matches_type = matches!(
item_impl.self_ty.as_ref(),
syn::Type::Path(type_path)
if type_path
.path
.segments
.last()
.is_some_and(|segment| segment.ident == "UntrustedScratch")
);
if !matches_type {
continue;
}
for impl_item in &item_impl.items {
if let ImplItem::Fn(function) = impl_item {
untrusted_calls.visit_block(&function.block);
}
}
}
assert!(
untrusted_calls
.calls
.iter()
.any(|args| args.contains("UntrustedRestore")),
"the untrusted scratch pipeline must call migrations with UntrustedRestore"
);
assert!(
untrusted_calls
.calls
.iter()
.all(|args| !args.contains("LocalUpgrade")),
"the untrusted pipeline must never request tolerant LocalUpgrade repairs"
);
// 4) 本地区域:database/mod.rs ∪ schema.rs ∪ migration.rs 生产函数内的
// 迁移调用(排除 backup.rs;调用实参而非函数体,match 分支无法假绿)
let mod_source =
fs::read_to_string(source_root().join("database/mod.rs")).expect("read database/mod.rs");
let mod_parsed = syn::parse_file(&mod_source).expect("parse database/mod.rs");
let mut local_calls = MigrationCallCollector::default();
fn visit_production_fns(items: &[Item], collector: &mut MigrationCallCollector) {
for item in items {
match item {
Item::Fn(function) if !attrs_mark_test_only(&function.attrs) => {
collector.visit_block(&function.block);
}
Item::Impl(item_impl) if !attrs_mark_test_only(&item_impl.attrs) => {
for impl_item in &item_impl.items {
if let ImplItem::Fn(function) = impl_item {
if !attrs_mark_test_only(&function.attrs) {
collector.visit_block(&function.block);
}
}
}
}
Item::Mod(item_mod) if !attrs_mark_test_only(&item_mod.attrs) => {
if let Some((_, nested)) = &item_mod.content {
visit_production_fns(nested, collector);
}
}
_ => {}
}
}
}
visit_production_fns(&mod_parsed.items, &mut local_calls);
visit_production_fns(&schema.items, &mut local_calls);
visit_production_fns(&migration.items, &mut local_calls);
assert!(
local_calls
.calls
.iter()
.any(|args| args.contains("LocalUpgrade")),
"a production local-upgrade call site must pass LocalUpgrade"
);
assert!(
local_calls
.calls
.iter()
.all(|args| !args.contains("UntrustedRestore")),
"local upgrade paths must never run under UntrustedRestore semantics"
);
}
// ---------------------------------------------------------------------------
// R7(应红,裁决7):REAL 值域——熔断阈值必须限于有限 [0,1](定向篡改)
// ---------------------------------------------------------------------------
#[test]
fn certify_imported_circuit_threshold_domain_is_enforced() {
let database = Database::memory().expect("memory db");
let exported = database.export_sql_string().expect("export");
// 定向替换:只动 proxy_config 表插入行里的阈值,不碰 DDL 默认值与
// model_pricing 的 0.60/0.65 子串(R9 终审确认全局替换会误伤)。
for hostile_value in ["6.5", "-0.5", "1e999"] {
let mut replaced = 0usize;
let hostile: String = exported
.lines()
.map(|line| {
if replaced == 0
&& line.contains("proxy_config")
&& !line.to_ascii_uppercase().contains("CREATE TABLE")
&& line.contains(", 0.6,")
{
replaced += 1;
let mut tampered = line.replacen(", 0.6,", &format!(", {hostile_value},"), 1);
tampered.push('\n');
tampered
} else {
let mut kept = line.to_string();
kept.push('\n');
kept
}
})
.collect();
assert_eq!(
replaced, 1,
"exactly one proxy_config insert row must carry the 0.6 sentinel \
(export format drifted; escalate for adjudication)"
);
let target = Database::memory().expect("target db");
target
.import_sql_string(&hostile)
.expect_err("circuit_error_rate_threshold outside finite [0,1] must abort the import");
let conn = target.conn.lock().expect("lock target");
let out_of_domain: i64 = conn
.query_row(
"SELECT COUNT(*) FROM proxy_config
WHERE circuit_error_rate_threshold < 0.0
OR circuit_error_rate_threshold > 1.0",
[],
|row| row.get(0),
)
.expect("count out-of-domain thresholds");
assert_eq!(
out_of_domain, 0,
"aborted import must leave no out-of-domain breaker thresholds behind"
);
}
}
src-tauri/src/database/migration.rs
+3 -2
View File
@@ -2,6 +2,7 @@
//!
//! 将旧版 config.json (MultiAppConfig) 数据迁移到 SQLite 数据库。
use super::schema::MigrationRunContext;
use super::{lock_conn, to_json_string, Database};
use crate::app_config::MultiAppConfig;
use crate::error::AppError;
@@ -28,8 +29,8 @@ impl Database {
pub fn migrate_from_json_dry_run(config: &MultiAppConfig) -> Result<(), AppError> {
let mut conn =
Connection::open_in_memory().map_err(|e| AppError::Database(e.to_string()))?;
Self::create_tables_on_conn(&conn)?;
Self::apply_schema_migrations_on_conn(&conn)?;
Self::create_tables_on_conn(&conn, MigrationRunContext::LocalUpgrade)?;
Self::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)?;
let tx = conn
.transaction()
src-tauri/src/database/mod.rs
+82 -3
View File
@@ -50,9 +50,11 @@ pub use dao::Profile;
use crate::config::get_app_config_dir;
use crate::error::AppError;
use rusqlite::{hooks::Action, Connection};
use rusqlite::{hooks::Action, Connection, OpenFlags};
use serde::Serialize;
use std::sync::Mutex;
use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::{Mutex, OnceLock};
// DAO 方法通过 impl Database 提供，无需额外导出
@@ -84,6 +86,79 @@ pub(crate) use lock_conn;
/// rusqlite::Connection 本身不是 Sync 的，因此需要这层包装。
pub struct Database {
pub(crate) conn: Mutex<Connection>,
// Keep this field after `conn`: field drop order closes the writable
// connection before releasing its process-local ownership lease.
_live_write_lease: Option<LiveDatabaseWriteLease>,
}
static LIVE_DATABASE_WRITERS: OnceLock<Mutex<HashSet<PathBuf>>> = OnceLock::new();
struct LiveDatabaseWriteLease {
identity: PathBuf,
}
impl LiveDatabaseWriteLease {
fn acquire(path: &Path) -> Result<Self, AppError> {
let identity = live_database_identity(path)?;
let mut writers = LIVE_DATABASE_WRITERS
.get_or_init(|| Mutex::new(HashSet::new()))
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if !writers.insert(identity.clone()) {
return Err(AppError::Conflict(format!(
"live database already has a writable connection: {}",
identity.display()
)));
}
Ok(Self { identity })
}
}
impl Drop for LiveDatabaseWriteLease {
fn drop(&mut self) {
LIVE_DATABASE_WRITERS
.get_or_init(|| Mutex::new(HashSet::new()))
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.remove(&self.identity);
}
}
fn live_database_identity(path: &Path) -> Result<PathBuf, AppError> {
match std::fs::canonicalize(path) {
Ok(identity) => Ok(identity),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
match std::fs::symlink_metadata(path) {
Ok(_) => {
// Do not treat an existing broken symlink as a new
// database path: SQLite could follow it to an alias
// outside the lease identity selected below.
return Err(AppError::io(path, error));
}
Err(metadata_error) if metadata_error.kind() == std::io::ErrorKind::NotFound => {}
Err(metadata_error) => return Err(AppError::io(path, metadata_error)),
}
// A new database has no file identity yet. Canonicalizing the
// already-created parent still collapses relative and symlinked
// directory aliases before the single writable open occurs.
let parent = path.parent().ok_or_else(|| {
AppError::Config(format!(
"live database path has no parent: {}",
path.display()
))
})?;
let filename = path.file_name().ok_or_else(|| {
AppError::Config(format!(
"live database path has no filename: {}",
path.display()
))
})?;
let parent =
std::fs::canonicalize(parent).map_err(|error| AppError::io(parent, error))?;
Ok(parent.join(filename))
}
Err(error) => Err(AppError::io(path, error)),
}
}
fn register_db_change_hook(conn: &Connection) {
@@ -111,6 +186,7 @@ impl Database {
std::fs::create_dir_all(parent).map_err(|e| AppError::io(parent, e))?;
}
let live_write_lease = LiveDatabaseWriteLease::acquire(&db_path)?;
let conn = Connection::open(&db_path).map_err(|e| AppError::Database(e.to_string()))?;
// 启用外键约束
@@ -126,6 +202,7 @@ impl Database {
let db = Self {
conn: Mutex::new(conn),
_live_write_lease: Some(live_write_lease),
};
db.create_tables()?;
@@ -182,7 +259,8 @@ impl Database {
if !db_path.exists() {
return Ok(None);
}
let conn = Connection::open(db_path).map_err(|e| AppError::Database(e.to_string()))?;
let conn = Connection::open_with_flags(db_path, OpenFlags::SQLITE_OPEN_READ_ONLY)
.map_err(|e| AppError::Database(e.to_string()))?;
let version = Self::get_user_version(&conn)?;
Ok((version > SCHEMA_VERSION).then_some(version))
}
@@ -200,6 +278,7 @@ impl Database {
let db = Self {
conn: Mutex::new(conn),
_live_write_lease: None,
};
db.create_tables()?;
// Keep the test database structurally identical to a fresh production
src-tauri/src/database/schema.rs
+129 -93
View File
@@ -4,7 +4,7 @@
use super::{lock_conn, Database, SCHEMA_VERSION};
use crate::error::AppError;
use rusqlite::{params, Connection};
use rusqlite::{params, Connection, OptionalExtension};
use serde::Serialize;
use tempfile::NamedTempFile;
@@ -26,6 +26,14 @@ impl CanonicalStage {
}
}
/// Selects whether schema migrations may repair locally trusted historical
/// data or must fail closed on an input that would require reconciliation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum MigrationRunContext {
LocalUpgrade,
UntrustedRestore,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum CanonicalRestoreClass {
@@ -525,9 +533,9 @@ impl Database {
// Starting from version zero exercises the normal migration chain and
// yields the exact same current objects as a fresh production database.
Self::create_tables_on_conn(&connection)?;
Self::create_tables_on_conn(&connection, MigrationRunContext::LocalUpgrade)?;
Self::set_user_version(&connection, 0)?;
Self::apply_schema_migrations_on_conn(&connection)?;
Self::apply_schema_migrations_on_conn(&connection, MigrationRunContext::LocalUpgrade)?;
if Self::get_user_version(&connection)? != SCHEMA_VERSION {
return Err(AppError::Database(
"canonical stage factory did not reach the current schema version".to_string(),
@@ -581,11 +589,14 @@ impl Database {
/// 创建所有数据库表
pub(crate) fn create_tables(&self) -> Result<(), AppError> {
let conn = lock_conn!(self.conn);
Self::create_tables_on_conn(&conn)
Self::create_tables_on_conn(&conn, MigrationRunContext::LocalUpgrade)
}
/// 在指定连接上创建表（供迁移和测试使用）
pub(crate) fn create_tables_on_conn(conn: &Connection) -> Result<(), AppError> {
pub(crate) fn create_tables_on_conn(
conn: &Connection,
context: MigrationRunContext,
) -> Result<(), AppError> {
// 1. Providers 表
Self::create_canonical_table_on_conn(conn, "providers", true)?;
@@ -685,7 +696,9 @@ impl Database {
// 兼容旧数据库：
// - 老版本 proxy_config 是单例表（没有 app_type 列），此时不能执行三行 seed insert；
// - 旧表会在 apply_schema_migrations() 中迁移为三行结构后再插入。
if Self::has_column(conn, "proxy_config", "app_type")? {
if context == MigrationRunContext::LocalUpgrade
&& Self::has_column(conn, "proxy_config", "app_type")?
{
conn.execute(
"INSERT OR IGNORE INTO proxy_config (app_type, max_retries,
streaming_first_byte_timeout, streaming_idle_timeout, non_streaming_timeout,
@@ -865,96 +878,92 @@ impl Database {
)
.map_err(|e| AppError::Database(e.to_string()))?;
// 修复跑过未发布开发版的库：current 标记曾是全局 key，现按应用分组
// （随 v12 定稿为 current_profile_id_<scope>，不单独 bump 版本）
if conn
.execute(
"INSERT OR REPLACE INTO settings (key, value)
SELECT 'current_profile_id_claude', value FROM settings
WHERE key = 'current_profile_id'",
if context == MigrationRunContext::LocalUpgrade {
// These compatibility repairs are intentionally local-only. An
// untrusted restore must be accepted by its declared migration
// version and canonical validation, never silently normalized by
// startup repair code.
if conn
.execute(
"INSERT OR REPLACE INTO settings (key, value)
SELECT 'current_profile_id_claude', value FROM settings
WHERE key = 'current_profile_id'",
[],
)
.is_ok()
{
let _ = conn.execute("DELETE FROM settings WHERE key = 'current_profile_id'", []);
}
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN live_takeover_active INTEGER NOT NULL DEFAULT 0",
[],
)
.is_ok()
{
let _ = conn.execute("DELETE FROM settings WHERE key = 'current_profile_id'", []);
);
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN proxy_enabled INTEGER NOT NULL DEFAULT 0",
[],
);
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN listen_address TEXT NOT NULL DEFAULT '127.0.0.1'",
[],
);
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN listen_port INTEGER NOT NULL DEFAULT 15721",
[],
);
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN enable_logging INTEGER NOT NULL DEFAULT 1",
[],
);
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN streaming_first_byte_timeout INTEGER NOT NULL DEFAULT 60",
[],
);
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN streaming_idle_timeout INTEGER NOT NULL DEFAULT 120",
[],
);
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN non_streaming_timeout INTEGER NOT NULL DEFAULT 600",
[],
);
if Self::table_exists(conn, "proxy_config")?
&& !Self::has_column(conn, "proxy_config", "app_type")?
{
Self::migrate_proxy_config_to_per_app(conn)?;
}
Self::add_column_if_missing(
conn,
"providers",
"in_failover_queue",
"BOOLEAN NOT NULL DEFAULT 0",
)?;
let _ = conn.execute("DROP INDEX IF EXISTS idx_failover_queue_order", []);
let _ = conn.execute("DROP TABLE IF EXISTS failover_queue", []);
let _ = conn.execute(
"CREATE INDEX IF NOT EXISTS idx_providers_failover
ON providers(app_type, in_failover_queue, sort_index)",
[],
);
}
// 尝试添加 live_takeover_active 列到 proxy_config 表
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN live_takeover_active INTEGER NOT NULL DEFAULT 0",
[],
);
// 尝试添加基础配置列到 proxy_config 表（兼容 v3.9.0-2 升级）
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN proxy_enabled INTEGER NOT NULL DEFAULT 0",
[],
);
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN listen_address TEXT NOT NULL DEFAULT '127.0.0.1'",
[],
);
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN listen_port INTEGER NOT NULL DEFAULT 15721",
[],
);
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN enable_logging INTEGER NOT NULL DEFAULT 1",
[],
);
// 尝试添加超时配置列到 proxy_config 表
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN streaming_first_byte_timeout INTEGER NOT NULL DEFAULT 60",
[],
);
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN streaming_idle_timeout INTEGER NOT NULL DEFAULT 120",
[],
);
let _ = conn.execute(
"ALTER TABLE proxy_config ADD COLUMN non_streaming_timeout INTEGER NOT NULL DEFAULT 600",
[],
);
// 兼容：若旧版 proxy_config 仍为单例结构（无 app_type），则在启动时直接转换为三行结构
// 说明：user_version=2 时不会再触发 v1->v2 迁移，但新代码查询依赖 app_type 列。
if Self::table_exists(conn, "proxy_config")?
&& !Self::has_column(conn, "proxy_config", "app_type")?
{
Self::migrate_proxy_config_to_per_app(conn)?;
}
// 确保 in_failover_queue 列存在（对于已存在的 v2 数据库）
Self::add_column_if_missing(
conn,
"providers",
"in_failover_queue",
"BOOLEAN NOT NULL DEFAULT 0",
)?;
// 删除旧的 failover_queue 表（如果存在）
let _ = conn.execute("DROP INDEX IF EXISTS idx_failover_queue_order", []);
let _ = conn.execute("DROP TABLE IF EXISTS failover_queue", []);
// 为故障转移队列创建索引（基于 providers 表）
let _ = conn.execute(
"CREATE INDEX IF NOT EXISTS idx_providers_failover
ON providers(app_type, in_failover_queue, sort_index)",
[],
);
Ok(())
}
/// 应用 Schema 迁移
pub(crate) fn apply_schema_migrations(&self) -> Result<(), AppError> {
let conn = lock_conn!(self.conn);
Self::apply_schema_migrations_on_conn(&conn)
Self::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)
}
/// 在指定连接上应用 Schema 迁移
pub(crate) fn apply_schema_migrations_on_conn(conn: &Connection) -> Result<(), AppError> {
pub(crate) fn apply_schema_migrations_on_conn(
conn: &Connection,
context: MigrationRunContext,
) -> Result<(), AppError> {
conn.execute("SAVEPOINT schema_migration;", [])
.map_err(|e| AppError::Database(format!("开启迁移 savepoint 失败: {e}")))?;
@@ -1057,7 +1066,7 @@ impl Database {
log::info!(
"迁移数据库从 v16 到 v17（添加 Pi aggregate 与设备本地 ledger）"
);
Self::migrate_v16_to_v17(conn)?;
Self::migrate_v16_to_v17(conn, context)?;
Self::set_user_version(conn, 17)?;
}
_ => {
@@ -2075,8 +2084,35 @@ impl Database {
/// v16 -> v17: add the Pi desired bit, lossless endpoint metadata, and
/// device-local ownership ledgers. No ownership is inferred during
/// migration; both ledgers intentionally start empty.
fn migrate_v16_to_v17(conn: &Connection) -> Result<(), AppError> {
fn migrate_v16_to_v17(conn: &Connection, context: MigrationRunContext) -> Result<(), AppError> {
if Self::table_exists(conn, "provider_endpoints")? {
if context == MigrationRunContext::UntrustedRestore {
let duplicate = conn
.query_row(
"SELECT provider_id, app_type, url, COUNT(*)
FROM provider_endpoints
GROUP BY provider_id, app_type, url
HAVING COUNT(*) > 1
LIMIT 1",
[],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, i64>(3)?,
))
},
)
.optional()
.map_err(|error| AppError::Database(error.to_string()))?;
if let Some((provider_id, app_type, url, count)) = duplicate {
return Err(AppError::InvalidInput(format!(
"untrusted v16 restore contains {count} duplicate endpoint rows \
for ({provider_id}, {app_type}, {url}); migration repair is forbidden"
)));
}
}
Self::add_column_if_missing(conn, "provider_endpoints", "last_used", "INTEGER")?;
// Older builds allowed duplicate rows for one logical endpoint.
// Merge their timestamps before rebuilding from the canonical
@@ -3650,7 +3686,7 @@ mod tests {
)?;
Database::set_user_version(&conn, 12)?;
Database::apply_schema_migrations_on_conn(&conn)?;
Database::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)?;
assert_eq!(Database::get_user_version(&conn)?, SCHEMA_VERSION);
assert!(Database::has_column(
@@ -3677,7 +3713,7 @@ mod tests {
#[test]
fn migrate_v13_to_v14_adds_grokbuild_proxy_row_and_preserves_values() -> Result<(), AppError> {
let conn = Connection::open_in_memory()?;
Database::create_tables_on_conn(&conn)?;
Database::create_tables_on_conn(&conn, MigrationRunContext::LocalUpgrade)?;
conn.execute("DELETE FROM proxy_config WHERE app_type = 'grokbuild'", [])?;
conn.execute(
"UPDATE proxy_config SET enabled = 1, max_retries = 9 WHERE app_type = 'codex'",
@@ -3685,7 +3721,7 @@ mod tests {
)?;
Database::set_user_version(&conn, 13)?;
Database::apply_schema_migrations_on_conn(&conn)?;
Database::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)?;
assert_eq!(Database::get_user_version(&conn)?, SCHEMA_VERSION);
let grok_rows: i64 = conn.query_row(
@@ -3727,7 +3763,7 @@ mod tests {
)?;
Database::set_user_version(&conn, 14)?;
Database::apply_schema_migrations_on_conn(&conn)?;
Database::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)?;
assert_eq!(Database::get_user_version(&conn)?, SCHEMA_VERSION);
assert!(Database::has_column(
@@ -3755,7 +3791,7 @@ mod tests {
#[test]
fn migrate_v15_to_v16_resets_only_codex_session_usage() -> Result<(), AppError> {
let conn = Connection::open_in_memory()?;
Database::create_tables_on_conn(&conn)?;
Database::create_tables_on_conn(&conn, MigrationRunContext::LocalUpgrade)?;
conn.execute_batch(
"INSERT INTO proxy_request_logs (
request_id, provider_id, app_type, model, input_tokens,
@@ -3776,7 +3812,7 @@ mod tests {
)?;
Database::set_user_version(&conn, 15)?;
Database::apply_schema_migrations_on_conn(&conn)?;
Database::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)?;
assert_eq!(Database::get_user_version(&conn)?, SCHEMA_VERSION);
let counts: (i64, i64, i64, i64) = conn.query_row(
@@ -3822,7 +3858,7 @@ mod tests {
)?;
Database::set_user_version(&conn, 16)?;
Database::apply_schema_migrations_on_conn(&conn)?;
Database::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)?;
assert_eq!(Database::get_user_version(&conn)?, SCHEMA_VERSION);
assert!(Database::has_column(
src-tauri/src/database/tests.rs
+31 -16
View File
@@ -4,6 +4,7 @@
//!
//! 包含 Schema 迁移和基本功能的测试。
use super::schema::MigrationRunContext;
use super::*;
use crate::app_config::MultiAppConfig;
use crate::provider::{Provider, ProviderManager};
@@ -189,13 +190,15 @@ fn existing_skill_repo_selection_is_not_supplemented() {
fn schema_migration_sets_user_version_when_missing() {
let conn = Connection::open_in_memory().expect("open memory db");
Database::create_tables_on_conn(&conn).expect("create tables");
Database::create_tables_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("create tables");
assert_eq!(
Database::get_user_version(&conn).expect("read version before"),
0
);
Database::apply_schema_migrations_on_conn(&conn).expect("apply migration");
Database::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("apply migration");
assert_eq!(
Database::get_user_version(&conn).expect("read version after"),
@@ -206,11 +209,12 @@ fn schema_migration_sets_user_version_when_missing() {
#[test]
fn schema_migration_rejects_future_version() {
let conn = Connection::open_in_memory().expect("open memory db");
Database::create_tables_on_conn(&conn).expect("create tables");
Database::create_tables_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("create tables");
Database::set_user_version(&conn, SCHEMA_VERSION + 1).expect("set future version");
let err =
Database::apply_schema_migrations_on_conn(&conn).expect_err("should reject higher version");
let err = Database::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect_err("should reject higher version");
assert!(
err.to_string().contains("数据库版本过新"),
"unexpected error: {err}"
@@ -225,7 +229,8 @@ fn schema_migration_adds_missing_columns_for_providers() {
conn.execute_batch(LEGACY_SCHEMA_SQL)
.expect("seed old schema");
Database::apply_schema_migrations_on_conn(&conn).expect("apply migrations");
Database::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("apply migrations");
// 验证关键新增列已补齐
for (table, column) in [
@@ -264,7 +269,8 @@ fn schema_migration_aligns_column_defaults_and_types() {
conn.execute_batch(LEGACY_SCHEMA_SQL)
.expect("seed old schema");
Database::apply_schema_migrations_on_conn(&conn).expect("apply migrations");
Database::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("apply migrations");
let is_current = get_column_info(&conn, "providers", "is_current");
assert_eq!(is_current.r#type, "BOOLEAN");
@@ -305,7 +311,8 @@ fn schema_migration_aligns_column_defaults_and_types() {
#[test]
fn schema_create_tables_include_pricing_model_columns() {
let conn = Connection::open_in_memory().expect("open memory db");
Database::create_tables_on_conn(&conn).expect("create tables");
Database::create_tables_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("create tables");
let multiplier = get_column_info(&conn, "proxy_config", "default_cost_multiplier");
assert_eq!(multiplier.r#type, "TEXT");
@@ -354,7 +361,8 @@ fn schema_migration_v4_adds_pricing_model_columns() {
.expect("seed v4 schema");
Database::set_user_version(&conn, 4).expect("set user_version=4");
Database::apply_schema_migrations_on_conn(&conn).expect("apply migrations");
Database::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("apply migrations");
let multiplier = get_column_info(&conn, "proxy_config", "default_cost_multiplier");
assert_eq!(multiplier.r#type, "TEXT");
@@ -416,7 +424,8 @@ fn migration_v10_to_v11_rebuilds_rollups_with_request_model_dimension() {
.expect("seed v10 rollup table");
Database::set_user_version(&conn, 10).expect("set user_version=10");
Database::apply_schema_migrations_on_conn(&conn).expect("apply migrations");
Database::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("apply migrations");
// 新列存在且 NOT NULL DEFAULT ''
let request_model = get_column_info(&conn, "usage_daily_rollups", "request_model");
@@ -484,7 +493,8 @@ fn schema_create_tables_repairs_dev_global_profile_marker() {
.expect("seed dev v12 shape");
Database::set_user_version(&conn, 12).expect("set user_version=12");
Database::create_tables_on_conn(&conn).expect("create tables should repair marker");
Database::create_tables_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("create tables should repair marker");
// 全局 current 标记改名为 claude 组标记，旧 key 删除
let claude_marker: String = conn
@@ -505,7 +515,8 @@ fn schema_create_tables_repairs_dev_global_profile_marker() {
assert_eq!(old_marker, 0);
// 修复必须幂等：再跑一遍不应破坏已迁移的标记
Database::create_tables_on_conn(&conn).expect("repair is idempotent");
Database::create_tables_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("repair is idempotent");
let claude_marker: String = conn
.query_row(
"SELECT value FROM settings WHERE key = 'current_profile_id_claude'",
@@ -541,7 +552,8 @@ fn schema_create_tables_repairs_legacy_proxy_config_singleton_to_per_app() {
)
.expect("seed legacy proxy_config");
Database::create_tables_on_conn(&conn).expect("create tables should repair proxy_config");
Database::create_tables_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("create tables should repair proxy_config");
assert!(
Database::has_column(&conn, "proxy_config", "app_type").expect("check app_type"),
@@ -605,8 +617,10 @@ fn migration_from_v3_8_schema_v1_to_current_schema_v3() {
.expect("seed legacy skill");
// 按应用启动流程：先 create_tables（补齐新增表），再 apply_schema_migrations（按 user_version 迁移）
Database::create_tables_on_conn(&conn).expect("create tables");
Database::apply_schema_migrations_on_conn(&conn).expect("apply migrations");
Database::create_tables_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("create tables");
Database::apply_schema_migrations_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("apply migrations");
assert_eq!(
Database::get_user_version(&conn).expect("user_version after migration"),
@@ -901,7 +915,8 @@ fn ensure_incremental_auto_vacuum_rebuilds_existing_file_db() {
let conn = Connection::open(&path).expect("open temp db");
conn.execute("PRAGMA auto_vacuum = NONE;", [])
.expect("set none auto_vacuum");
Database::create_tables_on_conn(&conn).expect("create tables");
Database::create_tables_on_conn(&conn, MigrationRunContext::LocalUpgrade)
.expect("create tables");
assert_eq!(
Database::get_auto_vacuum_mode(&conn).expect("auto_vacuum before rebuild"),