diff --git a/CHANGELOG.md b/CHANGELOG.md index 57aa9d673..9e94400c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [v0.5.1] - 2026-08-20 + +### Changed + +- **存储策略凭据兼容层完成收口** — 移除 0.5.x 启动阶段的 legacy credential importer、connector legacy import hook、OneDrive 旧 OAuth 转换、deprecated credential entities / repositories，以及 `database-migrate` 的旧凭据复制与导入路径。当前运行时只消费 `connector_id`、typed `storage_config` 和 `storage_policy_connector_credentials`。 +- **存储策略最终 schema migration** — 新增 `m20260820_000001_remove_storage_policy_legacy`。迁移会在任何 DDL 前检查旧凭据表和旧静态凭据列；发现未完成 0.5.x 转换时硬失败并保留原 schema / 数据，检查通过后删除两个旧凭据表、旧 `storage_policies` 列、索引和远端节点外键。 +- **跨数据库迁移边界** — `database-migrate` 只复制当前 policy envelope 与 connector credential；带有未迁移 legacy credential 的 source database 会在复制前拒绝，空的历史 legacy stores 不再进入目标库。 + +### Fixed + +- **迁移幂等与回滚边界** — 覆盖 SQLite、PostgreSQL 和 MySQL 的旧列 / 索引 / foreign-key 清理路径，保持 SQLite foreign-key 状态并验证引用 `storage_policies` 的现有数据不丢失。 +- **schema drift 与历史测试边界** — 区分历史 migration、0.5.x compatibility schema 和最终 schema，补充未迁移凭据硬失败、空旧表清理、最终列集合和重复执行测试。 ### Added - **内置登录方式控制** — 新增可热更新的密码登录开关，并继续与 Passkey 开关独立组合；关闭密码登录会同时关闭公开注册、激活重发、密码邀请接受、密码重置和外部身份密码绑定，未完成的密码第一因子 MFA flow 会在完成时重新检查策略，外部认证和 Passkey 登录不再被遗留的强制改密标记阻塞。后端仅在存在已启用外部认证 provider 时允许同时关闭密码与 Passkey，并阻止禁用或删除最后一个外部 provider，避免保存后失去全部登录入口。 diff --git a/crates/aster_drive_migration/src/lib.rs b/crates/aster_drive_migration/src/lib.rs index c8d2ce2b9..8a1475988 100644 --- a/crates/aster_drive_migration/src/lib.rs +++ b/crates/aster_drive_migration/src/lib.rs @@ -69,6 +69,7 @@ mod m20260810_000001_folder_tree_operation_members; mod m20260813_000001_canonical_file_revision_ledger; mod m20260815_000001_virtual_empty_file_blobs; mod m20260817_000001_add_remote_binding_control_state; +mod m20260820_000001_remove_storage_policy_legacy; pub const BASELINE_MIGRATION_NAME: &str = "m20260512_000001_baseline_schema"; const MIGRATION_TABLE: &str = "seaql_migrations"; @@ -211,6 +212,7 @@ impl MigratorTrait for CurrentMigrator { Box::new( m20260817_000001_add_remote_binding_control_state::Migration, ), + Box::new(m20260820_000001_remove_storage_policy_legacy::Migration), ] } } diff --git a/crates/aster_drive_migration/src/m20260820_000001_remove_storage_policy_legacy.rs b/crates/aster_drive_migration/src/m20260820_000001_remove_storage_policy_legacy.rs new file mode 100644 index 000000000..f25b3ecc2 --- /dev/null +++ b/crates/aster_drive_migration/src/m20260820_000001_remove_storage_policy_legacy.rs @@ -0,0 +1,287 @@ +//! Finalize the storage-policy schema introduced in AsterDrive 0.5.1. +//! +//! The 0.5.x application migration must have imported every legacy secret +//! before this migration is allowed to remove the old stores. Checks are +//! deliberately completed before any DDL so a rejected upgrade leaves the +//! schema and data untouched. + +use sea_orm_migration::prelude::*; +use sea_orm_migration::sea_orm::{ConnectionTrait, DbBackend, Statement}; + +const LEGACY_POLICY_COLUMNS: &[&str] = &[ + "driver_type", + "endpoint", + "bucket", + "access_key", + "secret_key", + "base_path", + "remote_node_id", + "remote_storage_target_key", + "options", +]; + +const LEGACY_TABLES: &[&str] = &[ + "storage_policy_credentials", + "storage_connector_application_configs", +]; + +const LEGACY_INDEXES: &[&str] = &[ + "idx_storage_policies_remote_target", + "idx_storage_policies_remote_node_id", +]; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let connection = manager.get_connection(); + ensure_no_unmigrated_credentials(manager).await?; + drop_legacy_tables(manager).await?; + drop_legacy_policy_constraints(manager).await?; + drop_legacy_policy_columns(manager).await?; + // Keep the final schema explicit: if a future historical branch adds + // one of these columns again, this migration still removes it. + let remaining = existing_legacy_columns(manager).await?; + if !remaining.is_empty() { + return Err(DbErr::Migration(format!( + "storage policy legacy columns remain after cleanup: {}", + remaining.join(", ") + ))); + } + // A backend may keep foreign-key metadata after dropping a column; + // force SQLite to validate the resulting schema before recording the + // migration. Other backends validate this during ALTER TABLE. + if connection.get_database_backend() == DbBackend::Sqlite { + let violations = connection + .query_all_raw(Statement::from_string( + DbBackend::Sqlite, + "PRAGMA foreign_key_check", + )) + .await?; + if !violations.is_empty() { + return Err(DbErr::Migration(format!( + "storage policy legacy cleanup left {} foreign-key violation(s)", + violations.len() + ))); + } + } + Ok(()) + } + + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + // Historical compatibility tables and columns are intentionally not + // recreated. They contained secrets and have no safe downgrade. + Ok(()) + } +} + +async fn ensure_no_unmigrated_credentials(manager: &SchemaManager<'_>) -> Result<(), DbErr> { + let connection = manager.get_connection(); + let backend = manager.get_database_backend(); + + for table in LEGACY_TABLES { + if !manager.has_table(*table).await? { + continue; + } + let count = count_rows(connection, backend, table).await?; + if count > 0 { + return Err(DbErr::Migration(format!( + "storage policy legacy table {table} contains {count} row(s); start the database successfully on AsterDrive 0.5.0 before upgrading to 0.5.1" + ))); + } + } + + if !manager.has_table("storage_policies").await? { + return Ok(()); + } + let columns = existing_legacy_columns(manager).await?; + let static_columns = columns + .iter() + .filter(|column| **column == "access_key" || **column == "secret_key") + .copied() + .collect::>(); + if static_columns.is_empty() { + return Ok(()); + } + let predicates = static_columns + .iter() + .map(|column| format!("TRIM(COALESCE({column}, '')) <> ''")) + .collect::>(); + let sql = format!( + "SELECT COUNT(*) FROM storage_policies WHERE {}", + predicates.join(" OR ") + ); + let count = scalar_count(connection, backend, &sql).await?; + if count > 0 { + return Err(DbErr::Migration(format!( + "storage_policies contains {count} unmigrated static credential row(s); start the database successfully on AsterDrive 0.5.0 before upgrading to 0.5.1" + ))); + } + Ok(()) +} + +async fn existing_legacy_columns(manager: &SchemaManager<'_>) -> Result, DbErr> { + let mut columns = Vec::new(); + for column in LEGACY_POLICY_COLUMNS { + if manager.has_column("storage_policies", *column).await? { + columns.push(*column); + } + } + Ok(columns) +} + +async fn drop_legacy_tables(manager: &SchemaManager<'_>) -> Result<(), DbErr> { + for table in LEGACY_TABLES { + if manager.has_table(*table).await? { + manager + .drop_table(Table::drop().table(Alias::new(*table)).to_owned()) + .await?; + } + } + Ok(()) +} + +async fn drop_legacy_policy_constraints(manager: &SchemaManager<'_>) -> Result<(), DbErr> { + if !manager.has_table("storage_policies").await? { + return Ok(()); + } + let backend = manager.get_database_backend(); + let connection = manager.get_connection(); + let mysql_constraint_present = if backend == DbBackend::MySql { + mysql_constraint_exists(connection, "fk_storage_policies_remote_node_id").await? + } else { + false + }; + match backend { + DbBackend::Sqlite => {} + DbBackend::Postgres => { + connection + .execute_unprepared("ALTER TABLE \"storage_policies\" DROP CONSTRAINT IF EXISTS \"fk_storage_policies_remote_node_id\"") + .await?; + } + DbBackend::MySql if mysql_constraint_present => { + connection + .execute_unprepared("ALTER TABLE `storage_policies` DROP FOREIGN KEY `fk_storage_policies_remote_node_id`") + .await?; + } + DbBackend::MySql => {} + _ => {} + } + for index in LEGACY_INDEXES { + drop_index_if_present(manager, index).await?; + } + Ok(()) +} + +async fn drop_legacy_policy_columns(manager: &SchemaManager<'_>) -> Result<(), DbErr> { + if !manager.has_table("storage_policies").await? { + return Ok(()); + } + let sqlite_foreign_keys = if manager.get_database_backend() == DbBackend::Sqlite { + let row = manager + .get_connection() + .query_one_raw(Statement::from_string( + DbBackend::Sqlite, + "PRAGMA foreign_keys", + )) + .await? + .ok_or_else(|| DbErr::Migration("SQLite foreign-key status returned no row".into()))?; + let enabled = row.try_get_by_index::(0)? != 0; + if enabled { + manager + .get_connection() + .execute_unprepared("PRAGMA foreign_keys = OFF") + .await?; + } + Some(enabled) + } else { + None + }; + let result = async { + for column in LEGACY_POLICY_COLUMNS { + if manager.has_column("storage_policies", *column).await? { + manager + .alter_table( + Table::alter() + .table(Alias::new("storage_policies")) + .drop_column(Alias::new(*column)) + .to_owned(), + ) + .await?; + } + } + Ok::<(), DbErr>(()) + } + .await; + if sqlite_foreign_keys == Some(true) { + manager + .get_connection() + .execute_unprepared("PRAGMA foreign_keys = ON") + .await?; + } + result +} + +async fn drop_index_if_present(manager: &SchemaManager<'_>, index: &str) -> Result<(), DbErr> { + let backend = manager.get_database_backend(); + let sql = match backend { + DbBackend::Sqlite => format!("DROP INDEX IF EXISTS \"{index}\""), + DbBackend::Postgres => format!("DROP INDEX IF EXISTS \"{index}\""), + DbBackend::MySql => { + if !mysql_index_exists(manager.get_connection(), index).await? { + return Ok(()); + } + format!("DROP INDEX `{index}` ON `storage_policies`") + } + _ => return Ok(()), + }; + manager.get_connection().execute_unprepared(&sql).await?; + Ok(()) +} + +async fn count_rows( + connection: &C, + backend: DbBackend, + table: &str, +) -> Result { + scalar_count( + connection, + backend, + &format!("SELECT COUNT(*) FROM {table}"), + ) + .await +} + +async fn scalar_count( + connection: &C, + backend: DbBackend, + sql: &str, +) -> Result { + let row = connection + .query_one_raw(Statement::from_string(backend, sql.to_owned())) + .await? + .ok_or_else(|| DbErr::Migration(format!("count query returned no row: {sql}")))?; + row.try_get_by_index(0) +} + +async fn mysql_index_exists( + connection: &C, + index: &str, +) -> Result { + let sql = format!( + "SELECT COUNT(*) FROM information_schema.statistics WHERE table_schema = DATABASE() AND table_name = 'storage_policies' AND index_name = '{index}'" + ); + Ok(scalar_count(connection, DbBackend::MySql, &sql).await? > 0) +} + +async fn mysql_constraint_exists( + connection: &C, + constraint: &str, +) -> Result { + let sql = format!( + "SELECT COUNT(*) FROM information_schema.table_constraints WHERE table_schema = DATABASE() AND table_name = 'storage_policies' AND constraint_name = '{constraint}'" + ); + Ok(scalar_count(connection, DbBackend::MySql, &sql).await? > 0) +} diff --git a/crates/aster_drive_model/src/deprecated/mod.rs b/crates/aster_drive_model/src/deprecated/mod.rs deleted file mode 100644 index b63574227..000000000 --- a/crates/aster_drive_model/src/deprecated/mod.rs +++ /dev/null @@ -1,22 +0,0 @@ -//! Legacy database models exclusive to the AsterDrive 0.5.0 upgrade path. -//! -//! These modules exist only for upgrade-time data migration and will be -//! completely removed in AsterDrive 0.6.0. - -#[cfg_attr( - not(test), - deprecated( - since = "0.5.0", - note = "legacy migration-only entity; scheduled for removal in AsterDrive 0.6.0" - ) -)] -pub mod storage_connector_application_config; - -#[cfg_attr( - not(test), - deprecated( - since = "0.5.0", - note = "legacy migration-only entity; scheduled for removal in AsterDrive 0.6.0" - ) -)] -pub mod storage_policy_credential; diff --git a/crates/aster_drive_model/src/deprecated/storage_connector_application_config.rs b/crates/aster_drive_model/src/deprecated/storage_connector_application_config.rs deleted file mode 100644 index 1ed11e765..000000000 --- a/crates/aster_drive_model/src/deprecated/storage_connector_application_config.rs +++ /dev/null @@ -1,72 +0,0 @@ -//! Deprecated SeaORM entity for `storage_connector_application_configs`. -//! -//! This definition is exclusive to the AsterDrive 0.5.0 upgrade path. It only -//! migrates legacy rows into connector-owned credentials and will be completely -//! removed in AsterDrive 0.6.0. - -use sea_orm::entity::prelude::*; -use serde::{Deserialize, Serialize}; -use std::fmt; - -use crate::types::StorageCredentialProvider; - -#[derive(Clone, PartialEq, DeriveEntityModel, Serialize, Deserialize)] -#[sea_orm(table_name = "storage_connector_application_configs")] -pub struct Model { - #[sea_orm(primary_key)] - pub id: i64, - pub policy_id: i64, - pub provider: StorageCredentialProvider, - pub tenant_id: Option, - pub scopes: String, - pub client_id: Option, - #[serde(skip_serializing)] - pub client_secret_ciphertext: Option, - pub metadata: String, - pub created_at: DateTimeUtc, - pub updated_at: DateTimeUtc, -} - -impl fmt::Debug for Model { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter - .debug_struct("DeprecatedStorageConnectorApplicationConfig") - .field("id", &self.id) - .field("policy_id", &self.policy_id) - .field("provider", &self.provider) - .field("tenant_id", &self.tenant_id) - .field("scopes", &self.scopes) - .field("client_id", &self.client_id) - .field( - "client_secret_ciphertext", - &self - .client_secret_ciphertext - .as_ref() - .map(|_| "***REDACTED***"), - ) - .field("metadata", &"***REDACTED***") - .field("created_at", &self.created_at) - .field("updated_at", &self.updated_at) - .finish() - } -} - -#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] -pub enum Relation { - #[sea_orm( - belongs_to = "crate::entities::storage_policy::Entity", - from = "Column::PolicyId", - to = "crate::entities::storage_policy::Column::Id", - on_update = "NoAction", - on_delete = "Cascade" - )] - StoragePolicy, -} - -impl Related for Entity { - fn to() -> RelationDef { - Relation::StoragePolicy.def() - } -} - -impl ActiveModelBehavior for ActiveModel {} diff --git a/crates/aster_drive_model/src/deprecated/storage_policy_credential.rs b/crates/aster_drive_model/src/deprecated/storage_policy_credential.rs deleted file mode 100644 index 955e17f71..000000000 --- a/crates/aster_drive_model/src/deprecated/storage_policy_credential.rs +++ /dev/null @@ -1,136 +0,0 @@ -//! Deprecated SeaORM entity for `storage_policy_credentials`. -//! -//! This definition is exclusive to the AsterDrive 0.5.0 upgrade path. It only -//! migrates legacy rows into connector-owned credentials and will be completely -//! removed in AsterDrive 0.6.0. - -use sea_orm::entity::prelude::*; -use serde::{Deserialize, Serialize}; -use std::fmt; - -use crate::types::{StorageCredentialKind, StorageCredentialProvider, StorageCredentialStatus}; - -#[derive(Clone, PartialEq, DeriveEntityModel, Serialize, Deserialize)] -#[sea_orm(table_name = "storage_policy_credentials")] -pub struct Model { - #[sea_orm(primary_key)] - pub id: i64, - pub policy_id: i64, - pub provider: StorageCredentialProvider, - pub credential_kind: StorageCredentialKind, - pub account_label: Option, - pub subject: Option, - pub tenant_id: Option, - pub scopes: String, - #[serde(skip_serializing)] - pub access_token_ciphertext: Option, - #[serde(skip_serializing)] - pub refresh_token_ciphertext: Option, - pub metadata: String, - pub status: StorageCredentialStatus, - pub status_reason: Option, - pub expires_at: Option, - pub authorized_at: Option, - pub last_refreshed_at: Option, - pub last_validated_at: Option, - pub created_at: DateTimeUtc, - pub updated_at: DateTimeUtc, -} - -impl fmt::Debug for Model { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter - .debug_struct("DeprecatedStoragePolicyCredential") - .field("id", &self.id) - .field("policy_id", &self.policy_id) - .field("provider", &self.provider) - .field("credential_kind", &self.credential_kind) - .field("account_label", &self.account_label) - .field("subject", &self.subject) - .field("tenant_id", &self.tenant_id) - .field("scopes", &self.scopes) - .field( - "access_token_ciphertext", - &self - .access_token_ciphertext - .as_ref() - .map(|_| "***REDACTED***"), - ) - .field( - "refresh_token_ciphertext", - &self - .refresh_token_ciphertext - .as_ref() - .map(|_| "***REDACTED***"), - ) - .field("metadata", &"***REDACTED***") - .field("status", &self.status) - .field("status_reason", &self.status_reason) - .field("expires_at", &self.expires_at) - .field("authorized_at", &self.authorized_at) - .field("last_refreshed_at", &self.last_refreshed_at) - .field("last_validated_at", &self.last_validated_at) - .field("created_at", &self.created_at) - .field("updated_at", &self.updated_at) - .finish() - } -} - -#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] -pub enum Relation { - #[sea_orm( - belongs_to = "crate::entities::storage_policy::Entity", - from = "Column::PolicyId", - to = "crate::entities::storage_policy::Column::Id", - on_update = "NoAction", - on_delete = "Cascade" - )] - StoragePolicy, -} - -impl Related for Entity { - fn to() -> RelationDef { - Relation::StoragePolicy.def() - } -} - -impl ActiveModelBehavior for ActiveModel {} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn debug_redacts_legacy_storage_credential_secrets() { - let now = chrono::Utc::now(); - let model = Model { - id: 1, - policy_id: 2, - provider: StorageCredentialProvider::MicrosoftGraph, - credential_kind: StorageCredentialKind::OauthDelegated, - account_label: Some("admin@example.com".to_string()), - subject: Some("subject".to_string()), - tenant_id: Some("tenant".to_string()), - scopes: r#"["offline_access","Files.ReadWrite.All"]"#.to_string(), - access_token_ciphertext: Some("access-secret".to_string()), - refresh_token_ciphertext: Some("refresh-secret".to_string()), - metadata: r#"{"drive_id":"secret-drive"}"#.to_string(), - status: StorageCredentialStatus::Authorized, - status_reason: None, - expires_at: Some(now), - authorized_at: Some(now), - last_refreshed_at: None, - last_validated_at: None, - created_at: now, - updated_at: now, - }; - - let debug = format!("{model:?}"); - assert!(debug.contains(r#"access_token_ciphertext: Some("***REDACTED***")"#)); - assert!(debug.contains(r#"refresh_token_ciphertext: Some("***REDACTED***")"#)); - assert!(debug.contains(r#"metadata: "***REDACTED***""#)); - assert!(!debug.contains("access-secret")); - assert!(!debug.contains("refresh-secret")); - assert!(!debug.contains("secret-drive")); - } -} diff --git a/crates/aster_drive_model/src/lib.rs b/crates/aster_drive_model/src/lib.rs index a76920c02..7b48286cd 100644 --- a/crates/aster_drive_model/src/lib.rs +++ b/crates/aster_drive_model/src/lib.rs @@ -11,6 +11,5 @@ ) )] -pub mod deprecated; pub mod entities; pub mod types; diff --git a/src/cli/database_migration/apply/convert.rs b/src/cli/database_migration/apply/convert.rs index febe419a5..f000df59d 100644 --- a/src/cli/database_migration/apply/convert.rs +++ b/src/cli/database_migration/apply/convert.rs @@ -40,33 +40,11 @@ pub(super) fn decode_row_values( &plan.name, &column.name, )?; - let cell = normalize_0_5_storage_compatibility_cell(&plan.name, &column.name, cell); cell_into_target_value(cell, target_kind, &plan.name, &column.name) }) .collect() } -/// Normalizes backend-specific values in the temporary 0.5.x storage schema. -/// -/// MySQL keeps the deprecated `storage_policies.options` column nullable because -/// historical MySQL versions do not support the same TEXT default as SQLite and -/// PostgreSQL. Cross-database copy must materialize the historical empty object -/// before inserting into a backend where that retained column is still NOT NULL. -/// Issue #463 removes both the column and this 0.5.x-only normalization in 0.6.0. -fn normalize_0_5_storage_compatibility_cell( - table_name: &str, - column_name: &str, - cell: CellValue, -) -> CellValue { - if table_name == "storage_policies" - && column_name == "options" - && matches!(cell, CellValue::Null) - { - return CellValue::String("{}".to_string()); - } - cell -} - fn decode_source_cell( row: &QueryResult, index: usize, @@ -295,39 +273,7 @@ mod tests { use chrono::DateTime; use sea_orm::Value; - use super::{ - BindingKind, CellValue, cell_into_target_value, normalize_0_5_storage_compatibility_cell, - }; - - #[test] - fn null_legacy_storage_policy_options_normalize_to_empty_object() { - let normalized = normalize_0_5_storage_compatibility_cell( - "storage_policies", - "options", - CellValue::Null, - ); - assert!(matches!(normalized, CellValue::String(value) if value == "{}")); - } - - #[test] - fn storage_compatibility_normalization_does_not_change_other_cells() { - assert!(matches!( - normalize_0_5_storage_compatibility_cell( - "storage_policies", - "access_key", - CellValue::Null, - ), - CellValue::Null - )); - assert!(matches!( - normalize_0_5_storage_compatibility_cell( - "storage_policies", - "options", - CellValue::String(r#"{"content_dedup":true}"#.to_string()), - ), - CellValue::String(value) if value == r#"{"content_dedup":true}"# - )); - } + use super::{BindingKind, CellValue, cell_into_target_value}; #[test] fn string_bool_cells_convert_into_bool_values() { diff --git a/src/cli/database_migration/apply/mod.rs b/src/cli/database_migration/apply/mod.rs index fa7d3bc67..15bc7f092 100644 --- a/src/cli/database_migration/apply/mod.rs +++ b/src/cli/database_migration/apply/mod.rs @@ -117,19 +117,6 @@ pub(super) async fn execute_apply_mode(ctx: ApplyModeContext<'_>) -> Result) -> Result Result<()> { - use sea_orm::TransactionTrait; - - let config = crate::config::load_config_read_only()?; - let connectors = crate::storage::connectors::builtin_storage_connector_registry()?; - aster_drive_migration::with_database_migration_lock(target_db, move |connection| { - let config = config.clone(); - let connectors = connectors.clone(); - Box::pin(async move { - let transaction = connection.begin().await?; - crate::services::storage_policy::credential::migrate_legacy_storage_credentials( - &transaction, - &config, - &connectors, - ) - .await - .map_err(|error| sea_orm::DbErr::Custom(error.to_string()))?; - transaction.commit().await - }) - }) - .await - .map_aster_err(AsterError::database_operation) -} - async fn mark_checkpoint_failed_best_effort( target_db: &DatabaseConnection, checkpoint: &mut MigrationCheckpoint, diff --git a/src/cli/database_migration/mod.rs b/src/cli/database_migration/mod.rs index e23c09024..42e905904 100644 --- a/src/cli/database_migration/mod.rs +++ b/src/cli/database_migration/mod.rs @@ -41,8 +41,6 @@ const COPY_TABLE_ORDER: &[&str] = &[ "remote_tunnel_owners", "storage_policies", "storage_policy_connector_credentials", - "storage_connector_application_configs", - "storage_policy_credentials", "storage_policy_groups", "storage_policy_group_items", "follower_enrollment_sessions", diff --git a/src/cli/database_migration/schema.rs b/src/cli/database_migration/schema.rs index d9ffdf554..825755079 100644 --- a/src/cli/database_migration/schema.rs +++ b/src/cli/database_migration/schema.rs @@ -7,7 +7,9 @@ use std::collections::{BTreeMap, BTreeSet}; use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, Statement}; -use crate::cli::db_shared::{backend_name, join_strings, quote_literal, quote_sqlite_literal}; +use crate::cli::db_shared::{ + backend_name, join_strings, quote_ident, quote_literal, quote_sqlite_literal, scalar_i64, +}; use crate::db; use crate::errors::{AsterError, MapAsterErr, Result}; @@ -53,6 +55,7 @@ pub(super) async fn load_source_plans(source: &DatabaseConnection) -> Result = existing_tables.iter().map(String::as_str).collect(); let mut plans = Vec::with_capacity(COPY_TABLE_ORDER.len()); @@ -63,7 +66,24 @@ pub(super) async fn load_source_plans(source: &DatabaseConnection) -> Result Res Ok(()) } +async fn validate_legacy_storage_source( + source: &DatabaseConnection, + backend: DbBackend, + existing_tables: &[String], +) -> Result<()> { + for table in [ + "storage_policy_credentials", + "storage_connector_application_configs", + ] { + if existing_tables.iter().any(|name| name == table) { + let rows = count_rows(source, backend, table).await?; + if rows > 0 { + return Err(AsterError::validation_error(format!( + "source database contains {rows} unmigrated rows in removed legacy table '{table}'; start it successfully on AsterDrive 0.5.x before copying" + ))); + } + } + } + + if !existing_tables + .iter() + .any(|name| name == "storage_policies") + { + return Ok(()); + } + let columns = load_column_type_rows(source, backend, "storage_policies").await?; + let static_columns = columns + .iter() + .map(|(name, _)| name.as_str()) + .filter(|name| matches!(*name, "access_key" | "secret_key")) + .collect::>(); + if static_columns.is_empty() { + return Ok(()); + } + let predicates = static_columns + .iter() + .map(|column| format!("TRIM(COALESCE({}, '')) <> ''", quote_ident(backend, column))) + .collect::>() + .join(" OR "); + let count = scalar_i64( + source, + backend, + &format!( + "SELECT COUNT(*) FROM {} WHERE {predicates}", + quote_ident(backend, "storage_policies") + ), + ) + .await?; + if count > 0 { + return Err(AsterError::validation_error(format!( + "source database contains {count} unmigrated static storage credential row(s); start it successfully on AsterDrive 0.5.x before copying" + ))); + } + Ok(()) +} + async fn load_table_plan(db: &C, backend: DbBackend, table_name: &str) -> Result where C: ConnectionTrait, diff --git a/src/db/repository/mod.rs b/src/db/repository/mod.rs index f3253cd28..f245b7f7d 100644 --- a/src/db/repository/mod.rs +++ b/src/db/repository/mod.rs @@ -34,19 +34,9 @@ pub mod remote_tunnel_owner_repo; pub mod revision_repo; pub mod search_repo; pub mod share_repo; -#[deprecated( - since = "0.5.0", - note = "legacy migration-only repository; scheduled for removal in AsterDrive 0.6.0" -)] -pub mod storage_connector_application_config_repo; pub mod storage_migration_checkpoint_repo; pub mod storage_policy_authorization_flow_repo; pub mod storage_policy_connector_credential_repo; -#[deprecated( - since = "0.5.0", - note = "legacy migration-only repository; scheduled for removal in AsterDrive 0.6.0" -)] -pub mod storage_policy_credential_repo; pub mod system_initialization_repo; pub mod tag_repo; pub mod team_member_repo; diff --git a/src/db/repository/storage_connector_application_config_repo.rs b/src/db/repository/storage_connector_application_config_repo.rs deleted file mode 100644 index 1519d7edd..000000000 --- a/src/db/repository/storage_connector_application_config_repo.rs +++ /dev/null @@ -1,88 +0,0 @@ -//! Deprecated repository helpers for `storage_connector_application_configs`. -//! -//! This repository is exclusive to the AsterDrive 0.5.0 upgrade path and will -//! be completely removed with the legacy table in AsterDrive 0.6.0. -#![expect( - deprecated, - reason = "AsterDrive 0.5.0-only repository is removed with the legacy table in 0.6.0" -)] - -use chrono::Utc; -use sea_orm::{ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, Set, sea_query::OnConflict}; - -use crate::errors::{AsterError, Result}; -use aster_drive_model::deprecated::storage_connector_application_config::{ - self, Entity as StorageConnectorApplicationConfig, -}; -use aster_drive_model::types::StorageCredentialProvider; - -pub async fn find_by_policy_provider( - db: &C, - policy_id: i64, - provider: StorageCredentialProvider, -) -> Result> { - StorageConnectorApplicationConfig::find() - .filter(storage_connector_application_config::Column::PolicyId.eq(policy_id)) - .filter(storage_connector_application_config::Column::Provider.eq(provider)) - .one(db) - .await - .map_err(AsterError::from) -} - -pub async fn upsert_by_policy_provider( - db: &C, - mut model: storage_connector_application_config::ActiveModel, - now: chrono::DateTime, -) -> Result { - let policy_id = active_i64(&model.policy_id, "policy_id")?; - let provider = active_provider(&model.provider)?; - - model.created_at = Set(now); - model.updated_at = Set(now); - - StorageConnectorApplicationConfig::insert(model) - .on_conflict( - OnConflict::columns([ - storage_connector_application_config::Column::PolicyId, - storage_connector_application_config::Column::Provider, - ]) - .update_columns([ - storage_connector_application_config::Column::TenantId, - storage_connector_application_config::Column::Scopes, - storage_connector_application_config::Column::ClientId, - storage_connector_application_config::Column::ClientSecretCiphertext, - storage_connector_application_config::Column::Metadata, - storage_connector_application_config::Column::UpdatedAt, - ]) - .to_owned(), - ) - .exec(db) - .await - .map_err(AsterError::from)?; - - find_by_policy_provider(db, policy_id, provider) - .await? - .ok_or_else(|| { - AsterError::record_not_found("storage connector application config after upsert") - }) -} - -fn active_i64(value: &sea_orm::ActiveValue, field: &str) -> Result { - match value { - sea_orm::ActiveValue::Set(value) | sea_orm::ActiveValue::Unchanged(value) => Ok(*value), - sea_orm::ActiveValue::NotSet => Err(AsterError::internal_error(format!( - "storage connector application config active model missing {field}" - ))), - } -} - -fn active_provider( - value: &sea_orm::ActiveValue, -) -> Result { - match value { - sea_orm::ActiveValue::Set(value) | sea_orm::ActiveValue::Unchanged(value) => Ok(*value), - sea_orm::ActiveValue::NotSet => Err(AsterError::internal_error( - "storage connector application config active model missing provider".to_string(), - )), - } -} diff --git a/src/db/repository/storage_policy_credential_repo.rs b/src/db/repository/storage_policy_credential_repo.rs deleted file mode 100644 index 1c959c338..000000000 --- a/src/db/repository/storage_policy_credential_repo.rs +++ /dev/null @@ -1,196 +0,0 @@ -//! Deprecated repository helpers for `storage_policy_credentials`. -//! -//! This repository is exclusive to the AsterDrive 0.5.0 upgrade path and will -//! be completely removed with the legacy table in AsterDrive 0.6.0. -#![expect( - deprecated, - reason = "AsterDrive 0.5.0-only repository is removed with the legacy table in 0.6.0" -)] - -use chrono::Utc; -use sea_orm::{ - ActiveEnum, ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, Set, - sea_query::{Expr, OnConflict}, -}; - -use crate::errors::{AsterError, Result}; -use aster_drive_model::deprecated::storage_policy_credential::{ - self, Entity as StoragePolicyCredential, -}; -use aster_drive_model::types::{ - StorageCredentialKind, StorageCredentialProvider, StorageCredentialStatus, -}; - -pub async fn find_all(db: &C) -> Result> { - StoragePolicyCredential::find() - .all(db) - .await - .map_err(AsterError::from) -} - -pub async fn find_by_policy_provider_kind( - db: &C, - policy_id: i64, - provider: StorageCredentialProvider, - credential_kind: StorageCredentialKind, -) -> Result> { - StoragePolicyCredential::find() - .filter(storage_policy_credential::Column::PolicyId.eq(policy_id)) - .filter(storage_policy_credential::Column::Provider.eq(provider)) - .filter(storage_policy_credential::Column::CredentialKind.eq(credential_kind)) - .one(db) - .await - .map_err(AsterError::from) -} - -pub async fn list_by_policy( - db: &C, - policy_id: i64, -) -> Result> { - StoragePolicyCredential::find() - .filter(storage_policy_credential::Column::PolicyId.eq(policy_id)) - .all(db) - .await - .map_err(AsterError::from) -} - -pub async fn upsert_by_policy_provider_kind( - db: &C, - mut model: storage_policy_credential::ActiveModel, - now: chrono::DateTime, -) -> Result { - let policy_id = active_i64(&model.policy_id, "policy_id")?; - let provider = active_provider(&model.provider)?; - let credential_kind = active_credential_kind(&model.credential_kind)?; - - model.created_at = Set(now); - model.updated_at = Set(now); - - StoragePolicyCredential::insert(model) - .on_conflict( - OnConflict::columns([ - storage_policy_credential::Column::PolicyId, - storage_policy_credential::Column::Provider, - storage_policy_credential::Column::CredentialKind, - ]) - .update_columns([ - storage_policy_credential::Column::AccountLabel, - storage_policy_credential::Column::Subject, - storage_policy_credential::Column::TenantId, - storage_policy_credential::Column::Scopes, - storage_policy_credential::Column::AccessTokenCiphertext, - storage_policy_credential::Column::RefreshTokenCiphertext, - storage_policy_credential::Column::Metadata, - storage_policy_credential::Column::Status, - storage_policy_credential::Column::StatusReason, - storage_policy_credential::Column::ExpiresAt, - storage_policy_credential::Column::AuthorizedAt, - storage_policy_credential::Column::LastRefreshedAt, - storage_policy_credential::Column::LastValidatedAt, - storage_policy_credential::Column::UpdatedAt, - ]) - .to_owned(), - ) - .exec(db) - .await - .map_err(AsterError::from)?; - - find_by_policy_provider_kind(db, policy_id, provider, credential_kind) - .await? - .ok_or_else(|| AsterError::record_not_found("storage policy credential after upsert")) -} - -pub struct OAuthRefreshUpdate<'a> { - pub policy_id: i64, - pub provider: StorageCredentialProvider, - pub credential_kind: StorageCredentialKind, - pub expected_refresh_token_ciphertext: &'a str, - pub access_token_ciphertext: String, - pub refresh_token_ciphertext: Option, - pub expires_at: Option>, - pub scopes: Option, - pub now: chrono::DateTime, -} - -pub async fn update_oauth_refresh_result_if_refresh_token_matches( - db: &C, - input: OAuthRefreshUpdate<'_>, -) -> Result { - let mut update = StoragePolicyCredential::update_many() - .col_expr( - storage_policy_credential::Column::AccessTokenCiphertext, - Expr::value(Some(input.access_token_ciphertext)), - ) - .col_expr( - storage_policy_credential::Column::ExpiresAt, - Expr::value(input.expires_at), - ) - .col_expr( - storage_policy_credential::Column::LastRefreshedAt, - Expr::value(Some(input.now)), - ) - .col_expr( - storage_policy_credential::Column::Status, - Expr::value(StorageCredentialStatus::Authorized.to_value()), - ) - .col_expr( - storage_policy_credential::Column::StatusReason, - Expr::value(Option::::None), - ) - .col_expr( - storage_policy_credential::Column::UpdatedAt, - Expr::value(input.now), - ) - .filter(storage_policy_credential::Column::PolicyId.eq(input.policy_id)) - .filter(storage_policy_credential::Column::Provider.eq(input.provider)) - .filter(storage_policy_credential::Column::CredentialKind.eq(input.credential_kind)) - .filter( - storage_policy_credential::Column::RefreshTokenCiphertext - .eq(input.expected_refresh_token_ciphertext), - ); - if let Some(scopes) = input.scopes { - update = update.col_expr( - storage_policy_credential::Column::Scopes, - Expr::value(scopes), - ); - } - if let Some(refresh_token_ciphertext) = input.refresh_token_ciphertext { - update = update.col_expr( - storage_policy_credential::Column::RefreshTokenCiphertext, - Expr::value(Some(refresh_token_ciphertext)), - ); - } - let result = update.exec(db).await.map_err(AsterError::from)?; - Ok(result.rows_affected == 1) -} - -fn active_i64(value: &sea_orm::ActiveValue, field: &str) -> Result { - match value { - sea_orm::ActiveValue::Set(value) | sea_orm::ActiveValue::Unchanged(value) => Ok(*value), - sea_orm::ActiveValue::NotSet => Err(AsterError::internal_error(format!( - "storage credential active model missing {field}" - ))), - } -} - -fn active_provider( - value: &sea_orm::ActiveValue, -) -> Result { - match value { - sea_orm::ActiveValue::Set(value) | sea_orm::ActiveValue::Unchanged(value) => Ok(*value), - sea_orm::ActiveValue::NotSet => Err(AsterError::internal_error( - "storage credential active model missing provider".to_string(), - )), - } -} - -fn active_credential_kind( - value: &sea_orm::ActiveValue, -) -> Result { - match value { - sea_orm::ActiveValue::Set(value) | sea_orm::ActiveValue::Unchanged(value) => Ok(*value), - sea_orm::ActiveValue::NotSet => Err(AsterError::internal_error( - "storage credential active model missing credential_kind".to_string(), - )), - } -} diff --git a/src/runtime/startup/common.rs b/src/runtime/startup/common.rs index 5205cecc7..ccd9320ff 100644 --- a/src/runtime/startup/common.rs +++ b/src/runtime/startup/common.rs @@ -6,7 +6,6 @@ use crate::errors::{AsterError, MapAsterErr, Result}; use crate::storage::DriverRegistry; use aster_drive_metrics::SharedMetricsRecorder; use aster_drive_migration::Migrator; -use sea_orm::TransactionTrait; use std::sync::Arc; pub(super) struct CommonRuntimeParts { @@ -106,24 +105,6 @@ pub async fn initialize_database_state( .await .map_aster_err(AsterError::database_operation)?; let connector_registry = crate::storage::connectors::builtin_storage_connector_registry()?; - let upgrade_config = cfg.clone(); - let upgrade_connectors = connector_registry.clone(); - aster_drive_migration::with_database_migration_lock(database, move |connection| { - Box::pin(async move { - let credential_transaction = connection.begin().await?; - crate::services::storage_policy::credential::migrate_legacy_storage_credentials( - &credential_transaction, - &upgrade_config, - &upgrade_connectors, - ) - .await - .map_err(|error| sea_orm::DbErr::Custom(error.to_string()))?; - credential_transaction.commit().await?; - Ok(()) - }) - }) - .await - .map_err(|error| AsterError::database_operation(error.to_string()))?; if let Some(sqlite_search) = db::sqlite_search::ensure_sqlite_search_ready(database).await? { tracing::info!( @@ -300,11 +281,36 @@ mod tests { "options", ] { assert!( - schema + !schema .has_column("storage_policies", legacy_column) .await .unwrap(), - "0.5 startup should retain legacy storage policy column {legacy_column}" + "0.5.1 startup should remove legacy storage policy column {legacy_column}" + ); + } + for current_column in ["connector_id", "storage_config"] { + assert!( + schema + .has_column("storage_policies", current_column) + .await + .unwrap(), + "0.5.1 startup should retain current storage policy column {current_column}" + ); + } + assert!( + schema + .has_table("storage_policy_connector_credentials") + .await + .unwrap(), + "0.5.1 startup should retain connector-owned credentials" + ); + for legacy_table in [ + "storage_policy_credentials", + "storage_connector_application_configs", + ] { + assert!( + !schema.has_table(legacy_table).await.unwrap(), + "0.5.1 startup should remove legacy credential table {legacy_table}" ); } crate::storage::connectors::test_support::insertable_policy( @@ -312,7 +318,7 @@ mod tests { ) .insert(&db) .await - .expect("current storage policy entity should ignore retained legacy columns"); + .expect("current storage policy entity should match the final schema"); } } diff --git a/src/services/preview/wopi/targets.rs b/src/services/preview/wopi/targets.rs index 7ea76a45d..0fc304a8b 100644 --- a/src/services/preview/wopi/targets.rs +++ b/src/services/preview/wopi/targets.rs @@ -507,8 +507,10 @@ pub(crate) fn decode_wopi_filename(value: &str) -> Result { } let utf16 = bytes - .chunks_exact(2) - .map(|chunk| u16::from_be_bytes([chunk[0], chunk[1]])); + .as_chunks::<2>() + .0 + .iter() + .map(|chunk| u16::from_be_bytes(*chunk)); for ch in char::decode_utf16(utf16) { decoded.push(ch.map_aster_err_with(|| { AsterError::validation_error("invalid UTF-16 sequence in WOPI target header") diff --git a/src/services/storage_policy/credential/migration.rs b/src/services/storage_policy/credential/migration.rs deleted file mode 100644 index 6b2c2aa01..000000000 --- a/src/services/storage_policy/credential/migration.rs +++ /dev/null @@ -1,1082 +0,0 @@ -//! AsterDrive 0.5.0-only startup migration from deprecated credential stores. -//! -//! This is intentionally application-level rather than a historical schema -//! migration: connector code owns payload conversion, while the already-loaded -//! runtime config supplies the encryption key. Credential import, legacy value -//! clearing, and deprecated-row deletion run in one transaction before the -//! server begins listening. The compatibility schema itself remains until the -//! 0.6.0 migration tracked by issue #463. -//! -//! This module and both deprecated source tables are scheduled for complete -//! removal in AsterDrive 0.6.0. - -#![expect( - deprecated, - reason = "AsterDrive 0.5.0 startup migration reads deprecated credential stores until 0.6.0" -)] - -use std::collections::BTreeMap; - -use aster_drive_migration::SchemaManager; -use sea_orm::{ - ColumnTrait, ConnectionTrait, DatabaseTransaction, DbBackend, EntityTrait, QueryFilter, - QueryOrder, QuerySelect, - sea_query::{Alias, Expr, Query}, -}; - -use crate::config::Config; -use crate::errors::{AsterError, Result}; -use crate::storage::connectors::{ - LegacyStorageConnectorCredentialInput, LegacyStoragePolicyStaticCredential, - StorageConnectorRegistry, -}; -use aster_drive_model::deprecated::{ - storage_connector_application_config, storage_policy_credential, -}; -use aster_drive_model::entities::{storage_policy, storage_policy_connector_credential}; - -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] -pub(crate) struct LegacyStorageCredentialMigrationReport { - pub(crate) migrated: usize, - pub(crate) already_current: usize, - pub(crate) legacy_rows_deleted: usize, -} - -#[derive(Debug)] -struct LegacyStaticCredentialRow { - policy_id: i64, - access_key: String, - secret_key: String, -} - -/// AsterDrive 0.5.0-only import after config and schema loading. -/// -/// Policy rows are locked on shared SQL backends so concurrent primary startup -/// cannot race the import. A conversion, decryption, conflict, or cleanup error -/// aborts the transaction and therefore aborts startup without partial writes. -/// This entrypoint is scheduled for removal in AsterDrive 0.6.0. -pub(crate) async fn migrate_legacy_storage_credentials( - transaction: &DatabaseTransaction, - config: &Config, - connectors: &StorageConnectorRegistry, -) -> Result { - let mut policy_query = storage_policy::Entity::find().order_by_asc(storage_policy::Column::Id); - if transaction.get_database_backend() != DbBackend::Sqlite { - policy_query = policy_query.lock_exclusive(); - } - let policies = policy_query - .all(transaction) - .await - .map_err(AsterError::from)?; - let static_rows = load_legacy_static_credentials(transaction).await?; - let applications = storage_connector_application_config::Entity::find() - .order_by_asc(storage_connector_application_config::Column::Id) - .all(transaction) - .await - .map_err(AsterError::from)?; - let authorizations = storage_policy_credential::Entity::find() - .order_by_asc(storage_policy_credential::Column::Id) - .all(transaction) - .await - .map_err(AsterError::from)?; - - let has_legacy_data = static_rows - .iter() - .any(|row| !row.access_key.is_empty() || !row.secret_key.is_empty()) - || !applications.is_empty() - || !authorizations.is_empty(); - if !has_legacy_data { - return Ok(LegacyStorageCredentialMigrationReport::default()); - } - - let mut static_by_policy = static_rows - .into_iter() - .map(|row| (row.policy_id, row)) - .collect::>(); - let mut applications_by_policy = group_application_rows(applications); - let mut authorizations_by_policy = group_authorization_rows(authorizations); - let mut report = LegacyStorageCredentialMigrationReport::default(); - - for policy in &policies { - let static_credential = static_by_policy.remove(&policy.id).and_then(|row| { - let access_key = row.access_key.trim().to_string(); - let secret_key = row.secret_key.trim().to_string(); - (!access_key.is_empty() || !secret_key.is_empty()).then_some( - LegacyStoragePolicyStaticCredential { - access_key, - secret_key, - }, - ) - }); - let application_config = take_single_legacy_row( - &mut applications_by_policy, - policy.id, - "application credential", - )?; - let authorization = take_single_legacy_row( - &mut authorizations_by_policy, - policy.id, - "authorization credential", - )?; - let input = LegacyStorageConnectorCredentialInput { - static_credential, - application_config, - authorization, - }; - if input.is_empty() { - continue; - } - - let connector = connectors.require_policy(policy)?; - let descriptor = connector.descriptor(); - let Some(imported) = connector.import_legacy_credential( - &config.auth.storage_credential_secret_key, - policy, - input, - )? - else { - continue; - }; - let existing = storage_policy_connector_credential::Entity::find() - .filter(storage_policy_connector_credential::Column::PolicyId.eq(policy.id)) - .one(transaction) - .await - .map_err(AsterError::from)?; - if let Some(existing) = existing { - let existing_payload = crate::storage::connectors::decode_connector_credential( - &config.auth.storage_credential_secret_key, - &existing, - &descriptor.connector_id, - crate::storage::connectors::credential_schema_version(&descriptor)?, - )?; - if existing_payload != imported { - return Err(AsterError::database_operation(format!( - "storage policy {} has conflicting legacy and connector-owned credentials", - policy.id - ))); - } - report.already_current += 1; - continue; - } - - crate::storage::connectors::persist_connector_credential_payload( - transaction, - &config.auth.storage_credential_secret_key, - policy.id, - &descriptor.connector_id, - crate::storage::connectors::credential_schema_version(&descriptor)?, - &imported, - ) - .await?; - report.migrated += 1; - } - - ensure_no_orphaned_legacy_rows( - &static_by_policy, - &applications_by_policy, - &authorizations_by_policy, - )?; - clear_legacy_static_credentials(transaction).await?; - let deleted_applications = storage_connector_application_config::Entity::delete_many() - .exec(transaction) - .await - .map_err(AsterError::from)? - .rows_affected; - let deleted_authorizations = storage_policy_credential::Entity::delete_many() - .exec(transaction) - .await - .map_err(AsterError::from)? - .rows_affected; - report.legacy_rows_deleted = usize::try_from( - deleted_applications - .checked_add(deleted_authorizations) - .ok_or_else(|| { - AsterError::database_operation("legacy credential delete count overflow") - })?, - ) - .map_err(|_| AsterError::database_operation("legacy credential delete count exceeds usize"))?; - - tracing::info!( - migrated = report.migrated, - already_current = report.already_current, - legacy_rows_deleted = report.legacy_rows_deleted, - "legacy storage credentials migrated to connector-owned payloads" - ); - Ok(report) -} - -fn group_application_rows( - rows: Vec, -) -> BTreeMap> { - let mut grouped = BTreeMap::new(); - for row in rows { - grouped - .entry(row.policy_id) - .or_insert_with(Vec::new) - .push(row); - } - grouped -} - -fn group_authorization_rows( - rows: Vec, -) -> BTreeMap> { - let mut grouped = BTreeMap::new(); - for row in rows { - grouped - .entry(row.policy_id) - .or_insert_with(Vec::new) - .push(row); - } - grouped -} - -fn take_single_legacy_row( - rows: &mut BTreeMap>, - policy_id: i64, - kind: &str, -) -> Result> { - let Some(mut rows) = rows.remove(&policy_id) else { - return Ok(None); - }; - if rows.len() != 1 { - return Err(AsterError::database_operation(format!( - "storage policy {policy_id} has multiple legacy {kind} rows" - ))); - } - Ok(rows.pop()) -} - -fn ensure_no_orphaned_legacy_rows( - static_rows: &BTreeMap, - application_rows: &BTreeMap>, - authorization_rows: &BTreeMap>, -) -> Result<()> { - let orphaned_policy_id = static_rows - .keys() - .chain(application_rows.keys()) - .chain(authorization_rows.keys()) - .next() - .copied(); - if let Some(policy_id) = orphaned_policy_id { - return Err(AsterError::database_operation(format!( - "legacy storage credentials reference missing storage policy {policy_id}" - ))); - } - Ok(()) -} - -async fn load_legacy_static_credentials( - db: &sea_orm::DatabaseTransaction, -) -> Result> { - if !legacy_static_credential_columns_exist(db).await? { - return Ok(Vec::new()); - } - let statement = Query::select() - .columns([ - Alias::new("id"), - Alias::new("access_key"), - Alias::new("secret_key"), - ]) - .from(Alias::new("storage_policies")) - .order_by(Alias::new("id"), sea_orm::sea_query::Order::Asc) - .to_owned(); - db.query_all(&statement) - .await - .map_err(AsterError::from)? - .into_iter() - .map(|row| { - Ok(LegacyStaticCredentialRow { - policy_id: row.try_get_by_index(0).map_err(AsterError::from)?, - access_key: row.try_get_by_index(1).map_err(AsterError::from)?, - secret_key: row.try_get_by_index(2).map_err(AsterError::from)?, - }) - }) - .collect() -} - -async fn clear_legacy_static_credentials(db: &sea_orm::DatabaseTransaction) -> Result<()> { - if !legacy_static_credential_columns_exist(db).await? { - return Ok(()); - } - let statement = Query::update() - .table(Alias::new("storage_policies")) - .values([ - (Alias::new("access_key"), Expr::value("")), - (Alias::new("secret_key"), Expr::value("")), - ]) - .to_owned(); - db.execute(&statement) - .await - .map(|_| ()) - .map_err(AsterError::from) -} - -async fn legacy_static_credential_columns_exist(db: &sea_orm::DatabaseTransaction) -> Result { - let manager = SchemaManager::new(db); - let has_access_key = manager - .has_column("storage_policies", "access_key") - .await - .map_err(AsterError::from)?; - let has_secret_key = manager - .has_column("storage_policies", "secret_key") - .await - .map_err(AsterError::from)?; - match (has_access_key, has_secret_key) { - (true, true) => Ok(true), - (false, false) => Ok(false), - _ => Err(AsterError::database_operation( - "legacy storage policy static credential columns are partially present", - )), - } -} - -#[cfg(test)] -mod tests { - use super::*; - use chrono::Utc; - use sea_orm::{ - ActiveModelTrait, DatabaseConnection, Set, TransactionTrait, sea_query::ExprTrait, - }; - use serde::Serialize; - - use aster_drive_model::types::{ - MicrosoftGraphCloud, StorageCredentialKind, StorageCredentialProvider, - StorageCredentialStatus, - }; - use aster_drive_storage::{ - ConnectorConfigEnvelope, ConnectorId, StoragePolicyBehaviorConfig, - encode_storage_policy_config, - }; - - use crate::storage::connectors::{ - OneDriveAccountMode, OneDriveCredentialV1, builtin_storage_connector_registry, - encrypt_application_client_secret, - }; - - const KEY: &str = "legacy-storage-credential-test-key-32bytes"; - const OTHER_KEY: &str = "different-storage-credential-key-32bytes"; - - #[derive(Serialize)] - struct EmptyTestConnectorConfig {} - - #[derive(Serialize)] - struct TestOneDriveMetadata<'a> { - cloud: MicrosoftGraphCloud, - drive_id: &'a str, - root_item_id: &'a str, - root_item_name: &'a str, - id_token: &'a str, - } - - async fn database() -> DatabaseConnection { - let db = crate::db::connect_with_metrics( - &crate::config::DatabaseConfig { - url: "sqlite::memory:".into(), - pool_size: 1, - retry_count: 0, - }, - aster_drive_metrics::NoopMetrics::arc(), - ) - .await - .expect("credential migration test database should connect"); - aster_drive_migration::Migrator::up(&db, None) - .await - .expect("credential migration test schema should migrate"); - db - } - - fn config(encryption_key: &str) -> Config { - let mut config = Config::default(); - config.auth.storage_credential_secret_key = encryption_key.to_string(); - config - } - - fn onedrive_config() -> serde_json::Value { - let policy = crate::storage::connectors::test_support::onedrive_policy( - OneDriveAccountMode::Personal, - Some("drive-id".to_string()), - None, - None, - StoragePolicyBehaviorConfig::default(), - ); - aster_drive_storage::decode_storage_policy_config::( - policy.storage_config.as_ref(), - &ConnectorId::declared("asterdrive.storage.onedrive"), - 1, - ) - .expect("typed OneDrive test policy should decode") - .0 - } - - async fn insert_policy( - db: &DatabaseConnection, - id: i64, - connector_id: &str, - connector_config: T, - ) { - let connector_config = - serde_json::to_value(connector_config).expect("test connector config should serialize"); - let storage_config = encode_storage_policy_config( - ConnectorConfigEnvelope::new(ConnectorId::declared(connector_id), 1, connector_config), - StoragePolicyBehaviorConfig::default(), - ) - .expect("test storage policy config should encode"); - // Keep the 0.5.0 migration fixture compatible with databases created - // before the legacy policy columns are removed in 0.6.0. The current - // production entity deliberately has no fields for these columns. - let driver_type = connector_id - .rsplit('.') - .next() - .expect("connector id should contain a driver suffix"); - let now = Utc::now(); - let statement = Query::insert() - .into_table(Alias::new("storage_policies")) - .columns([ - Alias::new("id"), - Alias::new("name"), - Alias::new("driver_type"), - Alias::new("endpoint"), - Alias::new("bucket"), - Alias::new("access_key"), - Alias::new("secret_key"), - Alias::new("base_path"), - Alias::new("remote_node_id"), - Alias::new("remote_storage_target_key"), - Alias::new("max_file_size"), - Alias::new("allowed_types"), - Alias::new("options"), - Alias::new("is_default"), - Alias::new("chunk_size"), - Alias::new("created_at"), - Alias::new("updated_at"), - Alias::new("connector_id"), - Alias::new("storage_config"), - ]) - .values([ - Expr::value(id), - Expr::value(format!("policy-{id}")), - Expr::value(driver_type), - Expr::value(""), - Expr::value(""), - Expr::value(""), - Expr::value(""), - Expr::value(""), - Expr::value(Option::::None), - Expr::value(Option::::None), - Expr::value(0_i64), - Expr::value("[]"), - Expr::value("{}"), - Expr::value(false), - Expr::value(0_i64), - Expr::value(now), - Expr::value(now), - Expr::value(connector_id), - Expr::value(storage_config), - ]) - .expect("test storage policy insert values should be valid") - .to_owned(); - db.execute(&statement) - .await - .expect("test storage policy should insert"); - } - - async fn set_legacy_static_credential( - db: &DatabaseConnection, - policy_id: i64, - access_key: &str, - secret_key: &str, - ) { - let statement = Query::update() - .table(Alias::new("storage_policies")) - .values([ - (Alias::new("access_key"), Expr::value(access_key)), - (Alias::new("secret_key"), Expr::value(secret_key)), - ]) - .and_where(Expr::col(Alias::new("id")).eq(policy_id)) - .to_owned(); - db.execute(&statement) - .await - .expect("legacy static credential should update"); - } - - async fn insert_legacy_application( - db: &DatabaseConnection, - policy_id: i64, - encryption_key: &str, - provider: StorageCredentialProvider, - ciphertext: Option, - ) { - let now = Utc::now(); - let ciphertext = ciphertext.or_else(|| { - Some( - encrypt_application_client_secret(encryption_key, policy_id, "client-secret") - .expect("legacy application secret should encrypt"), - ) - }); - storage_connector_application_config::ActiveModel { - policy_id: Set(policy_id), - provider: Set(provider), - tenant_id: Set(Some(" common ".to_string())), - scopes: Set( - serde_json::to_string(&vec!["offline_access", "Files.ReadWrite"]) - .expect("legacy application scopes should serialize"), - ), - client_id: Set(Some(" client-id ".to_string())), - client_secret_ciphertext: Set(ciphertext), - metadata: Set(serde_json::to_string( - &serde_json::Map::::new(), - ) - .expect("legacy application metadata should serialize")), - created_at: Set(now), - updated_at: Set(now), - ..Default::default() - } - .insert(db) - .await - .expect("legacy application credential should insert"); - } - - async fn insert_legacy_authorization( - db: &DatabaseConnection, - policy_id: i64, - encryption_key: &str, - provider: StorageCredentialProvider, - credential_kind: StorageCredentialKind, - access_ciphertext: Option, - ) { - let now = Utc::now(); - let access_ciphertext = access_ciphertext.or_else(|| { - Some( - crate::services::storage_policy::credential::crypto::encrypt_token( - encryption_key, - crate::services::storage_policy::credential::crypto::token_aad( - policy_id, - StorageCredentialProvider::MicrosoftGraph.as_str(), - "access", - ) - .as_bytes(), - "access-token", - ) - .expect("legacy access token should encrypt"), - ) - }); - let refresh_ciphertext = - crate::services::storage_policy::credential::crypto::encrypt_token( - encryption_key, - crate::services::storage_policy::credential::crypto::token_aad( - policy_id, - StorageCredentialProvider::MicrosoftGraph.as_str(), - "refresh", - ) - .as_bytes(), - "refresh-token", - ) - .expect("legacy refresh token should encrypt"); - let metadata = serde_json::to_string(&TestOneDriveMetadata { - cloud: MicrosoftGraphCloud::Global, - drive_id: "drive-id", - root_item_id: "root-item-id", - root_item_name: "Documents", - id_token: "***REDACTED***", - }) - .expect("legacy authorization metadata should serialize"); - storage_policy_credential::ActiveModel { - policy_id: Set(policy_id), - provider: Set(provider), - credential_kind: Set(credential_kind), - account_label: Set(Some(" Documents ".to_string())), - subject: Set(Some(" subject-id ".to_string())), - tenant_id: Set(Some(" common ".to_string())), - scopes: Set( - serde_json::to_string(&vec!["offline_access", "Files.ReadWrite"]) - .expect("legacy authorization scopes should serialize"), - ), - access_token_ciphertext: Set(access_ciphertext), - refresh_token_ciphertext: Set(Some(refresh_ciphertext)), - metadata: Set(metadata), - status: Set(StorageCredentialStatus::Authorized), - status_reason: Set(None), - expires_at: Set(Some(now + chrono::Duration::hours(1))), - authorized_at: Set(Some(now)), - last_refreshed_at: Set(None), - last_validated_at: Set(Some(now)), - created_at: Set(now), - updated_at: Set(now), - ..Default::default() - } - .insert(db) - .await - .expect("legacy authorization credential should insert"); - } - - async fn stored_payload( - db: &DatabaseConnection, - encryption_key: &str, - policy_id: i64, - connector_id: &str, - ) -> serde_json::Value { - let record = - crate::db::repository::storage_policy_connector_credential_repo::find_by_policy( - db, policy_id, - ) - .await - .expect("connector credential lookup should succeed") - .expect("connector credential should exist"); - crate::storage::connectors::decode_connector_credential( - encryption_key, - &record, - &ConnectorId::declared(connector_id), - 1, - ) - .expect("connector credential should decrypt") - } - - async fn run_migration( - db: &DatabaseConnection, - config: &Config, - connectors: &StorageConnectorRegistry, - ) -> Result { - let transaction = db - .begin() - .await - .expect("credential migration transaction should begin"); - match migrate_legacy_storage_credentials(&transaction, config, connectors).await { - Ok(report) => { - transaction - .commit() - .await - .expect("successful credential migration should commit"); - Ok(report) - } - Err(error) => { - transaction - .rollback() - .await - .expect("failed credential migration should roll back"); - Err(error) - } - } - } - - async fn legacy_static_rows(db: &DatabaseConnection) -> Vec { - let transaction = db - .begin() - .await - .expect("legacy static verification transaction should begin"); - let rows = load_legacy_static_credentials(&transaction) - .await - .expect("legacy static rows should load"); - transaction - .rollback() - .await - .expect("read-only legacy static verification should roll back"); - rows - } - - async fn assert_legacy_static_cleared(db: &DatabaseConnection) { - let rows = legacy_static_rows(db).await; - assert!( - rows.iter() - .all(|row| row.access_key.is_empty() && row.secret_key.is_empty()) - ); - } - - #[tokio::test] - async fn slim_schema_without_static_credential_columns_is_idempotent() { - let db = database().await; - db.execute_unprepared("ALTER TABLE storage_policies DROP COLUMN access_key") - .await - .expect("test slim schema should drop access_key"); - db.execute_unprepared("ALTER TABLE storage_policies DROP COLUMN secret_key") - .await - .expect("test slim schema should drop secret_key"); - let connectors = builtin_storage_connector_registry().unwrap(); - - let first = run_migration(&db, &config(KEY), &connectors) - .await - .expect("slim schema should skip legacy static credential scan"); - let second = run_migration(&db, &config(KEY), &connectors) - .await - .expect("repeated slim-schema migration should remain idempotent"); - - assert_eq!(first, LegacyStorageCredentialMigrationReport::default()); - assert_eq!(second, LegacyStorageCredentialMigrationReport::default()); - let manager = SchemaManager::new(&db); - assert!( - !manager - .has_column("storage_policies", "access_key") - .await - .unwrap() - ); - assert!( - !manager - .has_column("storage_policies", "secret_key") - .await - .unwrap() - ); - } - - #[tokio::test] - async fn partially_removed_static_credential_columns_abort_migration() { - let db = database().await; - db.execute_unprepared("ALTER TABLE storage_policies DROP COLUMN access_key") - .await - .expect("test schema should drop only the legacy access key column"); - - let error = run_migration( - &db, - &config(KEY), - &builtin_storage_connector_registry().unwrap(), - ) - .await - .expect_err("partially finalized static credential columns must abort"); - - assert!(error.to_string().contains("partially present")); - let manager = SchemaManager::new(&db); - assert!( - !manager - .has_column("storage_policies", "access_key") - .await - .unwrap() - ); - assert!( - manager - .has_column("storage_policies", "secret_key") - .await - .unwrap() - ); - } - - #[tokio::test] - async fn migrates_all_static_connector_credentials_and_clears_legacy_columns() { - let db = database().await; - let connectors = builtin_storage_connector_registry().unwrap(); - let cases = [ - ( - 1, - "asterdrive.storage.s3", - "s3_access_key_id", - "s3_secret_access_key", - ), - ( - 2, - "asterdrive.storage.sftp", - "sftp_username", - "sftp_password", - ), - ( - 3, - "asterdrive.storage.azure_blob", - "azure_blob_account_name", - "azure_blob_account_key", - ), - ( - 4, - "asterdrive.storage.tencent_cos", - "tencent_cos_secret_id", - "tencent_cos_secret_key", - ), - ]; - for (policy_id, connector_id, _, _) in cases { - insert_policy(&db, policy_id, connector_id, EmptyTestConnectorConfig {}).await; - set_legacy_static_credential(&db, policy_id, " legacy-id ", " legacy-secret ").await; - } - - let report = run_migration(&db, &config(KEY), &connectors).await.unwrap(); - assert_eq!(report.migrated, cases.len()); - assert_eq!(report.already_current, 0); - for (policy_id, connector_id, id_field, secret_field) in cases { - let payload = stored_payload(&db, KEY, policy_id, connector_id).await; - assert_eq!(payload[id_field], "legacy-id"); - assert_eq!(payload[secret_field], "legacy-secret"); - assert!(payload.get("access_key").is_none()); - assert!(payload.get("secret_key").is_none()); - } - assert_legacy_static_cleared(&db).await; - } - - #[tokio::test] - async fn migrates_onedrive_application_without_authorization() { - let db = database().await; - insert_policy(&db, 1, "asterdrive.storage.onedrive", onedrive_config()).await; - insert_legacy_application(&db, 1, KEY, StorageCredentialProvider::MicrosoftGraph, None) - .await; - - let report = run_migration( - &db, - &config(KEY), - &builtin_storage_connector_registry().unwrap(), - ) - .await - .unwrap(); - assert_eq!(report.migrated, 1); - assert_eq!(report.legacy_rows_deleted, 1); - let payload: OneDriveCredentialV1 = serde_json::from_value( - stored_payload(&db, KEY, 1, "asterdrive.storage.onedrive").await, - ) - .unwrap(); - assert_eq!(payload.application.client_id, "client-id"); - assert_eq!(payload.application.client_secret, "client-secret"); - assert!(payload.authorization.is_none()); - assert!( - storage_connector_application_config::Entity::find() - .all(&db) - .await - .unwrap() - .is_empty() - ); - } - - #[tokio::test] - async fn merges_onedrive_application_and_oauth_credentials() { - let db = database().await; - insert_policy(&db, 1, "asterdrive.storage.onedrive", onedrive_config()).await; - insert_legacy_application(&db, 1, KEY, StorageCredentialProvider::MicrosoftGraph, None) - .await; - insert_legacy_authorization( - &db, - 1, - KEY, - StorageCredentialProvider::MicrosoftGraph, - StorageCredentialKind::OauthDelegated, - None, - ) - .await; - - let report = run_migration( - &db, - &config(KEY), - &builtin_storage_connector_registry().unwrap(), - ) - .await - .unwrap(); - assert_eq!(report.migrated, 1); - assert_eq!(report.legacy_rows_deleted, 2); - let payload: OneDriveCredentialV1 = serde_json::from_value( - stored_payload(&db, KEY, 1, "asterdrive.storage.onedrive").await, - ) - .unwrap(); - let authorization = payload.authorization.unwrap(); - assert_eq!(authorization.access_token, "access-token"); - assert_eq!( - authorization.refresh_token.as_deref(), - Some("refresh-token") - ); - assert_eq!(authorization.metadata.drive_id, "drive-id"); - assert_eq!(authorization.metadata.root_item_id, "root-item-id"); - assert!(authorization.metadata.id_token_present); - assert_eq!(authorization.account_label.as_deref(), Some("Documents")); - assert_eq!(authorization.subject.as_deref(), Some("subject-id")); - } - - #[tokio::test] - async fn rejects_onedrive_oauth_without_application_and_preserves_legacy_row() { - let db = database().await; - insert_policy(&db, 1, "asterdrive.storage.onedrive", onedrive_config()).await; - insert_legacy_authorization( - &db, - 1, - KEY, - StorageCredentialProvider::MicrosoftGraph, - StorageCredentialKind::OauthDelegated, - None, - ) - .await; - - let error = run_migration( - &db, - &config(KEY), - &builtin_storage_connector_registry().unwrap(), - ) - .await - .unwrap_err(); - assert!( - error - .to_string() - .contains("without application credentials") - ); - assert_eq!( - storage_policy_credential::Entity::find() - .all(&db) - .await - .unwrap() - .len(), - 1 - ); - assert!( - crate::db::repository::storage_policy_connector_credential_repo::find_by_policy(&db, 1) - .await - .unwrap() - .is_none() - ); - } - - #[tokio::test] - async fn rejects_onedrive_provider_and_kind_mismatches() { - for (provider, credential_kind, expected) in [ - ( - StorageCredentialProvider::GoogleDrive, - StorageCredentialKind::OauthDelegated, - "application provider", - ), - ( - StorageCredentialProvider::MicrosoftGraph, - StorageCredentialKind::ServiceAccount, - "authorization provider", - ), - ] { - let db = database().await; - insert_policy(&db, 1, "asterdrive.storage.onedrive", onedrive_config()).await; - insert_legacy_application(&db, 1, KEY, provider, None).await; - if provider == StorageCredentialProvider::MicrosoftGraph { - insert_legacy_authorization(&db, 1, KEY, provider, credential_kind, None).await; - } - - let error = run_migration( - &db, - &config(KEY), - &builtin_storage_connector_registry().unwrap(), - ) - .await - .unwrap_err(); - assert!(error.to_string().contains(expected), "{error}"); - } - } - - #[tokio::test] - async fn rejects_incomplete_static_credentials_but_accepts_empty_columns() { - let db = database().await; - insert_policy(&db, 1, "asterdrive.storage.s3", EmptyTestConnectorConfig {}).await; - set_legacy_static_credential(&db, 1, "only-id", "").await; - let error = run_migration( - &db, - &config(KEY), - &builtin_storage_connector_registry().unwrap(), - ) - .await - .unwrap_err(); - assert!( - error - .to_string() - .contains("incomplete legacy static credentials") - ); - assert_eq!(legacy_static_rows(&db).await[0].access_key, "only-id"); - - set_legacy_static_credential(&db, 1, "", "").await; - let report = run_migration( - &db, - &config(KEY), - &builtin_storage_connector_registry().unwrap(), - ) - .await - .unwrap(); - assert_eq!(report, LegacyStorageCredentialMigrationReport::default()); - assert!( - crate::db::repository::storage_policy_connector_credential_repo::find_by_policy(&db, 1) - .await - .unwrap() - .is_none() - ); - } - - #[tokio::test] - async fn rejects_corrupt_or_wrong_key_onedrive_ciphertext_without_cleanup() { - for (stored_key, startup_key, ciphertext) in [ - (KEY, KEY, Some("not-a-ciphertext".to_string())), - (KEY, OTHER_KEY, None), - ] { - let db = database().await; - insert_policy(&db, 1, "asterdrive.storage.onedrive", onedrive_config()).await; - insert_legacy_application( - &db, - 1, - stored_key, - StorageCredentialProvider::MicrosoftGraph, - ciphertext, - ) - .await; - - let error = run_migration( - &db, - &config(startup_key), - &builtin_storage_connector_registry().unwrap(), - ) - .await - .unwrap_err(); - assert!( - error.to_string().contains("decrypt") || error.to_string().contains("ciphertext") - ); - assert_eq!( - storage_connector_application_config::Entity::find() - .all(&db) - .await - .unwrap() - .len(), - 1 - ); - } - } - - #[tokio::test] - async fn matching_target_is_idempotent_but_conflicting_target_aborts() { - let db = database().await; - insert_policy(&db, 1, "asterdrive.storage.s3", EmptyTestConnectorConfig {}).await; - set_legacy_static_credential(&db, 1, "id-one", "secret-one").await; - let connectors = builtin_storage_connector_registry().unwrap(); - run_migration(&db, &config(KEY), &connectors).await.unwrap(); - - set_legacy_static_credential(&db, 1, "id-one", "secret-one").await; - let report = run_migration(&db, &config(KEY), &connectors).await.unwrap(); - assert_eq!(report.migrated, 0); - assert_eq!(report.already_current, 1); - assert_legacy_static_cleared(&db).await; - - set_legacy_static_credential(&db, 1, "id-two", "secret-two").await; - let error = run_migration(&db, &config(KEY), &connectors) - .await - .unwrap_err(); - assert!(error.to_string().contains("conflicting legacy")); - let payload = stored_payload(&db, KEY, 1, "asterdrive.storage.s3").await; - assert_eq!(payload["s3_access_key_id"], "id-one"); - assert_eq!(legacy_static_rows(&db).await[0].access_key, "id-two"); - } - - #[tokio::test] - async fn failure_rolls_back_prior_policy_import_and_all_cleanup() { - let db = database().await; - insert_policy(&db, 1, "asterdrive.storage.s3", EmptyTestConnectorConfig {}).await; - insert_policy( - &db, - 2, - "asterdrive.storage.sftp", - EmptyTestConnectorConfig {}, - ) - .await; - set_legacy_static_credential(&db, 1, "good-id", "good-secret").await; - set_legacy_static_credential(&db, 2, "broken-user", "").await; - - let error = run_migration( - &db, - &config(KEY), - &builtin_storage_connector_registry().unwrap(), - ) - .await - .unwrap_err(); - assert!( - error - .to_string() - .contains("incomplete legacy static credentials") - ); - assert!( - crate::db::repository::storage_policy_connector_credential_repo::find_all(&db) - .await - .unwrap() - .is_empty() - ); - let rows = legacy_static_rows(&db).await; - assert_eq!(rows[0].access_key, "good-id"); - assert_eq!(rows[0].secret_key, "good-secret"); - assert_eq!(rows[1].access_key, "broken-user"); - } -} diff --git a/src/services/storage_policy/credential/mod.rs b/src/services/storage_policy/credential/mod.rs index 273f33f32..be4787e87 100644 --- a/src/services/storage_policy/credential/mod.rs +++ b/src/services/storage_policy/credential/mod.rs @@ -1,4 +1,4 @@ -//! Connector credential orchestration and startup migration. +//! Connector credential orchestration. //! //! Provider payloads, authorization protocol handling, and refresh state are //! connector-owned. This module keeps only cross-connector persistence, @@ -6,7 +6,6 @@ pub(crate) mod crypto; mod management; -mod migration; mod oauth; pub use management::{ @@ -22,5 +21,3 @@ pub use oauth::{ StorageAuthorizationCallbackOutcome, StorageAuthorizationCallbackQuery, StorageAuthorizationStartResponse, start_authorization, }; - -pub(crate) use migration::migrate_legacy_storage_credentials; diff --git a/src/storage/connectors/azure_blob.rs b/src/storage/connectors/azure_blob.rs index 1540fe7ff..d2d98a030 100644 --- a/src/storage/connectors/azure_blob.rs +++ b/src/storage/connectors/azure_blob.rs @@ -202,20 +202,6 @@ impl StorageConnector for AzureBlobConnector { ) } - fn import_legacy_credential( - &self, - _encryption_key: &str, - _policy: &storage_policy::Model, - input: super::LegacyStorageConnectorCredentialInput, - ) -> Result> { - super::common::import_legacy_static_credential(Self::ID, input, |legacy| { - AzureBlobStaticCredentialsV1 { - azure_blob_account_name: legacy.access_key, - azure_blob_account_key: legacy.secret_key, - } - }) - } - async fn build_draft_driver( &self, context: &super::StorageConnectorContext<'_>, diff --git a/src/storage/connectors/common.rs b/src/storage/connectors/common.rs index 0993ad5f5..9d5f1f844 100644 --- a/src/storage/connectors/common.rs +++ b/src/storage/connectors/common.rs @@ -465,43 +465,6 @@ pub(super) fn merge_saved_static_credential( Ok(StorageConnectorCredentialInput::Static(current)) } -/// Convert the deprecated `access_key`/`secret_key` policy columns into the -/// current connector-owned static credential struct. -/// -/// This helper is exclusive to the AsterDrive 0.5.0 startup migration and will -/// be completely removed together with the legacy columns in AsterDrive 0.6.0. -pub(super) fn import_legacy_static_credential( - connector_id: &str, - input: super::LegacyStorageConnectorCredentialInput, - build: impl FnOnce(super::LegacyStoragePolicyStaticCredential) -> T, -) -> Result> { - if input.application_config.is_some() || input.authorization.is_some() { - return Err(AsterError::database_operation(format!( - "connector '{connector_id}' received incompatible legacy authorization credentials", - ))); - } - let Some(mut credential) = input.static_credential else { - return Ok(None); - }; - credential.access_key = credential.access_key.trim().to_string(); - credential.secret_key = credential.secret_key.trim().to_string(); - if credential.access_key.is_empty() && credential.secret_key.is_empty() { - return Ok(None); - } - if credential.access_key.is_empty() || credential.secret_key.is_empty() { - return Err(AsterError::database_operation(format!( - "connector '{connector_id}' has incomplete legacy static credentials", - ))); - } - serde_json::to_value(build(credential)) - .map(Some) - .map_err(|error| { - AsterError::database_operation(format!( - "serialize migrated credential for connector '{connector_id}': {error}", - )) - }) -} - pub(super) fn decode_normalized_connector_action_input( descriptor: &StorageConnectorActionDescriptor, values: &std::collections::BTreeMap, diff --git a/src/storage/connectors/contract.rs b/src/storage/connectors/contract.rs index d90937606..ea3fd480c 100644 --- a/src/storage/connectors/contract.rs +++ b/src/storage/connectors/contract.rs @@ -23,8 +23,7 @@ use aster_drive_storage::{ConnectorId, MultipartStorageDriver, StorageDriver, St use super::common; use super::models::{ ExecuteDraftStorageConnectorActionInput, ExecuteSavedStorageConnectorActionInput, - LegacyStorageConnectorCredentialInput, LocalFilesystemPolicyProjection, - RemotePolicyBindingProjection, StorageConnectorActionResult, + LocalFilesystemPolicyProjection, RemotePolicyBindingProjection, StorageConnectorActionResult, StorageConnectorAuthorizationCallback, StorageConnectorAuthorizationError, StorageConnectorAuthorizationStart, StorageConnectorCredentialInfo, StorageConnectorCredentialInput, StorageConnectorRuntimeCredential, @@ -345,29 +344,6 @@ pub(crate) trait StorageConnector: Send + Sync { Ok(None) } - /// Convert rows from the deprecated credential stores into this - /// connector's current typed payload during the AsterDrive 0.5.0-only - /// startup migration. - /// - /// The default rejects unexpected legacy data so a missing connector hook - /// stops startup instead of silently discarding credentials. This contract - /// and the deprecated inputs are scheduled for removal in AsterDrive 0.6.0. - fn import_legacy_credential( - &self, - _encryption_key: &str, - policy: &storage_policy::Model, - input: LegacyStorageConnectorCredentialInput, - ) -> Result> { - if input.is_empty() { - return Ok(None); - } - Err(AsterError::database_operation(format!( - "storage policy {} has legacy credentials unsupported by connector '{}'", - policy.id, - self.descriptor().connector_id.as_str(), - ))) - } - async fn load_runtime_credential( &self, _db: &DatabaseConnection, diff --git a/src/storage/connectors/mod.rs b/src/storage/connectors/mod.rs index 64fbb67cb..ce62207bd 100644 --- a/src/storage/connectors/mod.rs +++ b/src/storage/connectors/mod.rs @@ -53,7 +53,6 @@ pub use models::{ TestDraftStorageConnectorConnectionInput, }; pub(crate) use models::{ - LegacyStorageConnectorCredentialInput, LegacyStoragePolicyStaticCredential, LocalFilesystemPolicyProjection, RemotePolicyBindingProjection, StorageAuthorizationFailureReason, StorageConnectorAuthorizationAudit, StorageConnectorAuthorizationCallback, StorageConnectorAuthorizationError, @@ -61,11 +60,9 @@ pub(crate) use models::{ StorageCredentialValidationOutcome, StoragePolicyCleanupDriverSnapshot, StoragePolicyCleanupSnapshots, }; -pub(crate) use onedrive::OneDriveConnector; #[cfg(test)] -pub(crate) use onedrive::{ - OneDriveAccountMode, OneDriveCredentialV1, encrypt_application_client_secret, -}; +pub(crate) use onedrive::OneDriveAccountMode; +pub(crate) use onedrive::OneDriveConnector; use qiniu::QiniuConnector; use remote::RemoteConnector; use s3::S3Connector; diff --git a/src/storage/connectors/models.rs b/src/storage/connectors/models.rs index c076d70c9..4ffdd6692 100644 --- a/src/storage/connectors/models.rs +++ b/src/storage/connectors/models.rs @@ -127,42 +127,6 @@ pub struct StorageConnectorConnectionInput { pub credential: StorageConnectorCredentialInput, } -/// Strongly typed legacy credential rows used only by AsterDrive 0.5.0. -/// -/// The deprecated table models stay outside the normal entity namespace. Core -/// migration code loads them, while each connector owns conversion into its -/// current credential payload. This type and the legacy tables will be -/// completely removed in AsterDrive 0.6.0. -#[derive(Clone, Debug, Default)] -pub(crate) struct LegacyStorageConnectorCredentialInput { - pub static_credential: Option, - #[expect( - deprecated, - reason = "AsterDrive 0.5.0 migration input is removed with legacy application credentials in 0.6.0" - )] - pub application_config: - Option, - #[expect( - deprecated, - reason = "AsterDrive 0.5.0 migration input is removed with legacy authorization credentials in 0.6.0" - )] - pub authorization: Option, -} - -impl LegacyStorageConnectorCredentialInput { - pub(crate) fn is_empty(&self) -> bool { - self.static_credential.is_none() - && self.application_config.is_none() - && self.authorization.is_none() - } -} - -#[derive(Clone, Debug, PartialEq, Eq)] -pub(crate) struct LegacyStoragePolicyStaticCredential { - pub access_key: String, - pub secret_key: String, -} - #[derive(Clone)] pub(crate) struct StorageConnectorRuntimeCredential { connector_id: aster_drive_storage::ConnectorId, diff --git a/src/storage/connectors/onedrive.rs b/src/storage/connectors/onedrive.rs index b9ba85251..3ed3d8dc3 100644 --- a/src/storage/connectors/onedrive.rs +++ b/src/storage/connectors/onedrive.rs @@ -53,9 +53,6 @@ mod localization; mod oauth; mod provider; -#[cfg(test)] -pub(crate) use oauth::encrypt_application_client_secret; - const AUTHORIZATION_FLOW_TTL_SECS: u64 = 300; const CLEANUP_SNAPSHOT_SCHEMA_VERSION: u32 = 1; @@ -106,22 +103,6 @@ struct OneDriveCleanupSnapshotV1 { expires_at: Option>, } -#[derive(Debug, Default, Deserialize)] -struct LegacyOneDriveMetadata { - #[serde(default)] - cloud: Option, - #[serde(default)] - drive_id: Option, - #[serde(default)] - root_item_id: Option, - #[serde(default)] - root_item_name: Option, - #[serde(default)] - id_token_present: bool, - #[serde(default)] - id_token: Option, -} - aster_drive_storage::storage_connector_schema! { pub struct OneDriveConnectorConfigV1 { config { @@ -1083,109 +1064,6 @@ impl StorageConnector for OneDriveConnector { ) } - /// AsterDrive 0.5.0-only legacy import; remove with the deprecated stores - /// and trait hook in AsterDrive 0.6.0. - #[expect( - deprecated, - reason = "AsterDrive 0.5.0 OneDrive credential import is removed in 0.6.0" - )] - fn import_legacy_credential( - &self, - encryption_key: &str, - policy: &storage_policy::Model, - input: super::LegacyStorageConnectorCredentialInput, - ) -> Result> { - if input.static_credential.is_some() { - return Err(AsterError::database_operation(format!( - "OneDrive storage policy {} contains incompatible legacy static credentials", - policy.id - ))); - } - - let Some(application) = input.application_config else { - if input.authorization.is_some() { - return Err(AsterError::database_operation(format!( - "OneDrive storage policy {} has legacy OAuth credentials without application credentials", - policy.id - ))); - } - return Ok(None); - }; - if application.provider != StorageCredentialProvider::MicrosoftGraph { - return Err(AsterError::database_operation(format!( - "OneDrive storage policy {} has incompatible legacy application provider '{}'", - policy.id, - application.provider.as_str() - ))); - } - - let connector_config = Self::decode_config(policy)?; - let client_id = required_legacy_value( - application.client_id, - policy.id, - "Microsoft Graph client_id", - )?; - let client_secret_ciphertext = required_legacy_value( - application.client_secret_ciphertext, - policy.id, - "Microsoft Graph client_secret ciphertext", - )?; - let client_secret = oauth::decrypt_application_client_secret( - encryption_key, - policy.id, - &client_secret_ciphertext, - )?; - let client_secret = required_legacy_value( - Some(client_secret.expose_secret().to_string()), - policy.id, - "Microsoft Graph client_secret", - )?; - let tenant = normalized_legacy_value(application.tenant_id) - .or_else(|| normalized_legacy_value(connector_config.tenant.clone())) - .unwrap_or_else(|| "common".to_string()); - let default_scopes = default_microsoft_graph_scopes(&connector_config); - let application_scopes = parse_legacy_scopes( - &application.scopes, - policy.id, - "Microsoft Graph application scopes", - )?; - let application_scopes = normalize_microsoft_graph_scopes( - (!application_scopes.is_empty()).then_some(application_scopes), - default_scopes, - ); - let application = OneDriveApplicationCredentialV1 { - cloud: connector_config.cloud, - tenant: tenant.clone(), - client_id, - client_secret, - scopes: application_scopes.clone(), - }; - - let authorization = input - .authorization - .map(|authorization| { - import_legacy_onedrive_authorization( - encryption_key, - policy, - &connector_config, - &application, - authorization, - ) - }) - .transpose()?; - serde_json::to_value(OneDriveCredentialV1 { - application, - authorization, - }) - .map(Some) - .map_err(|error| { - AsterError::database_operation(format!( - "serialize migrated OneDrive credential for policy {}: {error}", - policy.id - )) - }) - } - async fn persist_credential( &self, db: &sea_orm::DatabaseTransaction, @@ -1933,167 +1811,6 @@ fn non_empty_string(value: String) -> Option { if value.is_empty() { None } else { Some(value) } } -fn normalized_legacy_value(value: Option) -> Option { - value.and_then(non_empty_string) -} - -fn required_legacy_value(value: Option, policy_id: i64, field: &str) -> Result { - normalized_legacy_value(value).ok_or_else(|| { - AsterError::database_operation(format!( - "OneDrive storage policy {policy_id} is missing legacy {field}" - )) - }) -} - -fn parse_legacy_scopes(raw: &str, policy_id: i64, field: &str) -> Result> { - if raw.trim().is_empty() { - return Ok(Vec::new()); - } - let scopes = serde_json::from_str::>(raw).map_err(|error| { - AsterError::database_operation(format!( - "OneDrive storage policy {policy_id} has invalid legacy {field}: {error}" - )) - })?; - Ok(scopes - .into_iter() - .filter_map(non_empty_string) - .fold(Vec::new(), |mut normalized, scope| { - if !normalized.contains(&scope) { - normalized.push(scope); - } - normalized - })) -} - -/// AsterDrive 0.5.0-only OAuth row conversion; remove in AsterDrive 0.6.0. -#[expect( - deprecated, - reason = "AsterDrive 0.5.0 OneDrive authorization conversion is removed in 0.6.0" -)] -fn import_legacy_onedrive_authorization( - encryption_key: &str, - policy: &storage_policy::Model, - connector_config: &OneDriveConnectorConfigV1, - application: &OneDriveApplicationCredentialV1, - authorization: aster_drive_model::deprecated::storage_policy_credential::Model, -) -> Result { - if authorization.provider != StorageCredentialProvider::MicrosoftGraph - || authorization.credential_kind != StorageCredentialKind::OauthDelegated - { - return Err(AsterError::database_operation(format!( - "OneDrive storage policy {} has incompatible legacy authorization provider '{}' and kind '{}'", - policy.id, - authorization.provider.as_str(), - authorization.credential_kind.as_str() - ))); - } - let access_token_ciphertext = required_legacy_value( - authorization.access_token_ciphertext, - policy.id, - "Microsoft Graph access token ciphertext", - )?; - let access_token = crate::services::storage_policy::credential::crypto::decrypt_token( - encryption_key, - crate::services::storage_policy::credential::crypto::token_aad( - policy.id, - StorageCredentialProvider::MicrosoftGraph.as_str(), - "access", - ) - .as_bytes(), - &access_token_ciphertext, - )?; - let access_token = required_legacy_value( - Some(access_token), - policy.id, - "Microsoft Graph access token", - )?; - let refresh_token = authorization - .refresh_token_ciphertext - .map(|ciphertext| { - let ciphertext = required_legacy_value( - Some(ciphertext), - policy.id, - "Microsoft Graph refresh token ciphertext", - )?; - crate::services::storage_policy::credential::crypto::decrypt_token( - encryption_key, - crate::services::storage_policy::credential::crypto::token_aad( - policy.id, - StorageCredentialProvider::MicrosoftGraph.as_str(), - "refresh", - ) - .as_bytes(), - &ciphertext, - ) - .and_then(|value| { - required_legacy_value(Some(value), policy.id, "Microsoft Graph refresh token") - }) - }) - .transpose()?; - let metadata: LegacyOneDriveMetadata = - serde_json::from_str(&authorization.metadata).map_err(|error| { - AsterError::database_operation(format!( - "OneDrive storage policy {} has invalid legacy authorization metadata: {error}", - policy.id - )) - })?; - if metadata - .cloud - .is_some_and(|cloud| cloud != connector_config.cloud) - { - return Err(AsterError::database_operation(format!( - "OneDrive storage policy {} has conflicting legacy Microsoft Graph cloud", - policy.id - ))); - } - let drive_id = required_legacy_value( - metadata - .drive_id - .or_else(|| connector_config.drive_id.clone()), - policy.id, - "Microsoft Graph drive_id", - )?; - let root_item_id = required_legacy_value( - metadata - .root_item_id - .or_else(|| connector_config.root_item_id.clone()), - policy.id, - "Microsoft Graph root_item_id", - )?; - let scopes = parse_legacy_scopes( - &authorization.scopes, - policy.id, - "Microsoft Graph authorization scopes", - )?; - - Ok(OneDriveAuthorizationCredentialV1 { - account_label: normalized_legacy_value(authorization.account_label), - subject: normalized_legacy_value(authorization.subject), - tenant_id: normalized_legacy_value(authorization.tenant_id) - .or_else(|| Some(application.tenant.clone())), - scopes: if scopes.is_empty() { - application.scopes.clone() - } else { - scopes - }, - access_token, - refresh_token, - metadata: OneDriveAuthorizationMetadataV1 { - cloud: connector_config.cloud, - drive_id, - root_item_id, - root_item_name: normalized_legacy_value(metadata.root_item_name), - id_token_present: metadata.id_token_present || metadata.id_token.is_some(), - }, - status: authorization.status, - status_reason: normalized_legacy_value(authorization.status_reason), - expires_at: authorization.expires_at, - authorized_at: authorization.authorized_at, - last_refreshed_at: authorization.last_refreshed_at, - last_validated_at: authorization.last_validated_at, - }) -} - #[cfg(test)] mod tests { use super::*; diff --git a/src/storage/connectors/s3.rs b/src/storage/connectors/s3.rs index 4ce8ca36e..1deb3381d 100644 --- a/src/storage/connectors/s3.rs +++ b/src/storage/connectors/s3.rs @@ -301,20 +301,6 @@ impl StorageConnector for S3Connector { ) } - fn import_legacy_credential( - &self, - _encryption_key: &str, - _policy: &storage_policy::Model, - input: super::LegacyStorageConnectorCredentialInput, - ) -> Result> { - super::common::import_legacy_static_credential(Self::ID, input, |legacy| { - S3StaticCredentialsV1 { - s3_access_key_id: legacy.access_key, - s3_secret_access_key: legacy.secret_key, - } - }) - } - async fn build_draft_driver( &self, context: &super::StorageConnectorContext<'_>, diff --git a/src/storage/connectors/sftp.rs b/src/storage/connectors/sftp.rs index b276e4e6c..cfb115203 100644 --- a/src/storage/connectors/sftp.rs +++ b/src/storage/connectors/sftp.rs @@ -244,20 +244,6 @@ impl StorageConnector for SftpConnector { ) } - fn import_legacy_credential( - &self, - _encryption_key: &str, - _policy: &storage_policy::Model, - input: super::LegacyStorageConnectorCredentialInput, - ) -> Result> { - super::common::import_legacy_static_credential(Self::ID, input, |legacy| { - SftpStaticCredentialsV1 { - sftp_username: legacy.access_key, - sftp_password: legacy.secret_key, - } - }) - } - async fn build_draft_driver( &self, context: &super::StorageConnectorContext<'_>, diff --git a/src/storage/connectors/tencent_cos.rs b/src/storage/connectors/tencent_cos.rs index bb4ea95da..971251e77 100644 --- a/src/storage/connectors/tencent_cos.rs +++ b/src/storage/connectors/tencent_cos.rs @@ -370,20 +370,6 @@ impl StorageConnector for TencentCosConnector { ) } - fn import_legacy_credential( - &self, - _encryption_key: &str, - _policy: &storage_policy::Model, - input: super::LegacyStorageConnectorCredentialInput, - ) -> Result> { - super::common::import_legacy_static_credential(Self::ID, input, |legacy| { - TencentCosStaticCredentialsV1 { - tencent_cos_secret_id: legacy.access_key, - tencent_cos_secret_key: legacy.secret_key, - } - }) - } - async fn build_draft_driver( &self, context: &super::StorageConnectorContext<'_>, diff --git a/src/storage/connectors/test_support.rs b/src/storage/connectors/test_support.rs index d581b154d..17cd401cd 100644 --- a/src/storage/connectors/test_support.rs +++ b/src/storage/connectors/test_support.rs @@ -19,11 +19,7 @@ use super::remote::{RemoteConnector, RemoteConnectorConfigV1}; use super::s3::{S3Connector, S3ConnectorConfigV1}; use super::{StorageConnector, StorageConnectorConnectionInput, StorageConnectorCredentialInput}; -/// Build the AsterDrive 0.5.x database shape used by connector tests. -/// -/// Historical migrations intentionally retain legacy policy columns for the -/// startup credential importer. Issue #463 removes that compatibility schema in -/// 0.6.0; current entities safely ignore the extra columns. +/// Build the current database shape used by connector tests. pub(crate) async fn migrate_current_storage_test_schema(database: &DatabaseConnection) { aster_drive_migration::Migrator::up(database, None) .await diff --git a/tests/legacy_storage_credential_migration.rs b/tests/legacy_storage_credential_migration.rs deleted file mode 100644 index 3dd8d2928..000000000 --- a/tests/legacy_storage_credential_migration.rs +++ /dev/null @@ -1,630 +0,0 @@ -//! Integration coverage for the AsterDrive 0.5.0-only startup credential migration. -//! -//! Remove this test with the deprecated source stores in AsterDrive 0.6.0. - -#![expect( - deprecated, - reason = "AsterDrive 0.5.0 integration coverage reads deprecated credential stores until 0.6.0" -)] - -use aes_gcm::{ - Aes256Gcm, Nonce, - aead::{Aead, Generate, KeyInit}, -}; -use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; -use hkdf::Hkdf; -use sea_orm::{ - ActiveModelTrait, ColumnTrait, ConnectionTrait, Database, DatabaseConnection, EntityTrait, - QueryFilter, Set, - sea_query::ExprTrait, - sea_query::{Alias, Expr, Query}, -}; -use serde::{Deserialize, Serialize}; -use sha2::Sha256; - -use aster_drive::config::{Config, node_mode::NodeRuntimeMode}; -use aster_drive::runtime::startup::initialize_database_state; -use aster_drive_migration::Migrator; -use aster_drive_model::deprecated::{ - storage_connector_application_config, storage_policy_credential, -}; -use aster_drive_model::entities::storage_policy_connector_credential; -use aster_drive_model::types::{ - MicrosoftGraphCloud, StorageCredentialKind, StorageCredentialProvider, StorageCredentialStatus, -}; -use aster_drive_storage::{ - ConnectorConfigEnvelope, ConnectorId, StoragePolicyBehaviorConfig, encode_storage_policy_config, -}; - -const KEY: &str = "legacy-storage-credential-test-key-32bytes"; -const OTHER_KEY: &str = "different-storage-credential-key-32bytes"; -const STORAGE_CREDENTIAL_INFO: &[u8] = b"asterdrive:storage-credential-token:v1"; - -#[derive(Serialize)] -struct EmptyConnectorConfig {} - -#[derive(Serialize)] -struct OneDriveConfig { - base_path: String, - provider_resumable_upload_strategy: &'static str, - provider_download_strategy: &'static str, - provider_download_filename_mode: &'static str, - cloud: MicrosoftGraphCloud, - account_mode: &'static str, - tenant: Option, - drive_id: Option, - root_item_id: Option, - site_id: Option, - group_id: Option, -} - -#[derive(Serialize)] -struct OneDriveMetadata<'a> { - cloud: MicrosoftGraphCloud, - drive_id: &'a str, - root_item_id: &'a str, - root_item_name: &'a str, - id_token: &'a str, -} - -#[derive(Deserialize, Serialize)] -struct ConnectorCiphertextEnvelope { - format_version: u32, - connector_id: String, - schema_version: u32, - ciphertext: String, -} - -fn test_config(key: &str) -> Config { - let mut config = Config::default(); - config.auth.storage_credential_secret_key = key.to_string(); - config -} - -async fn database() -> DatabaseConnection { - let db = Database::connect("sqlite::memory:") - .await - .expect("credential migration integration database should connect"); - Migrator::up(&db, None) - .await - .expect("credential migration integration schema should migrate"); - db -} - -fn onedrive_config() -> OneDriveConfig { - OneDriveConfig { - base_path: String::new(), - provider_resumable_upload_strategy: "server_relay", - provider_download_strategy: "server_relay", - provider_download_filename_mode: "provider_native", - cloud: MicrosoftGraphCloud::Global, - account_mode: "personal", - tenant: Some("common".to_string()), - drive_id: Some("drive-id".to_string()), - root_item_id: Some("root-item-id".to_string()), - site_id: None, - group_id: None, - } -} - -async fn insert_policy( - db: &DatabaseConnection, - id: i64, - connector_id: &str, - connector_config: T, -) { - let connector_config = serde_json::to_value(connector_config) - .expect("integration connector config should serialize"); - let storage_config = encode_storage_policy_config( - ConnectorConfigEnvelope::new(ConnectorId::declared(connector_id), 1, connector_config), - StoragePolicyBehaviorConfig::default(), - ) - .expect("integration storage policy config should encode"); - // The 0.5.0 fixture intentionally writes the historical columns that are - // still present in an existing database. Production models no longer - // expose these columns; they are removed with the deprecated stores in - // 0.6.0. - let driver_type = connector_id - .rsplit('.') - .next() - .expect("connector id should contain a driver suffix"); - let now = chrono::Utc::now(); - let statement = Query::insert() - .into_table(Alias::new("storage_policies")) - .columns([ - Alias::new("id"), - Alias::new("name"), - Alias::new("driver_type"), - Alias::new("endpoint"), - Alias::new("bucket"), - Alias::new("access_key"), - Alias::new("secret_key"), - Alias::new("base_path"), - Alias::new("remote_node_id"), - Alias::new("remote_storage_target_key"), - Alias::new("max_file_size"), - Alias::new("allowed_types"), - Alias::new("options"), - Alias::new("is_default"), - Alias::new("chunk_size"), - Alias::new("created_at"), - Alias::new("updated_at"), - Alias::new("connector_id"), - Alias::new("storage_config"), - ]) - .values([ - Expr::value(id), - Expr::value(format!("policy-{id}")), - Expr::value(driver_type), - Expr::value(""), - Expr::value(""), - Expr::value(""), - Expr::value(""), - Expr::value(""), - Expr::value(Option::::None), - Expr::value(Option::::None), - Expr::value(0_i64), - Expr::value("[]"), - Expr::value("{}"), - Expr::value(false), - Expr::value(0_i64), - Expr::value(now), - Expr::value(now), - Expr::value(connector_id), - Expr::value(storage_config), - ]) - .expect("integration storage policy insert values should be valid") - .to_owned(); - db.execute(&statement) - .await - .expect("integration storage policy should insert"); -} - -async fn set_static(db: &DatabaseConnection, policy_id: i64, access_key: &str, secret_key: &str) { - let statement = Query::update() - .table(Alias::new("storage_policies")) - .values([ - (Alias::new("access_key"), Expr::value(access_key)), - (Alias::new("secret_key"), Expr::value(secret_key)), - ]) - .and_where(Expr::col(Alias::new("id")).eq(policy_id)) - .to_owned(); - db.execute(&statement) - .await - .expect("integration legacy static credential should update"); -} - -fn cipher(master_key: &str) -> Aes256Gcm { - let hk = Hkdf::::new(None, master_key.trim().as_bytes()); - let mut key = [0_u8; 32]; - hk.expand(STORAGE_CREDENTIAL_INFO, &mut key) - .expect("integration storage credential key should derive"); - Aes256Gcm::new_from_slice(&key).expect("integration AES key should be valid") -} - -fn encrypt_token(master_key: &str, aad: &[u8], plaintext: &str) -> String { - let nonce = Nonce::generate(); - let ciphertext = cipher(master_key) - .encrypt( - &nonce, - aes_gcm::aead::Payload { - msg: plaintext.as_bytes(), - aad, - }, - ) - .expect("integration legacy token should encrypt"); - format!( - "v1:{}:{}", - URL_SAFE_NO_PAD.encode(nonce), - URL_SAFE_NO_PAD.encode(ciphertext) - ) -} - -fn decrypt_token(master_key: &str, aad: &[u8], ciphertext: &str) -> String { - let mut parts = ciphertext.split(':'); - assert_eq!(parts.next(), Some("v1")); - let nonce = URL_SAFE_NO_PAD - .decode(parts.next().expect("ciphertext should contain nonce")) - .expect("ciphertext nonce should decode"); - let encrypted = URL_SAFE_NO_PAD - .decode(parts.next().expect("ciphertext should contain payload")) - .expect("ciphertext payload should decode"); - assert!(parts.next().is_none()); - let nonce = Nonce::try_from(nonce.as_slice()).expect("ciphertext nonce should be 12 bytes"); - String::from_utf8( - cipher(master_key) - .decrypt( - &nonce, - aes_gcm::aead::Payload { - msg: &encrypted, - aad, - }, - ) - .expect("connector credential should decrypt"), - ) - .expect("connector credential should be UTF-8") -} - -fn token_aad(policy_id: i64, token_name: &str) -> String { - format!("storage_policy_credential:{policy_id}:microsoft_graph:{token_name}") -} - -fn encrypt_connector_payload( - key: &str, - policy_id: i64, - connector_id: &str, - payload: &serde_json::Value, -) -> String { - let aad = format!("storage_policy_connector_credential:{policy_id}:{connector_id}:1"); - let ciphertext = encrypt_token( - key, - aad.as_bytes(), - &serde_json::to_string(payload).expect("current connector payload should serialize"), - ); - serde_json::to_string(&ConnectorCiphertextEnvelope { - format_version: 1, - connector_id: connector_id.to_string(), - schema_version: 1, - ciphertext, - }) - .expect("current connector ciphertext envelope should serialize") -} - -async fn insert_current_connector_credential( - db: &DatabaseConnection, - key: &str, - policy_id: i64, - connector_id: &str, - payload: serde_json::Value, -) { - let ciphertext = encrypt_connector_payload(key, policy_id, connector_id, &payload); - aster_drive::db::repository::storage_policy_connector_credential_repo::upsert( - db, - policy_id, - connector_id.to_string(), - 1, - ciphertext, - ) - .await - .expect("current connector credential should insert"); -} - -fn application_secret_aad(policy_id: i64) -> String { - format!("storage_connector_application_config:{policy_id}:microsoft_graph:client_secret") -} - -async fn insert_onedrive_application( - db: &DatabaseConnection, - policy_id: i64, - key: &str, - ciphertext: Option, -) { - let now = chrono::Utc::now(); - storage_connector_application_config::ActiveModel { - policy_id: Set(policy_id), - provider: Set(StorageCredentialProvider::MicrosoftGraph), - tenant_id: Set(Some(" common ".to_string())), - scopes: Set(serde_json::to_string(&vec!["offline_access", "Files.ReadWrite"]).unwrap()), - client_id: Set(Some(" client-id ".to_string())), - client_secret_ciphertext: Set(Some(ciphertext.unwrap_or_else(|| { - encrypt_token( - key, - application_secret_aad(policy_id).as_bytes(), - "client-secret", - ) - }))), - metadata: Set( - serde_json::to_string(&serde_json::Map::::new()).unwrap(), - ), - created_at: Set(now), - updated_at: Set(now), - ..Default::default() - } - .insert(db) - .await - .expect("integration legacy application should insert"); -} - -async fn insert_onedrive_authorization(db: &DatabaseConnection, policy_id: i64, key: &str) { - let now = chrono::Utc::now(); - storage_policy_credential::ActiveModel { - policy_id: Set(policy_id), - provider: Set(StorageCredentialProvider::MicrosoftGraph), - credential_kind: Set(StorageCredentialKind::OauthDelegated), - account_label: Set(Some(" Documents ".to_string())), - subject: Set(Some(" subject-id ".to_string())), - tenant_id: Set(Some(" common ".to_string())), - scopes: Set(serde_json::to_string(&vec!["offline_access", "Files.ReadWrite"]).unwrap()), - access_token_ciphertext: Set(Some(encrypt_token( - key, - token_aad(policy_id, "access").as_bytes(), - "access-token", - ))), - refresh_token_ciphertext: Set(Some(encrypt_token( - key, - token_aad(policy_id, "refresh").as_bytes(), - "refresh-token", - ))), - metadata: Set(serde_json::to_string(&OneDriveMetadata { - cloud: MicrosoftGraphCloud::Global, - drive_id: "drive-id", - root_item_id: "root-item-id", - root_item_name: "Documents", - id_token: "***REDACTED***", - }) - .unwrap()), - status: Set(StorageCredentialStatus::Authorized), - status_reason: Set(None), - expires_at: Set(Some(now + chrono::Duration::hours(1))), - authorized_at: Set(Some(now)), - last_refreshed_at: Set(None), - last_validated_at: Set(Some(now)), - created_at: Set(now), - updated_at: Set(now), - ..Default::default() - } - .insert(db) - .await - .expect("integration legacy authorization should insert"); -} - -async fn connector_payload( - db: &DatabaseConnection, - key: &str, - policy_id: i64, - connector_id: &str, -) -> serde_json::Value { - let record = storage_policy_connector_credential::Entity::find() - .filter(storage_policy_connector_credential::Column::PolicyId.eq(policy_id)) - .one(db) - .await - .expect("connector credential query should succeed") - .expect("connector credential should exist"); - let envelope: ConnectorCiphertextEnvelope = serde_json::from_str(&record.ciphertext) - .expect("connector ciphertext envelope should parse"); - assert_eq!(envelope.format_version, 1); - assert_eq!(envelope.connector_id, connector_id); - assert_eq!(envelope.schema_version, 1); - let aad = format!("storage_policy_connector_credential:{policy_id}:{connector_id}:1"); - serde_json::from_str(&decrypt_token(key, aad.as_bytes(), &envelope.ciphertext)) - .expect("connector payload should be JSON") -} - -async fn legacy_static_rows(db: &DatabaseConnection) -> Vec<(i64, String, String)> { - let statement = Query::select() - .columns([ - Alias::new("id"), - Alias::new("access_key"), - Alias::new("secret_key"), - ]) - .from(Alias::new("storage_policies")) - .order_by(Alias::new("id"), sea_orm::sea_query::Order::Asc) - .to_owned(); - db.query_all(&statement) - .await - .expect("legacy static rows should query") - .into_iter() - .map(|row| { - ( - row.try_get_by_index(0).unwrap(), - row.try_get_by_index(1).unwrap(), - row.try_get_by_index(2).unwrap(), - ) - }) - .collect() -} - -async fn assert_legacy_static_columns_retained_and_cleared(db: &DatabaseConnection) { - let manager = aster_drive_migration::SchemaManager::new(db); - assert!( - manager - .has_column("storage_policies", "access_key") - .await - .unwrap() - ); - assert!( - manager - .has_column("storage_policies", "secret_key") - .await - .unwrap() - ); - assert!( - legacy_static_rows(db) - .await - .iter() - .all(|(_, access_key, secret_key)| access_key.is_empty() && secret_key.is_empty()) - ); -} - -#[tokio::test] -async fn startup_migrates_all_static_connectors_with_typed_field_names() { - let config = test_config(KEY); - let db = database().await; - let cases = [ - ( - 1, - "asterdrive.storage.s3", - "s3_access_key_id", - "s3_secret_access_key", - ), - ( - 2, - "asterdrive.storage.sftp", - "sftp_username", - "sftp_password", - ), - ( - 3, - "asterdrive.storage.azure_blob", - "azure_blob_account_name", - "azure_blob_account_key", - ), - ( - 4, - "asterdrive.storage.tencent_cos", - "tencent_cos_secret_id", - "tencent_cos_secret_key", - ), - ]; - for (policy_id, connector_id, _, _) in cases { - insert_policy(&db, policy_id, connector_id, EmptyConnectorConfig {}).await; - set_static(&db, policy_id, " legacy-id ", " legacy-secret ").await; - } - - initialize_database_state(&db, &config, NodeRuntimeMode::Primary) - .await - .unwrap(); - - for (policy_id, connector_id, id_field, secret_field) in cases { - let payload = connector_payload(&db, KEY, policy_id, connector_id).await; - assert_eq!(payload[id_field], "legacy-id"); - assert_eq!(payload[secret_field], "legacy-secret"); - assert!(payload.get("access_key").is_none()); - assert!(payload.get("secret_key").is_none()); - } - assert_legacy_static_columns_retained_and_cleared(&db).await; -} - -#[tokio::test] -async fn startup_merges_onedrive_application_and_oauth_then_cleans_old_tables() { - let config = test_config(KEY); - let db = database().await; - insert_policy(&db, 1, "asterdrive.storage.onedrive", onedrive_config()).await; - insert_onedrive_application(&db, 1, KEY, None).await; - insert_onedrive_authorization(&db, 1, KEY).await; - - initialize_database_state(&db, &config, NodeRuntimeMode::Primary) - .await - .unwrap(); - - let payload = connector_payload(&db, KEY, 1, "asterdrive.storage.onedrive").await; - assert_eq!(payload["application"]["client_id"], "client-id"); - assert_eq!(payload["application"]["client_secret"], "client-secret"); - assert_eq!(payload["authorization"]["access_token"], "access-token"); - assert_eq!(payload["authorization"]["refresh_token"], "refresh-token"); - assert_eq!(payload["authorization"]["metadata"]["drive_id"], "drive-id"); - assert_eq!( - payload["authorization"]["metadata"]["root_item_id"], - "root-item-id" - ); - assert_eq!( - payload["authorization"]["metadata"]["id_token_present"], - true - ); - assert!( - storage_connector_application_config::Entity::find() - .all(&db) - .await - .unwrap() - .is_empty() - ); - assert!( - storage_policy_credential::Entity::find() - .all(&db) - .await - .unwrap() - .is_empty() - ); -} - -#[tokio::test] -async fn startup_rejects_wrong_key_and_rolls_back_prior_static_import() { - let config = test_config(OTHER_KEY); - let db = database().await; - insert_policy(&db, 1, "asterdrive.storage.s3", EmptyConnectorConfig {}).await; - set_static(&db, 1, "good-id", "good-secret").await; - insert_policy(&db, 2, "asterdrive.storage.onedrive", onedrive_config()).await; - insert_onedrive_application(&db, 2, KEY, None).await; - - let error = initialize_database_state(&db, &config, NodeRuntimeMode::Primary) - .await - .err() - .expect("wrong encryption key should abort startup migration"); - assert!( - error.to_string().contains("decrypt") || error.to_string().contains("ciphertext"), - "{error}" - ); - assert!( - storage_policy_connector_credential::Entity::find() - .all(&db) - .await - .unwrap() - .is_empty() - ); - let rows = legacy_static_rows(&db).await; - assert_eq!(rows[0].1, "good-id"); - assert_eq!(rows[0].2, "good-secret"); - assert_eq!( - storage_connector_application_config::Entity::find() - .all(&db) - .await - .unwrap() - .len(), - 1 - ); -} - -#[tokio::test] -async fn startup_is_idempotent_for_matching_target_and_rejects_conflicts() { - let config = test_config(KEY); - let db = database().await; - insert_policy(&db, 1, "asterdrive.storage.s3", EmptyConnectorConfig {}).await; - set_static(&db, 1, "id-one", "secret-one").await; - insert_current_connector_credential( - &db, - KEY, - 1, - "asterdrive.storage.s3", - serde_json::json!({ - "s3_access_key_id": "id-one", - "s3_secret_access_key": "secret-one", - }), - ) - .await; - initialize_database_state(&db, &config, NodeRuntimeMode::Primary) - .await - .unwrap(); - - initialize_database_state(&db, &config, NodeRuntimeMode::Primary) - .await - .unwrap(); - let record = storage_policy_connector_credential::Entity::find() - .filter(storage_policy_connector_credential::Column::PolicyId.eq(1)) - .one(&db) - .await - .unwrap() - .unwrap(); - assert_eq!(record.revision, 1); - assert_legacy_static_columns_retained_and_cleared(&db).await; - - let conflict_db = database().await; - insert_policy( - &conflict_db, - 1, - "asterdrive.storage.s3", - EmptyConnectorConfig {}, - ) - .await; - set_static(&conflict_db, 1, "id-two", "secret-two").await; - insert_current_connector_credential( - &conflict_db, - KEY, - 1, - "asterdrive.storage.s3", - serde_json::json!({ - "s3_access_key_id": "id-one", - "s3_secret_access_key": "secret-one", - }), - ) - .await; - let error = initialize_database_state(&conflict_db, &config, NodeRuntimeMode::Primary) - .await - .err() - .expect("conflicting credential should abort startup migration"); - assert!(error.to_string().contains("conflicting legacy")); - let payload = connector_payload(&conflict_db, KEY, 1, "asterdrive.storage.s3").await; - assert_eq!(payload["s3_access_key_id"], "id-one"); - let legacy_rows = legacy_static_rows(&conflict_db).await; - assert_eq!(legacy_rows[0].1, "id-two"); - assert_eq!(legacy_rows[0].2, "secret-two"); -} diff --git a/tests/operations/cli.rs b/tests/operations/cli.rs index 3d4c27e25..c8931c1af 100644 --- a/tests/operations/cli.rs +++ b/tests/operations/cli.rs @@ -85,74 +85,6 @@ async fn setup_empty_database_url(prefix: &str) -> String { url } -async fn setup_legacy_storage_upgrade_database_url() -> String { - let url = setup_database_url().await; - let db = db::connect_with_metrics( - &DatabaseConfig { - url: url.clone().into(), - pool_size: 1, - retry_count: 0, - }, - aster_drive_metrics::NoopMetrics::arc(), - ) - .await - .unwrap(); - let now = Utc::now().to_rfc3339(); - let storage_config = serde_json::json!({ - "format_version": 1, - "connector": { - "format_version": 1, - "connector_id": "asterdrive.storage.s3", - "schema_version": 1, - "values": { - "endpoint": "https://s3.example.test", - "bucket": "archive", - "base_path": "legacy", - "object_storage_upload_strategy": "relay_stream", - "object_storage_download_strategy": "relay_stream", - "s3_path_style": true, - "s3_region": "auto", - "s3_connect_timeout_secs": 5, - "s3_read_timeout_secs": 30, - "s3_operation_timeout_secs": 3600 - } - }, - "behavior": { "format_version": 1, "schema_version": 1, "values": {} } - }) - .to_string(); - db.execute_raw(Statement::from_sql_and_values( - DbBackend::Sqlite, - "INSERT INTO storage_policies \ - (id, name, driver_type, endpoint, bucket, access_key, secret_key, base_path, \ - max_file_size, allowed_types, options, is_default, chunk_size, created_at, updated_at, \ - connector_id, storage_config) \ - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - vec![ - 91_i64.into(), - "Legacy S3".into(), - "s3".into(), - "https://s3.example.test".into(), - "archive".into(), - "legacy-access".into(), - "legacy-secret".into(), - "legacy".into(), - 0_i64.into(), - "[]".into(), - "{}".into(), - false.into(), - 5_242_880_i64.into(), - now.clone().into(), - now.into(), - "asterdrive.storage.s3".into(), - storage_config.into(), - ], - )) - .await - .unwrap(); - db.close().await.unwrap(); - url -} - async fn setup_ready_database_url() -> String { let db_path = std::env::temp_dir().join(format!( "asterdrive-cli-ready-test-{}.db", @@ -2300,133 +2232,6 @@ async fn test_root_binary_database_migrate_sqlite_urls_without_mode_default_to_r .await; } -#[tokio::test] -async fn test_root_binary_database_migrate_imports_legacy_storage_credentials_on_sqlite() { - let source_database_url = setup_legacy_storage_upgrade_database_url().await; - let target_database_url = - setup_empty_database_url("asterdrive-cli-legacy-storage-target").await; - - assert_imported_legacy_storage_credentials( - &source_database_url, - &target_database_url, - DbBackend::Sqlite, - ) - .await; -} - -#[tokio::test] -async fn test_root_binary_database_migrate_imports_legacy_storage_credentials_on_postgres() { - let source_database_url = setup_legacy_storage_upgrade_database_url().await; - let target_database_url = common::postgres_test_database_url().await; - - assert_imported_legacy_storage_credentials( - &source_database_url, - &target_database_url, - DbBackend::Postgres, - ) - .await; -} - -#[tokio::test] -async fn test_root_binary_database_migrate_imports_legacy_storage_credentials_on_mysql() { - let source_database_url = setup_legacy_storage_upgrade_database_url().await; - let target_database_url = common::mysql_test_database_url().await; - - assert_imported_legacy_storage_credentials( - &source_database_url, - &target_database_url, - DbBackend::MySql, - ) - .await; -} - -async fn assert_imported_legacy_storage_credentials( - source_database_url: &str, - target_database_url: &str, - target_backend: DbBackend, -) { - let output = run_aster_drive(&[ - "database-migrate", - "--source-database-url", - source_database_url, - "--target-database-url", - target_database_url, - ]); - assert!( - output.status.success(), - "database-migrate stderr: {}", - String::from_utf8_lossy(&output.stderr) - ); - - let target_db = db::connect_with_metrics( - &DatabaseConfig { - url: target_database_url.to_string().into(), - pool_size: 1, - retry_count: 0, - }, - aster_drive_metrics::NoopMetrics::arc(), - ) - .await - .unwrap(); - assert!(column_exists(&target_db, target_backend, "storage_policies", "access_key").await); - assert!(column_exists(&target_db, target_backend, "storage_policies", "secret_key").await); - assert_eq!( - scalar_string( - &target_db, - target_backend, - "SELECT access_key FROM storage_policies WHERE id = 91", - ) - .await, - "" - ); - assert_eq!( - scalar_string( - &target_db, - target_backend, - "SELECT secret_key FROM storage_policies WHERE id = 91", - ) - .await, - "" - ); - assert_eq!( - scalar_i64( - &target_db, - target_backend, - "SELECT COUNT(*) FROM storage_connector_application_configs", - ) - .await, - 0 - ); - assert_eq!( - scalar_i64( - &target_db, - target_backend, - "SELECT COUNT(*) FROM storage_policy_credentials", - ) - .await, - 0 - ); - let credential = storage_policy_connector_credential_repo::find_by_policy(&target_db, 91) - .await - .unwrap() - .expect("legacy static credential should be imported after copy"); - let config = aster_drive::config::load_config_read_only().unwrap(); - let envelope: Value = serde_json::from_str(&credential.ciphertext).unwrap(); - let aad = "storage_policy_connector_credential:91:asterdrive.storage.s3:1"; - let plaintext = aster_forge_crypto::decrypt_secret( - config.auth.storage_credential_secret_key.as_bytes(), - b"asterdrive:storage-credential-token:v1", - aad.as_bytes(), - envelope["ciphertext"].as_str().unwrap(), - ) - .unwrap(); - let plaintext = String::from_utf8(plaintext).unwrap(); - let payload: Value = serde_json::from_str(&plaintext).unwrap(); - assert_eq!(payload["s3_access_key_id"], "legacy-access"); - assert_eq!(payload["s3_secret_access_key"], "legacy-secret"); - target_db.close().await.unwrap(); -} - #[tokio::test] async fn test_root_binary_database_migrate_allows_consumed_contact_verification_history() { let source_db_path = std::env::temp_dir().join(format!( diff --git a/tests/platform/database_backends.rs b/tests/platform/database_backends.rs index 970ac840b..822b825ea 100644 --- a/tests/platform/database_backends.rs +++ b/tests/platform/database_backends.rs @@ -14,12 +14,10 @@ use aster_drive::db::repository::background_task_repo; use aster_drive_migration::{CurrentMigrator, Migrator, MigratorTrait}; use aster_drive_model::entities::{ background_task, file_blob, file_create_idempotency, folder_tree_operation_member, - storage_policy, }; use aster_drive_model::types::file_blob::FileBlobBacking; use aster_drive_model::types::{ - BackgroundTaskKind, BackgroundTaskStatus, EntityType, StoredStoragePolicyAllowedTypes, - StoredStoragePolicyConfig, StoredTaskPayload, StoredTaskResult, + BackgroundTaskKind, BackgroundTaskStatus, EntityType, StoredTaskPayload, StoredTaskResult, }; const OLD_BACKGROUND_TASK_DISPLAY_NAME_LIMIT: usize = 255; @@ -897,126 +895,6 @@ async fn assert_folder_tree_staging_primary_key_and_task_cascade(db: &DatabaseCo ); } -async fn assert_current_storage_policy_ignores_retained_legacy_columns( - db: &DatabaseConnection, - backend: DbBackend, -) { - async fn insert_current_policy( - db: &DatabaseConnection, - name: String, - ) -> Result { - let now = chrono::Utc::now(); - storage_policy::ActiveModel { - name: Set(name), - connector_id: Set("asterdrive.storage.local".to_string()), - storage_config: Set(StoredStoragePolicyConfig::from( - r#"{"format_version":1,"connector":{"format_version":1,"connector_id":"asterdrive.storage.local","schema_version":1,"values":{"base_path":"./data/uploads","content_dedup":false}},"behavior":{"format_version":1,"schema_version":1,"values":{"storage_native_thumbnail_enabled":false,"storage_native_media_metadata_enabled":false}}}"# - .to_string(), - )), - max_file_size: Set(0), - allowed_types: Set(StoredStoragePolicyAllowedTypes::empty()), - is_default: Set(false), - chunk_size: Set(0), - created_at: Set(now), - updated_at: Set(now), - ..Default::default() - } - .insert(db) - .await - } - - let now = chrono::Utc::now(); - let policy = insert_current_policy(db, format!("connector-policy-{backend:?}-{now}")) - .await - .expect("current storage policy entity should omit retained 0.5.x legacy columns"); - - let row = db - .query_one_raw(Statement::from_string( - backend, - format!( - "SELECT driver_type FROM storage_policies WHERE id = {}", - policy.id - ), - )) - .await - .expect("retained storage policy driver_type should query") - .expect("inserted current storage policy should exist"); - assert_eq!( - row.try_get_by_index::(0) - .expect("retained driver_type should decode"), - "", - "0.5.x compatibility migration should supply the legacy driver_type default" - ); - - // Roll back the retained-column compatibility migration, which restores - // the legacy write requirements while leaving converted policy rows intact. - let later_migration_steps = CurrentMigrator::migrations() - .iter() - .rev() - .position(|migration| { - migration.name() == "m20260805_000001_allow_connector_policy_writes_with_legacy_schema" - }) - .map(|tail_index| u32::try_from(tail_index).expect("migration count should fit u32")) - .expect("retained-column compatibility migration should remain registered"); - if later_migration_steps > 0 { - CurrentMigrator::down(db, Some(later_migration_steps)) - .await - .expect( - "migrations after the retained-column compatibility migration should roll back", - ); - } - CurrentMigrator::down(db, Some(1)) - .await - .expect("retained-column compatibility migration should roll back on production backend"); - insert_current_policy(db, format!("connector-policy-down-{backend:?}-{now}")) - .await - .expect_err("historical retained schema should reject the current policy insert shape"); - if backend == DbBackend::MySql { - let row = db - .query_one_raw(Statement::from_string( - backend, - format!( - "SELECT options FROM storage_policies WHERE id = {}", - policy.id - ), - )) - .await - .expect("rolled-back MySQL legacy options should query") - .expect("inserted MySQL storage policy should remain"); - assert_eq!( - row.try_get_by_index::(0) - .expect("rolled-back MySQL legacy options should decode"), - "{}", - "rollback should backfill nullable 0.5.x compatibility values before restoring NOT NULL" - ); - } - - CurrentMigrator::up(db, Some(1)) - .await - .expect("retained-column compatibility migration should reapply on production backend"); - if later_migration_steps > 0 { - CurrentMigrator::up(db, Some(later_migration_steps)) - .await - .expect("migrations after the retained-column compatibility migration should reapply"); - } - let reapplied = - insert_current_policy(db, format!("connector-policy-reapplied-{backend:?}-{now}")) - .await - .expect("reapplied compatibility migration should restore current policy writes"); - let row = db - .query_one_raw(Statement::from_string( - backend, - format!( - "SELECT driver_type FROM storage_policies WHERE id = {}", - reapplied.id - ), - )) - .await - .expect("reapplied retained driver_type should query") - .expect("reapplied current storage policy should exist"); - assert_eq!(row.try_get_by_index::(0).unwrap(), ""); -} - #[actix_web::test] async fn test_sqlite_transactions_are_serialized_by_single_connection_pool() { use sea_orm::TransactionTrait; @@ -1079,7 +957,6 @@ async fn exercise_backend_smoke(database_url: &str, backend: DbBackend) { assert_background_task_display_name_column_len(state.writer_db(), backend).await; assert_background_task_display_name_accepts_expanded_len(state.writer_db()).await; assert_upload_session_kind_column(state.writer_db(), backend).await; - assert_current_storage_policy_ignores_retained_legacy_columns(state.writer_db(), backend).await; assert_folder_tree_staging_primary_key_and_task_cascade(state.writer_db()).await; let app = create_test_app!(state.clone()); diff --git a/tests/platform/migrations.rs b/tests/platform/migrations.rs index 07a10a427..20fbc1aac 100644 --- a/tests/platform/migrations.rs +++ b/tests/platform/migrations.rs @@ -10,13 +10,11 @@ use sea_orm::{ const ALLOW_SHARED_WEBDAV_LOCKS_MIGRATION: &str = "m20260604_000001_allow_shared_webdav_locks"; const RENAME_UPLOAD_SESSION_OBJECT_FIELDS_MIGRATION: &str = "m20260618_000001_rename_upload_session_object_fields"; -const ADD_STORAGE_CONNECTOR_APPLICATION_CONFIGS_MIGRATION: &str = - "m20260619_000001_add_storage_connector_application_configs"; +const REMOVE_STORAGE_POLICY_LEGACY_MIGRATION: &str = + "m20260820_000001_remove_storage_policy_legacy"; const ENFORCE_JSON_TEXT_NOT_NULL_MIGRATION: &str = "m20260620_000001_enforce_json_text_not_null"; const RENAME_MANAGED_INGRESS_PROFILES_MIGRATION: &str = "m20260704_000001_rename_managed_ingress_profiles_to_remote_storage_targets"; -const ADD_REMOTE_STORAGE_TARGET_KEY_TO_STORAGE_POLICIES_MIGRATION: &str = - "m20260704_000002_add_remote_storage_target_key_to_storage_policies"; const DROP_REMOTE_STORAGE_TARGET_MAX_FILE_SIZE_MIGRATION: &str = "m20260705_000001_drop_remote_storage_target_max_file_size"; const ALIGN_FORGE_AUDIT_CONTRACT_MIGRATION: &str = "m20260712_000001_align_forge_audit_contract"; @@ -1636,10 +1634,9 @@ async fn production_sqlite_migrator_rebuilds_referenced_storage_policy_parent_ta .await .expect("foreign-key fixture should reference the legacy storage policy"); - Migrator::up(&db, None) + CurrentMigrator::up(&db, Some(1)) .await - .expect("production migration coordinator should rebuild the referenced parent table"); - + .expect("compatibility migration should rebuild the referenced parent table"); let (_, driver_type_default) = sqlite_column_type_and_default(&db, "storage_policies", "driver_type").await; assert_eq!(driver_type_default.as_deref(), Some("''")); @@ -2222,12 +2219,93 @@ fn steps_before_migration(migration_name: &str) -> u32 { u32::try_from(position).expect("migration step count should fit u32") } -fn steps_to_roll_back_upload_session_object_fields() -> u32 { - steps_to_roll_back_migration(RENAME_UPLOAD_SESSION_OBJECT_FIELDS_MIGRATION) +#[tokio::test] +async fn storage_policy_legacy_cleanup_rejects_unmigrated_static_secrets_before_ddl() { + let db = Database::connect("sqlite::memory:") + .await + .expect("SQLite migration fixture should connect"); + CurrentMigrator::up( + &db, + Some(steps_before_migration( + REMOVE_STORAGE_POLICY_LEGACY_MIGRATION, + )), + ) + .await + .expect("schema before storage cleanup should apply"); + db.execute_unprepared( + "INSERT INTO storage_policies (name, driver_type, endpoint, bucket, access_key, secret_key, base_path, max_file_size, allowed_types, options, is_default, chunk_size, created_at, updated_at, connector_id, storage_config) VALUES ('legacy', 's3', '', '', 'SECRET', '', '', 0, '[]', '{}', 0, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, 'asterdrive.storage.s3', '{}')", + ) + .await + .expect("legacy static credential fixture should insert"); + + let error = CurrentMigrator::up(&db, Some(1)) + .await + .expect_err("unmigrated static credentials must block 0.5.1 cleanup"); + assert!( + error + .to_string() + .contains("start the database successfully on AsterDrive 0.5.0") + ); + assert!( + sqlite_table_columns(&db, "storage_policies") + .await + .iter() + .any(|column| column == "access_key") + ); + assert!(sqlite_table_exists(&db, "storage_policy_credentials").await); + assert_eq!( + db.query_one_raw(Statement::from_string( + DbBackend::Sqlite, + "SELECT access_key FROM storage_policies WHERE name = 'legacy'", + )) + .await + .expect("legacy access key should query") + .expect("legacy policy should exist") + .try_get_by_index::(0) + .expect("legacy access key should decode"), + "SECRET" + ); } -fn steps_to_roll_back_storage_connector_application_configs() -> u32 { - steps_to_roll_back_migration(ADD_STORAGE_CONNECTOR_APPLICATION_CONFIGS_MIGRATION) +#[tokio::test] +async fn storage_policy_legacy_cleanup_drops_empty_legacy_surface() { + let db = Database::connect("sqlite::memory:") + .await + .expect("SQLite migration fixture should connect"); + CurrentMigrator::up( + &db, + Some(steps_before_migration( + REMOVE_STORAGE_POLICY_LEGACY_MIGRATION, + )), + ) + .await + .expect("schema before storage cleanup should apply"); + CurrentMigrator::up(&db, Some(1)) + .await + .expect("empty legacy stores should be removed"); + for column in [ + "driver_type", + "endpoint", + "bucket", + "access_key", + "secret_key", + "base_path", + "remote_node_id", + "remote_storage_target_key", + "options", + ] { + assert!( + !sqlite_table_columns(&db, "storage_policies") + .await + .iter() + .any(|actual| actual == column) + ); + } + assert!(!sqlite_table_exists(&db, "storage_policy_credentials").await); + assert!(!sqlite_table_exists(&db, "storage_connector_application_configs").await); + CurrentMigrator::up(&db, None) + .await + .expect("reapplying the current migration set should be idempotent"); } fn steps_to_roll_back_rename_managed_ingress_profiles() -> u32 { @@ -2238,10 +2316,6 @@ fn steps_to_roll_back_remote_storage_target_max_file_size() -> u32 { steps_to_roll_back_migration(DROP_REMOTE_STORAGE_TARGET_MAX_FILE_SIZE_MIGRATION) } -fn steps_to_roll_back_storage_policy_remote_storage_target_key() -> u32 { - steps_to_roll_back_migration(ADD_REMOTE_STORAGE_TARGET_KEY_TO_STORAGE_POLICIES_MIGRATION) -} - async fn roll_back_allow_shared_webdav_locks( db: &sea_orm::DatabaseConnection, ) -> Result<(), DbErr> { @@ -2541,9 +2615,7 @@ async fn json_text_columns_are_not_null_in_current_schema() { let db = setup_current_schema().await; for (table, column) in [ ("external_auth_providers", "options"), - ("storage_policy_credentials", "metadata"), ("storage_policy_authorization_flows", "context"), - ("storage_connector_application_configs", "metadata"), ] { assert!( sqlite_column_is_not_null(&db, table, column).await, @@ -2767,59 +2839,6 @@ async fn forge_mail_outbox_contract_preserves_rows_and_named_indexes() { assert_eq!(preserved.try_get_by_index::(4).unwrap(), 2); } -#[tokio::test] -async fn storage_connector_application_config_migration_adds_canonical_config_table() { - assert!( - CurrentMigrator::migrations().iter().any( - |migration| migration.name() == ADD_STORAGE_CONNECTOR_APPLICATION_CONFIGS_MIGRATION - ), - "application config migration should be registered" - ); - - let db = setup_current_schema().await; - assert!( - sqlite_table_exists(&db, "storage_connector_application_configs").await, - "current schema should include storage_connector_application_configs" - ); - let current_columns = sqlite_table_columns(&db, "storage_connector_application_configs").await; - for expected in [ - "id", - "policy_id", - "provider", - "tenant_id", - "scopes", - "client_id", - "client_secret_ciphertext", - "metadata", - "created_at", - "updated_at", - ] { - assert!(has_column(¤t_columns, expected), "missing {expected}"); - } - - CurrentMigrator::down( - &db, - Some(steps_to_roll_back_storage_connector_application_configs()), - ) - .await - .expect("application config migration should roll back"); - assert!( - !sqlite_table_exists(&db, "storage_connector_application_configs").await, - "rollback should remove storage_connector_application_configs" - ); - - CurrentMigrator::up( - &db, - Some(steps_to_roll_back_storage_connector_application_configs()), - ) - .await - .expect("application config migration should reapply"); - assert!( - sqlite_table_exists(&db, "storage_connector_application_configs").await, - "reapply should recreate storage_connector_application_configs" - ); -} - #[tokio::test] async fn upload_session_object_field_migration_renames_legacy_columns() { assert!( @@ -2835,92 +2854,6 @@ async fn upload_session_object_field_migration_renames_legacy_columns() { assert!(has_column(¤t_columns, "object_multipart_id")); assert!(!has_column(¤t_columns, "s3_temp_key")); assert!(!has_column(¤t_columns, "s3_multipart_id")); - - CurrentMigrator::down(&db, Some(steps_to_roll_back_upload_session_object_fields())) - .await - .expect("object field rename migration should roll back"); - let rolled_back_columns = sqlite_table_columns(&db, "upload_sessions").await; - assert!(has_column(&rolled_back_columns, "s3_temp_key")); - assert!(has_column(&rolled_back_columns, "s3_multipart_id")); - assert!(!has_column(&rolled_back_columns, "object_temp_key")); - assert!(!has_column(&rolled_back_columns, "object_multipart_id")); - - CurrentMigrator::up(&db, Some(steps_to_roll_back_upload_session_object_fields())) - .await - .expect("object field rename migration should reapply"); - let reapplied_columns = sqlite_table_columns(&db, "upload_sessions").await; - assert!(has_column(&reapplied_columns, "object_temp_key")); - assert!(has_column(&reapplied_columns, "object_multipart_id")); - assert!(!has_column(&reapplied_columns, "s3_temp_key")); - assert!(!has_column(&reapplied_columns, "s3_multipart_id")); -} - -#[tokio::test] -async fn storage_policy_remote_storage_target_key_migration_round_trips_column() { - assert!( - CurrentMigrator::migrations().iter().any(|migration| { - migration.name() == ADD_REMOTE_STORAGE_TARGET_KEY_TO_STORAGE_POLICIES_MIGRATION - }), - "storage policy remote target key migration should be registered" - ); - - let db = setup_current_schema().await; - let current_columns = sqlite_table_columns(&db, "storage_policies").await; - assert!( - has_column(¤t_columns, "remote_storage_target_key"), - "current schema should include storage_policies.remote_storage_target_key" - ); - assert!( - sqlite_table_index_exists( - &db, - "storage_policies", - "idx_storage_policies_remote_target" - ) - .await, - "current schema should include idx_storage_policies_remote_target" - ); - - CurrentMigrator::down( - &db, - Some(steps_to_roll_back_storage_policy_remote_storage_target_key()), - ) - .await - .expect("remote target key migration should roll back"); - let rolled_back_columns = sqlite_table_columns(&db, "storage_policies").await; - assert!( - !has_column(&rolled_back_columns, "remote_storage_target_key"), - "rollback should remove storage_policies.remote_storage_target_key" - ); - assert!( - !sqlite_table_index_exists( - &db, - "storage_policies", - "idx_storage_policies_remote_target" - ) - .await, - "rollback should remove idx_storage_policies_remote_target" - ); - - CurrentMigrator::up( - &db, - Some(steps_to_roll_back_storage_policy_remote_storage_target_key()), - ) - .await - .expect("remote target key migration should reapply"); - let reapplied_columns = sqlite_table_columns(&db, "storage_policies").await; - assert!( - has_column(&reapplied_columns, "remote_storage_target_key"), - "reapply should restore storage_policies.remote_storage_target_key" - ); - assert!( - sqlite_table_index_exists( - &db, - "storage_policies", - "idx_storage_policies_remote_target" - ) - .await, - "reapply should restore idx_storage_policies_remote_target" - ); } #[tokio::test] @@ -3012,22 +2945,6 @@ async fn mysql_remote_storage_target_rename_migration_round_trips_indexes() { .await, "MySQL down should remove the remote storage target key index name" ); - - CurrentMigrator::up( - &db, - Some(steps_to_roll_back_rename_managed_ingress_profiles()), - ) - .await - .expect("remote storage target rename migration should reapply on MySQL"); - assert!( - mysql_table_index_exists( - &db, - "remote_storage_targets", - "idx_remote_storage_targets_binding_target_key" - ) - .await, - "MySQL reapply should restore the target key index name" - ); } #[tokio::test] diff --git a/tests/platform/schema_drift.rs b/tests/platform/schema_drift.rs index 25a81b6e5..113446d4a 100644 --- a/tests/platform/schema_drift.rs +++ b/tests/platform/schema_drift.rs @@ -165,24 +165,6 @@ where fn expected_database_only_columns(backend: DbBackend, table_name: &str) -> BTreeSet<&'static str> { match (backend, table_name) { - // AsterDrive 0.5.x keeps these columns so the startup credential - // importer can read pre-refactor policies. They remain required in the - // 0.5 schema but deliberately stay out of the current SeaORM entity. - // Issue #463 removes both the physical columns and this exact exception - // in 0.6.0. - (_, "storage_policies") => [ - "driver_type", - "endpoint", - "bucket", - "access_key", - "secret_key", - "base_path", - "remote_node_id", - "remote_storage_target_key", - "options", - ] - .into_iter() - .collect(), // MySQL's default text collation is case-insensitive. These generated // projections let the database enforce byte-sensitive XML property // identity; SeaORM and database-migrate select only business columns.