diff --git a/crates/app/src/cmd_migrate.rs b/crates/app/src/cmd_migrate.rs index da2e3c376..141455381 100755 --- a/crates/app/src/cmd_migrate.rs +++ b/crates/app/src/cmd_migrate.rs @@ -37,10 +37,10 @@ pub async fn run(args: MigrateArgs) -> anyhow::Result<()> { args.config, ); } - // Load the config for validation only: migrate drives the bootstrapper from - // the config path directly, but a malformed config should fail here rather - // than midway through a migration. - config::load(&args.config)?; + // Load the config for validation and to reach the catalog for the policy + // rewrite: migrate drives the schema steps through the bootstrapper, but a + // malformed config should fail here rather than midway through a migration. + let app_config = config::load(&args.config)?; println!("=== extenddb migrate ==="); println!("Config: {}", args.config); @@ -67,7 +67,12 @@ pub async fn run(args: MigrateArgs) -> anyhow::Result<()> { .acquire_migration_lock() .await .map_err(|e| anyhow::anyhow!("{e:?}"))?; - let result = apply_migrations(bootstrap.as_ref(), &args).await; + let result = apply_migrations( + bootstrap.as_ref(), + &args, + app_config.storage.connection_config(), + ) + .await; if let Err(e) = bootstrap.release_migration_lock().await { tracing::warn!("Failed to release migration lock: {e:?}"); } @@ -79,6 +84,7 @@ pub async fn run(args: MigrateArgs) -> anyhow::Result<()> { async fn apply_migrations( bootstrap: &dyn extenddb_storage::bootstrapper::Bootstrapper, args: &MigrateArgs, + connection: &str, ) -> anyhow::Result<()> { // Show current catalog version. println!("--- Checking current catalog version..."); @@ -131,20 +137,75 @@ async fn apply_migrations( } } - if !catalog_pending && data_pending.is_empty() && !repair_pending { + // Detect stored policies that still name a dynamodb:ResourceTag condition + // key. Detection is read-only and runs on every migrate (like the repair + // above), so an interrupted run is finished by the next. The catalog is + // reached through a catalog-only management store, not the bootstrapper's + // schema connection. + println!("--- Checking resource-tag condition keys..."); + let mgmt = extenddb_storage::management_store_factory::create_management_store(connection) + .await + .map_err(|e| anyhow::anyhow!("{e}"))?; + let settings = extenddb_storage::settings_store::create_settings_store(connection) + .await + .map_err(|e| anyhow::anyhow!("{e}"))?; + let rtag_detect = extenddb_storage::policy_migration::migrate_resource_tag_condition_keys( + mgmt.as_ref(), + false, + ) + .await; + let marker_present = settings + .get_setting(extenddb_core::settings_keys::RESOURCE_TAG_MIGRATION) + .await + .map_err(|e| anyhow::anyhow!("{e:?}"))? + .is_some(); + if rtag_detect.rewritten.is_empty() && rtag_detect.needs_attention.is_empty() { + println!(" No dynamodb:ResourceTag condition keys found."); + } else { + for r in &rtag_detect.rewritten { + println!( + " Rewrite: account {} {} policy {} ({} key(s))", + r.account_id, r.principal, r.policy_name, r.rewrites + ); + } + for n in &rtag_detect.needs_attention { + println!( + " NEEDS ATTENTION: account {} {} policy {} (aws:ResourceTag and \ + dynamodb:ResourceTag forms collide; left unchanged)", + n.account_id, n.principal, n.policy_name + ); + } + } + if rtag_detect.errors > 0 { + println!( + " WARNING: {} policy lookup(s) failed during detection.", + rtag_detect.errors + ); + } + let rtag_pending = !rtag_detect.rewritten.is_empty() || rtag_detect.errors > 0; + // A catalog with no marker must have one written, even when there is + // nothing to rewrite, so that serve will start. + let marker_pending = !marker_present; + + if !catalog_pending + && data_pending.is_empty() + && !repair_pending + && !rtag_pending + && !marker_pending + { println!(); - if detected.needs_attention.is_empty() { + if detected.needs_attention.is_empty() && rtag_detect.needs_attention.is_empty() { println!( "Everything is up to date (catalog version {expected}). No migrations needed." ); } else { // Nothing is automatically applicable, but saying "everything is up // to date" straight after a NEEDS ATTENTION line would be - // contradictory: those tables require a human. + // contradictory: those items require a human. println!( "No migrations needed (catalog version {expected}), but {} item(s) above \ need manual attention.", - detected.needs_attention.len() + detected.needs_attention.len() + rtag_detect.needs_attention.len() ); } return Ok(()); @@ -164,6 +225,28 @@ async fn apply_migrations( detected.repaired.join("; ") )); } + if !rtag_detect.rewritten.is_empty() { + what.push(format!( + "resource-tag condition-key rewrites [{} policy(ies)]", + rtag_detect.rewritten.len() + )); + } + if marker_pending { + what.push("record the resource-tag migration marker".to_owned()); + } + if rtag_detect.errors > 0 { + what.push(format!( + "{} policy lookup(s) that failed and must be retried", + rtag_detect.errors + )); + } + if !rtag_detect.needs_attention.is_empty() { + what.push(format!( + "{} policy(ies) listed as NEEDS ATTENTION, which must be fixed by hand before \ + the migration can complete", + rtag_detect.needs_attention.len() + )); + } anyhow::bail!( "--yes is required to apply migrations. Pending: {}.", what.join("; ") @@ -197,6 +280,59 @@ async fn apply_migrations( } } + // Rewrite stored policies from dynamodb:ResourceTag to aws:ResourceTag, + // after the schema migrations. Every storage error is counted; on a + // non-zero count migrate reports it, does NOT record the completion + // marker, and exits non-zero, so the next run finishes the rewrite. No + // policy body is logged. + let rtag_applied = extenddb_storage::policy_migration::migrate_resource_tag_condition_keys( + mgmt.as_ref(), + true, + ) + .await; + for r in &rtag_applied.rewritten { + println!( + " Rewrote resource-tag keys: account {} {} policy {} ({} key(s))", + r.account_id, r.principal, r.policy_name, r.rewrites + ); + } + for n in &rtag_applied.needs_attention { + println!( + " NEEDS ATTENTION: account {} {} policy {} left unchanged (collision)", + n.account_id, n.principal, n.policy_name + ); + } + if rtag_applied.errors > 0 { + anyhow::bail!( + "resource-tag migration incomplete: {} policy lookup(s) failed. \ + Re-run `extenddb migrate --yes` to finish; the migration marker was not recorded.", + rtag_applied.errors + ); + } + // A policy left unchanged because both resource-tag forms name the same + // tag key in one operator block is not migrated. Recording the marker + // would let serve evaluate it with the dynamodb: half absent, which can + // turn a Deny off. Withhold the marker so serve keeps refusing until the + // policy is fixed by hand and migrate is run again. + if !rtag_applied.needs_attention.is_empty() { + anyhow::bail!( + "resource-tag migration incomplete: {} policy(ies) above name both \ + aws:ResourceTag/ and dynamodb:ResourceTag/ for the same key in one \ + condition operator and were left unchanged. Remove one of the two forms from \ + each, then re-run `extenddb migrate --yes`; the migration marker was not recorded.", + rtag_applied.needs_attention.len() + ); + } + // Record the completion marker only after a clean rewrite. serve reads this + // row and refuses to start without it. + settings + .set_setting( + extenddb_core::settings_keys::RESOURCE_TAG_MIGRATION, + extenddb_core::settings_keys::RESOURCE_TAG_MIGRATION_DONE, + ) + .await + .map_err(|e| anyhow::anyhow!("{e:?}"))?; + // Read new catalog version. let new = bootstrap .read_catalog_version() diff --git a/crates/auth/src/policy/condition/tests.rs b/crates/auth/src/policy/condition/tests.rs index 8f0d3b3d6..b3c90cc91 100644 --- a/crates/auth/src/policy/condition/tests.rs +++ b/crates/auth/src/policy/condition/tests.rs @@ -64,4 +64,5 @@ fn cond(op: ConditionOperator, key: &str, values: Vec<&str>) -> Condition { mod multivalued; mod operators; +mod resource_tag; mod three_case; diff --git a/crates/auth/src/policy/condition/tests/multivalued.rs b/crates/auth/src/policy/condition/tests/multivalued.rs index cf892a096..17f28a0e8 100644 --- a/crates/auth/src/policy/condition/tests/multivalued.rs +++ b/crates/auth/src/policy/condition/tests/multivalued.rs @@ -247,12 +247,12 @@ fn string_not_like_multiple_policy_values_no_match() { #[test] fn policy_variable_expansion_principal_tag() { let ctx = TestContext::new() - .with("dynamodb:ResourceTag/Team", vec!["Alpha"]) + .with("aws:ResourceTag/Team", vec!["Alpha"]) .with("aws:PrincipalTag/Team", vec!["Alpha"]); assert!(evaluate_condition( &cond( ConditionOperator::StringEquals, - "dynamodb:ResourceTag/Team", + "aws:ResourceTag/Team", vec!["${aws:PrincipalTag/Team}"] ), &ctx @@ -262,12 +262,12 @@ fn policy_variable_expansion_principal_tag() { #[test] fn policy_variable_expansion_mismatch() { let ctx = TestContext::new() - .with("dynamodb:ResourceTag/Team", vec!["Beta"]) + .with("aws:ResourceTag/Team", vec!["Beta"]) .with("aws:PrincipalTag/Team", vec!["Alpha"]); assert!(!evaluate_condition( &cond( ConditionOperator::StringEquals, - "dynamodb:ResourceTag/Team", + "aws:ResourceTag/Team", vec!["${aws:PrincipalTag/Team}"] ), &ctx diff --git a/crates/auth/src/policy/condition/tests/resource_tag.rs b/crates/auth/src/policy/condition/tests/resource_tag.rs new file mode 100644 index 000000000..3de55b884 --- /dev/null +++ b/crates/auth/src/policy/condition/tests/resource_tag.rs @@ -0,0 +1,241 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Resource-tag condition-key evaluation on the real `RequestContext`. +//! +//! The live service evaluates resource tags only as `aws:ResourceTag/` and +//! leaves `dynamodb:ResourceTag/` absent. These tests pin that for both +//! prefixes under the bare, `IfExists`, `ForAllValues:`, `ForAnyValue:` and +//! `Null` forms, so a regression that reintroduced the `dynamodb:` resolution +//! would change at least one outcome here. + +use std::collections::HashMap; + +use crate::policy::condition::evaluate_condition; +use crate::policy::context::{RequestContext, RequestParams}; +use crate::policy::document::{Condition, ConditionOperator}; + +/// A request against a resource tagged `env=prod`. +fn ctx_env_prod() -> RequestContext { + RequestContext::build( + HashMap::new(), + HashMap::from([("env".to_owned(), "prod".to_owned())]), + false, + RequestParams::default(), + ) +} + +fn cond(op: ConditionOperator, key: &str, values: &[&str]) -> Condition { + Condition { + operator: op, + key: key.to_owned(), + values: values.iter().map(|s| (*s).to_owned()).collect(), + } +} + +fn bare(op: ConditionOperator) -> ConditionOperator { + op +} +fn if_exists(op: ConditionOperator) -> ConditionOperator { + ConditionOperator::IfExists(Box::new(op)) +} +fn for_all(op: ConditionOperator) -> ConditionOperator { + ConditionOperator::ForAllValues(Box::new(op)) +} +fn for_any(op: ConditionOperator) -> ConditionOperator { + ConditionOperator::ForAnyValue(Box::new(op)) +} + +// ── Bare positive ──────────────────────────────────────────────────────── + +#[test] +fn bare_string_equals_aws_matches_dynamodb_absent() { + let ctx = ctx_env_prod(); + // aws:ResourceTag/env resolves to "prod" and matches. + assert!(evaluate_condition( + &cond( + bare(ConditionOperator::StringEquals), + "aws:ResourceTag/env", + &["prod"] + ), + &ctx + )); + // dynamodb:ResourceTag/env is absent: a positive bare operator never matches. + assert!(!evaluate_condition( + &cond( + bare(ConditionOperator::StringEquals), + "dynamodb:ResourceTag/env", + &["prod"] + ), + &ctx + )); +} + +// ── Bare negated ───────────────────────────────────────────────────────── + +#[test] +fn bare_string_not_equals_dynamodb_absent_matches() { + let ctx = ctx_env_prod(); + // aws:ResourceTag/env resolves to "prod": StringNotEquals "prod" is false. + assert!(!evaluate_condition( + &cond( + bare(ConditionOperator::StringNotEquals), + "aws:ResourceTag/env", + &["prod"] + ), + &ctx + )); + // dynamodb:ResourceTag/env is absent: a negated bare operator matches + // (absent-key semantics), it never fails closed. + assert!(evaluate_condition( + &cond( + bare(ConditionOperator::StringNotEquals), + "dynamodb:ResourceTag/env", + &["prod"] + ), + &ctx + )); +} + +// ── IfExists ───────────────────────────────────────────────────────────── + +#[test] +fn if_exists_dynamodb_absent_passes_aws_present_evaluates() { + let ctx = ctx_env_prod(); + // IfExists passes when the key is absent: dynamodb:ResourceTag/env absent. + assert!(evaluate_condition( + &cond( + if_exists(ConditionOperator::StringEquals), + "dynamodb:ResourceTag/env", + &["dev"] + ), + &ctx + )); + // aws:ResourceTag/env is present ("prod"), so IfExists evaluates the base + // operator: StringEquals "dev" does not match. + assert!(!evaluate_condition( + &cond( + if_exists(ConditionOperator::StringEquals), + "aws:ResourceTag/env", + &["dev"] + ), + &ctx + )); +} + +// ── ForAllValues ───────────────────────────────────────────────────────── + +#[test] +fn for_all_values_dynamodb_absent_vacuously_true() { + let ctx = ctx_env_prod(); + // Absent key: ForAllValues is vacuously true. + assert!(evaluate_condition( + &cond( + for_all(ConditionOperator::StringEquals), + "dynamodb:ResourceTag/env", + &["dev"] + ), + &ctx + )); + // Present aws key with a non-matching value: not all values match. + assert!(!evaluate_condition( + &cond( + for_all(ConditionOperator::StringEquals), + "aws:ResourceTag/env", + &["dev"] + ), + &ctx + )); +} + +// ── ForAnyValue ────────────────────────────────────────────────────────── + +#[test] +fn for_any_value_dynamodb_absent_false_aws_present_matches() { + let ctx = ctx_env_prod(); + // Absent key: ForAnyValue is false. + assert!(!evaluate_condition( + &cond( + for_any(ConditionOperator::StringEquals), + "dynamodb:ResourceTag/env", + &["prod"] + ), + &ctx + )); + // Present aws key with a matching value: at least one value matches. + assert!(evaluate_condition( + &cond( + for_any(ConditionOperator::StringEquals), + "aws:ResourceTag/env", + &["prod"] + ), + &ctx + )); +} + +// ── Null ───────────────────────────────────────────────────────────────── + +#[test] +fn null_true_matches_absent_dynamodb_not_present_aws() { + let ctx = ctx_env_prod(); + // Null:true means "key is absent": true for dynamodb, false for aws. + assert!(evaluate_condition( + &cond( + ConditionOperator::Null, + "dynamodb:ResourceTag/env", + &["true"] + ), + &ctx + )); + assert!(!evaluate_condition( + &cond(ConditionOperator::Null, "aws:ResourceTag/env", &["true"]), + &ctx + )); +} + +#[test] +fn null_false_matches_present_aws_not_absent_dynamodb() { + let ctx = ctx_env_prod(); + // Null:false means "key is present": true for aws, false for dynamodb. + assert!(evaluate_condition( + &cond(ConditionOperator::Null, "aws:ResourceTag/env", &["false"]), + &ctx + )); + assert!(!evaluate_condition( + &cond( + ConditionOperator::Null, + "dynamodb:ResourceTag/env", + &["false"] + ), + &ctx + )); +} + +// ── Case-insensitive prefix ────────────────────────────────────────────── + +#[test] +fn dynamodb_prefix_absent_in_any_case() { + let ctx = ctx_env_prod(); + for key in [ + "dynamodb:ResourceTag/env", + "dynamodb:resourcetag/env", + "DYNAMODB:RESOURCETAG/env", + "DynamoDB:ResourceTag/env", + ] { + assert!( + evaluate_condition(&cond(ConditionOperator::Null, key, &["true"]), &ctx), + "{key} should be absent (Null:true)" + ); + } + // aws:ResourceTag resolves in any case. + for key in [ + "aws:ResourceTag/env", + "aws:resourcetag/env", + "AWS:RESOURCETAG/env", + ] { + assert!( + evaluate_condition(&cond(ConditionOperator::StringEquals, key, &["prod"]), &ctx), + "{key} should resolve to prod" + ); + } +} diff --git a/crates/auth/src/policy/context.rs b/crates/auth/src/policy/context.rs index ee8a9ff7f..25a4a0764 100755 --- a/crates/auth/src/policy/context.rs +++ b/crates/auth/src/policy/context.rs @@ -75,7 +75,7 @@ fn tag_lookup<'a>( } /// Strip a tag prefix (`aws:PrincipalTag/`, `aws:RequestTag/`, -/// `aws:ResourceTag/`, `dynamodb:ResourceTag/`) from the front of `key` +/// `aws:ResourceTag/`) from the front of `key` /// ASCII-case-insensitively, returning the original (un-normalized) tag sub-key. /// /// The prefix is ASCII, so the service (and ExtendDB) match it without regard to @@ -203,7 +203,7 @@ pub struct RequestParams { pub struct RequestContext { /// Tags on the authenticated principal (`aws:PrincipalTag/*`). pub principal_tags: HashMap, - /// Tags on the target resource (`dynamodb:ResourceTag/*`). + /// Tags on the target resource, resolved as `aws:ResourceTag/*`. pub resource_tags: HashMap, /// Partition key values being accessed. pub leading_keys: Option>, @@ -305,11 +305,11 @@ impl ConditionContext for RequestContext { if let Some(sub) = strip_prefix_ci(key, "aws:PrincipalTag/") { tag_lookup(&self.principal_tags, sub, super::casefold::case_fold) } else if let Some(sub) = strip_prefix_ci(key, "aws:ResourceTag/") { - // The service evaluates resource tags as aws:ResourceTag/*. ExtendDB - // resolves them from the same resource-tag map it uses for - // dynamodb:ResourceTag/*. - tag_lookup(&self.resource_tags, sub, super::casefold::case_fold) - } else if let Some(sub) = strip_prefix_ci(key, "dynamodb:ResourceTag/") { + // The service evaluates resource tags only as aws:ResourceTag/*, + // resolved from the resource-tag map. dynamodb:ResourceTag/* is + // left absent (it falls through to the match below and resolves to + // None), matching the live service, which does not populate that + // key. tag_lookup(&self.resource_tags, sub, super::casefold::case_fold) } else { match super::casefold::case_fold(key).as_str() { @@ -659,10 +659,12 @@ mod tests { false, RequestParams::default(), ); - assert_eq!( - ctx.resolve_key("dynamodb:ResourceTag/Team"), - Some(vec!["Alpha"]) - ); + assert_eq!(ctx.resolve_key("aws:ResourceTag/Team"), Some(vec!["Alpha"])); + // dynamodb:ResourceTag/* is absent, as on the live service, so it keeps + // absent-key semantics (StringEquals never matches, StringNotEquals + // matches) and never fails closed. + assert_eq!(ctx.resolve_key("dynamodb:ResourceTag/Team"), None); + assert!(!ctx.is_unresolvable_key("dynamodb:ResourceTag/Team")); } #[test] @@ -716,10 +718,7 @@ mod tests { ctx.resolve_key("aws:PrincipalTag/TEAM"), Some(vec!["contractors"]) ); - assert_eq!( - ctx.resolve_key("dynamodb:ResourceTag/env"), - Some(vec!["Prod"]) - ); + assert_eq!(ctx.resolve_key("aws:ResourceTag/env"), Some(vec!["Prod"])); // The value is not case-folded, only the key. assert_eq!(ctx.resolve_key("aws:PrincipalTag/missing"), None); } @@ -772,10 +771,8 @@ mod tests { Some(vec!["contractors"]) ); assert_eq!(ctx.resolve_key("aws:resourcetag/ENV"), Some(vec!["prod"])); - assert_eq!( - ctx.resolve_key("dynamodb:resourcetag/env"), - Some(vec!["prod"]) - ); + // dynamodb:ResourceTag/* is absent in any case, as on the service. + assert_eq!(ctx.resolve_key("dynamodb:resourcetag/env"), None); } #[test] diff --git a/crates/core/src/settings_keys.rs b/crates/core/src/settings_keys.rs index ee7a1f743..5380da38d 100644 --- a/crates/core/src/settings_keys.rs +++ b/crates/core/src/settings_keys.rs @@ -101,6 +101,23 @@ pub const VECTOR_INDEX_MIN_CREATING_MS: &str = "vector_index_min_creating_ms"; /// Unset, nothing waits and no branch is taken. pub const VECTOR_ALLOCATION_PHASE_DELAY_MS: &str = "vector_allocation_phase_delay_ms"; +/// Marker recording that the resource-tag condition-key migration has run on +/// this catalog, or that the catalog was created after the migration existed +/// and so has nothing to rewrite. +/// +/// `extenddb migrate --yes` writes it after rewriting every stored policy from +/// `dynamodb:ResourceTag/*` to `aws:ResourceTag/*` with no storage errors; +/// `init` and every fresh-catalog bootstrap write it on a new catalog. `serve` +/// reads this one row at startup and refuses to start when it is absent, naming +/// `extenddb migrate --yes`, so a pre-migration catalog is never served under +/// the new evaluator (which leaves `dynamodb:ResourceTag/*` absent). The value +/// is informational; presence of the key is the signal. +pub const RESOURCE_TAG_MIGRATION: &str = "resource_tag_migration"; + +/// The value written for [`RESOURCE_TAG_MIGRATION`] when the migration is +/// complete. +pub const RESOURCE_TAG_MIGRATION_DONE: &str = "applied"; + /// Resolve a caller-supplied settings key to its canonical name. /// /// Accepting the old name keeps `extenddb settings set gsi_propagation_delay_ms 0` diff --git a/crates/server/src/serve.rs b/crates/server/src/serve.rs index 802be8fca..9ebbaccd4 100644 --- a/crates/server/src/serve.rs +++ b/crates/server/src/serve.rs @@ -308,6 +308,26 @@ async fn serve_inner(params: ServeParams, port: u16) -> anyhow::Result<()> { let cred_store = components.credential_store; let runtime_hooks = components.runtime_hooks; + // Resource-tag migration guard. ExtendDB evaluates resource tags only as + // aws:ResourceTag/* and leaves dynamodb:ResourceTag/* absent, so a catalog + // whose stored policies still name the dynamodb: form must be rewritten by + // `extenddb migrate --yes` before this binary serves it. `migrate`, `init` + // and every fresh-catalog bootstrap record a settings marker; refuse to + // start when it is missing. One settings-row read, no policy scan. + if catalog_store + .get_setting(extenddb_core::settings_keys::RESOURCE_TAG_MIGRATION) + .await + .map_err(|e| anyhow::anyhow!("reading the resource-tag migration marker: {e:?}"))? + .is_none() + { + anyhow::bail!( + "this catalog has not had the resource-tag condition-key migration applied. \ + Run `extenddb migrate --yes` before starting the server. It rewrites stored \ + policies from dynamodb:ResourceTag to aws:ResourceTag, which this build \ + evaluates, and records a marker the server checks at startup." + ); + } + // Dev mode: seed the credentials the SDK may sign with and verify against // them (see `dev_credentials` for the vending contract). Seeded here so it // works for both file-backed and bootstrap-on-serve (in-memory) diff --git a/crates/storage-cassandra/migrations/catalog/V001__initial_schema.cql b/crates/storage-cassandra/migrations/catalog/V001__initial_schema.cql index 98289e2a3..001d9fd14 100644 --- a/crates/storage-cassandra/migrations/catalog/V001__initial_schema.cql +++ b/crates/storage-cassandra/migrations/catalog/V001__initial_schema.cql @@ -210,3 +210,4 @@ CREATE TABLE IF NOT EXISTS iam_permissions_boundaries ( -- Seed settings INSERT INTO settings (key, value) VALUES ('catalog_version', '0.0.1'); INSERT INTO settings (key, value) VALUES ('control_plane_delay_seconds', '0.25'); +INSERT INTO settings (key, value) VALUES ('resource_tag_migration', 'applied'); diff --git a/crates/storage-cassandra/src/lib.rs b/crates/storage-cassandra/src/lib.rs index d2af4aa67..f668fac6d 100644 --- a/crates/storage-cassandra/src/lib.rs +++ b/crates/storage-cassandra/src/lib.rs @@ -215,6 +215,20 @@ pub fn backend() -> extenddb_storage::Backend { >) }) }, + management_store: |connection_string| { + let connection_string = connection_string.to_string(); + Box::pin(async move { + let catalog_store = make_catalog_store_from_connection_string(&connection_string) + .await + .map_err( + extenddb_storage::management_store_factory::ManagementStoreError::ConnectionFailed, + )?; + Ok(Box::new(catalog_store) + as Box< + dyn extenddb_storage::management_store::ManagementStore, + >) + }) + }, diagnostics_store: |connection_string| { let connection_string = connection_string.to_string(); Box::pin(async move { diff --git a/crates/storage-cassandra/src/management_store/access_keys.rs b/crates/storage-cassandra/src/management_store/access_keys.rs index 8720996cc..271d6d4ba 100755 --- a/crates/storage-cassandra/src/management_store/access_keys.rs +++ b/crates/storage-cassandra/src/management_store/access_keys.rs @@ -269,6 +269,102 @@ impl CassandraCatalogStore { .await } + /// List stored session policies for the resource-tag migration. + pub(crate) async fn list_session_policies_impl( + &self, + ) -> OpResult> { + let query = format!( + "SELECT session_token, account_id, role_name, session_name, session_policy \ + FROM {}.iam_sessions", + self.catalog_keyspace() + ); + let rows = crate::cassandra_util::query_rows( + self.session(), + &query, + cdrs_tokio::query_values!(), + "list_session_policies", + ) + .await?; + let raw: Vec<(String, String, String, String, Option)> = + crate::cassandra_util::map_rows( + rows, + |row| { + let session_token = crate::cassandra_util::get_column( + row, + "session_token", + "list_session_policies", + )?; + let account_id = crate::cassandra_util::get_column( + row, + "account_id", + "list_session_policies", + )?; + let role_name = crate::cassandra_util::get_column( + row, + "role_name", + "list_session_policies", + )?; + let session_name = crate::cassandra_util::get_column( + row, + "session_name", + "list_session_policies", + )?; + let session_policy: Option = row.get_r_by_name("session_policy").ok(); + Ok(( + session_token, + account_id, + role_name, + session_name, + session_policy, + )) + }, + "list_session_policies", + )?; + let mut out = Vec::new(); + for (session_token, account_id, role_name, session_name, policy) in raw { + let Some(text) = policy else { continue }; + // A session policy that does not parse is a corrupt row. It + // cannot be evaluated under either key form, so no decision + // depends on the migration; report it and carry on. + match serde_json::from_str::(&text) { + Ok(policy) => out.push(extenddb_storage::management_store::SessionPolicyEntry { + session_token, + account_id, + role_name, + session_name, + policy, + }), + Err(_) => tracing::warn!( + account_id = %account_id, + role_name = %role_name, + session_name = %session_name, + "stored session policy does not parse; skipped by the resource-tag migration" + ), + } + } + Ok(out) + } + + /// Replace a stored session's policy for the resource-tag migration. + pub(crate) async fn update_session_policy_impl( + &self, + session_token: &str, + document: &serde_json::Value, + ) -> OpResult<()> { + let query = format!( + "UPDATE {}.iam_sessions SET session_policy = ? WHERE session_token = ?", + self.catalog_keyspace() + ); + let document = document.to_string(); + crate::cassandra_util::execute( + self.session(), + &query, + cdrs_tokio::query_values!(document.as_str(), session_token), + "update_session_policy", + ) + .await + } + // ── Caller tags ──────────────────────────────────────────────── pub(crate) async fn fetch_caller_tags_impl( diff --git a/crates/storage-cassandra/src/management_store/mod.rs b/crates/storage-cassandra/src/management_store/mod.rs index d83195199..a38bfb5e8 100755 --- a/crates/storage-cassandra/src/management_store/mod.rs +++ b/crates/storage-cassandra/src/management_store/mod.rs @@ -576,6 +576,27 @@ impl extenddb_storage::management_store::ManagementStore for CassandraCatalogSto }) } + // ── Session policies (resource-tag migration) ───────────────── + + fn list_session_policies( + &self, + ) -> BoxFuture<'_, OpResult>> { + Box::pin(async move { self.list_session_policies_impl().await }) + } + + fn update_session_policy( + &self, + session_token: &str, + document: &serde_json::Value, + ) -> BoxFuture<'_, OpResult<()>> { + let session_token = session_token.to_string(); + let document = document.clone(); + Box::pin(async move { + self.update_session_policy_impl(&session_token, &document) + .await + }) + } + // ── Caller tags ──────────────────────────────────────────────── fn fetch_caller_tags( diff --git a/crates/storage-mongodb/src/bootstrapper.rs b/crates/storage-mongodb/src/bootstrapper.rs index 32ecebb47..c37c3f36f 100644 --- a/crates/storage-mongodb/src/bootstrapper.rs +++ b/crates/storage-mongodb/src/bootstrapper.rs @@ -352,6 +352,18 @@ impl Bootstrapper for MongoBootstrapper { .upsert(true) .await; + // Seed the resource-tag migration marker on this fresh catalog so the + // server's startup guard passes. An existing catalog created before + // this change has no such row, so serve refuses it until + // `extenddb migrate --yes` records it. + let _ = settings + .update_one( + doc! { "_id": "resource_tag_migration" }, + doc! { "$setOnInsert": { "value": "applied" } }, + ) + .upsert(true) + .await; + // Record migration let schema_history = db.collection::("schema_history"); let _ = schema_history diff --git a/crates/storage-mongodb/src/lib.rs b/crates/storage-mongodb/src/lib.rs index fe57bb3e9..247dd782e 100644 --- a/crates/storage-mongodb/src/lib.rs +++ b/crates/storage-mongodb/src/lib.rs @@ -211,6 +211,22 @@ pub fn backend() -> extenddb_storage::Backend { >) }) }, + management_store: |connection_string| { + let connection_string = connection_string.to_string(); + Box::pin(async move { + let client = connect_guarded(&connection_string, None, false) + .await + .map_err(|e| { + extenddb_storage::management_store_factory::ManagementStoreError::ConnectionFailed( + e.to_string(), + ) + })?; + Ok(Box::new(MongoCatalogStore::new(client)) + as Box< + dyn extenddb_storage::management_store::ManagementStore, + >) + }) + }, diagnostics_store: |connection_string| { let connection_string = connection_string.to_string(); Box::pin(async move { diff --git a/crates/storage-mongodb/src/management_store.rs b/crates/storage-mongodb/src/management_store.rs index bf38a1395..91e60e2b9 100644 --- a/crates/storage-mongodb/src/management_store.rs +++ b/crates/storage-mongodb/src/management_store.rs @@ -10,7 +10,8 @@ use mongodb::options::{FindOptions, UpdateOptions}; use extenddb_storage::management_store::{ AccessKeyCreated, AccountDetail, AdminEntry, GroupDetail, GroupListEntry, ManagementStore, - MetricsRow, OpError, OpResult, RoleDetail, RoleListEntry, UserDetail, UserListEntry, + MetricsRow, OpError, OpResult, RoleDetail, RoleListEntry, SessionPolicyEntry, UserDetail, + UserListEntry, }; use crate::catalog_store::MongoCatalogStore; @@ -1858,6 +1859,78 @@ impl ManagementStore for MongoCatalogStore { }) } + fn list_session_policies(&self) -> BoxFuture<'_, OpResult>> { + Box::pin(async move { + let coll = self.catalog_db().collection::("iam_sessions"); + let cursor = coll + .find(doc! { "session_policy": { "$exists": true } }) + .await + .map_err(|e| { + tracing::error!("list_session_policies: {e}"); + OpError::Internal("Database error".to_owned()) + })?; + let docs: Vec = cursor.try_collect().await.map_err(|e| { + tracing::error!("list_session_policies cursor: {e}"); + OpError::Internal("Database error".to_owned()) + })?; + let mut out = Vec::new(); + for d in docs { + let Some(policy_bson) = d.get("session_policy") else { + continue; + }; + // A session policy that does not convert is a corrupt row. It + // cannot be evaluated under either key form, so no decision + // depends on the migration; report it and carry on. + let Ok(policy) = bson::from_bson::(policy_bson.clone()) else { + tracing::warn!( + account_id = d.get_str("account_id").unwrap_or_default(), + role_name = d.get_str("role_name").unwrap_or_default(), + session_name = d.get_str("session_name").unwrap_or_default(), + "stored session policy does not parse; skipped by the resource-tag migration" + ); + continue; + }; + if let (Ok(session_token), Ok(account_id), Ok(role_name), Ok(session_name)) = ( + d.get_str("session_token"), + d.get_str("account_id"), + d.get_str("role_name"), + d.get_str("session_name"), + ) { + out.push(SessionPolicyEntry { + session_token: session_token.to_owned(), + account_id: account_id.to_owned(), + role_name: role_name.to_owned(), + session_name: session_name.to_owned(), + policy, + }); + } + } + Ok(out) + }) + } + + fn update_session_policy( + &self, + session_token: &str, + document: &serde_json::Value, + ) -> BoxFuture<'_, OpResult<()>> { + let session_token = session_token.to_owned(); + let policy_bson = bson::to_bson(document).unwrap_or(bson::Bson::Null); + Box::pin(async move { + let coll = self.catalog_db().collection::("iam_sessions"); + coll.update_one( + doc! { "session_token": &session_token }, + doc! { "$set": { "session_policy": policy_bson } }, + ) + .await + .map_err(|e| { + tracing::error!("update_session_policy: {e}"); + OpError::Internal("Database error".to_owned()) + })?; + Ok(()) + }) + } + fn fetch_caller_tags( &self, account_id: &str, diff --git a/crates/storage-postgres/migrations/001_schema.sql b/crates/storage-postgres/migrations/001_schema.sql index 8ad5fea50..3203651f4 100644 --- a/crates/storage-postgres/migrations/001_schema.sql +++ b/crates/storage-postgres/migrations/001_schema.sql @@ -317,5 +317,7 @@ INSERT INTO settings (key, value) VALUES ('control_plane_delay_seconds', '0.25') ON CONFLICT (key) DO NOTHING; INSERT INTO settings (key, value) VALUES ('index_propagation_delay_ms', '10') ON CONFLICT (key) DO NOTHING; +INSERT INTO settings (key, value) VALUES ('resource_tag_migration', 'applied') +ON CONFLICT (key) DO NOTHING; COMMIT; diff --git a/crates/storage-postgres/src/lib.rs b/crates/storage-postgres/src/lib.rs index 0d7a51c0d..cd8ae9887 100755 --- a/crates/storage-postgres/src/lib.rs +++ b/crates/storage-postgres/src/lib.rs @@ -106,6 +106,22 @@ pub fn backend() -> extenddb_storage::Backend { >) }) }, + management_store: |connection_string| { + let connection_string = connection_string.to_string(); + Box::pin(async move { + let pool = sqlx::PgPool::connect(&connection_string) + .await + .map_err(|e| { + extenddb_storage::management_store_factory::ManagementStoreError::ConnectionFailed( + e.to_string(), + ) + })?; + Ok(Box::new(PostgresCatalogStore::new(pool)) + as Box< + dyn extenddb_storage::management_store::ManagementStore, + >) + }) + }, diagnostics_store: |connection_string| { let connection_string = connection_string.to_string(); Box::pin(async move { diff --git a/crates/storage-postgres/src/management_store/access_keys.rs b/crates/storage-postgres/src/management_store/access_keys.rs index 578e91c30..86002f3e9 100755 --- a/crates/storage-postgres/src/management_store/access_keys.rs +++ b/crates/storage-postgres/src/management_store/access_keys.rs @@ -200,6 +200,54 @@ impl PostgresCatalogStore { Ok(()) } + /// List stored session policies for the resource-tag migration. + pub(crate) async fn list_session_policies_impl( + &self, + ) -> OpResult> { + let rows: Vec<(String, String, String, String, serde_json::Value)> = sqlx::query_as( + "SELECT session_token, account_id, role_name, session_name, session_policy \ + FROM iam_sessions WHERE session_policy IS NOT NULL", + ) + .fetch_all(self.pool()) + .await + .map_err(|e| { + tracing::error!("list_session_policies failed: {e}"); + OpError::Internal("Database error".to_owned()) + })?; + Ok(rows + .into_iter() + .map( + |(session_token, account_id, role_name, session_name, policy)| { + extenddb_storage::management_store::SessionPolicyEntry { + session_token, + account_id, + role_name, + session_name, + policy, + } + }, + ) + .collect()) + } + + /// Replace a stored session's policy for the resource-tag migration. + pub(crate) async fn update_session_policy_impl( + &self, + session_token: &str, + document: &serde_json::Value, + ) -> OpResult<()> { + sqlx::query("UPDATE iam_sessions SET session_policy = $1 WHERE session_token = $2") + .bind(document) + .bind(session_token) + .execute(self.pool()) + .await + .map_err(|e| { + tracing::error!("update_session_policy failed: {e}"); + OpError::Internal("Database error".to_owned()) + })?; + Ok(()) + } + // ── Caller tags ──────────────────────────────────────────────── pub(crate) async fn fetch_caller_tags_impl( diff --git a/crates/storage-postgres/src/management_store/mod.rs b/crates/storage-postgres/src/management_store/mod.rs index 93ab71b48..28b6948f3 100755 --- a/crates/storage-postgres/src/management_store/mod.rs +++ b/crates/storage-postgres/src/management_store/mod.rs @@ -576,6 +576,27 @@ impl extenddb_storage::management_store::ManagementStore for PostgresCatalogStor }) } + // ── Session policies (resource-tag migration) ───────────────── + + fn list_session_policies( + &self, + ) -> BoxFuture<'_, OpResult>> { + Box::pin(async move { self.list_session_policies_impl().await }) + } + + fn update_session_policy( + &self, + session_token: &str, + document: &serde_json::Value, + ) -> BoxFuture<'_, OpResult<()>> { + let session_token = session_token.to_string(); + let document = document.clone(); + Box::pin(async move { + self.update_session_policy_impl(&session_token, &document) + .await + }) + } + // ── Caller tags ──────────────────────────────────────────────── fn fetch_caller_tags( diff --git a/crates/storage-sqlite/src/lib.rs b/crates/storage-sqlite/src/lib.rs index b68a7c62f..35f16c4e0 100644 --- a/crates/storage-sqlite/src/lib.rs +++ b/crates/storage-sqlite/src/lib.rs @@ -88,6 +88,7 @@ pub fn backend() -> extenddb_storage::Backend { storage_config: config::deserialize_config, operations: &operations::SqliteOperationsEngine, settings_store: sqlite_settings_store_factory, + management_store: sqlite_management_store_factory, diagnostics_store: sqlite_diagnostics_store_factory, server_components: sqlite_server_components_factory, } @@ -95,7 +96,8 @@ pub fn backend() -> extenddb_storage::Backend { use extenddb_storage::diagnostics::DiagnosticsStore; use extenddb_storage::diagnostics_store::DiagnosticsStoreError; -use extenddb_storage::management_store::SettingsStore; +use extenddb_storage::management_store::{ManagementStore, SettingsStore}; +use extenddb_storage::management_store_factory::ManagementStoreError; use extenddb_storage::settings_store::SettingsStoreError; use crate::sqlite_util::sqlite_url; @@ -154,6 +156,20 @@ fn sqlite_settings_store_factory( }) } +// ── Management store factory ─────────────────────────────────────────── + +fn sqlite_management_store_factory( + connection_string: &str, +) -> futures::future::BoxFuture<'static, Result, ManagementStoreError>> { + let connection_string = connection_string.to_owned(); + Box::pin(async move { + let pool = catalog_pool(&connection_string) + .await + .map_err(|e| ManagementStoreError::ConnectionFailed(e.to_string()))?; + Ok(Box::new(SqliteCatalogStore::new(pool)) as Box) + }) +} + // ── Diagnostics store factory ────────────────────────────────────────── fn sqlite_diagnostics_store_factory( diff --git a/crates/storage-sqlite/src/management_store/mod.rs b/crates/storage-sqlite/src/management_store/mod.rs index e9a9d5688..3222d325f 100644 --- a/crates/storage-sqlite/src/management_store/mod.rs +++ b/crates/storage-sqlite/src/management_store/mod.rs @@ -18,7 +18,7 @@ mod users; use extenddb_storage::management_store::{ AccessKeyCreated, AccountDetail, GroupDetail, GroupListEntry, ManagementStore, OpError, - OpResult, RoleDetail, RoleListEntry, UserDetail, UserListEntry, + OpResult, RoleDetail, RoleListEntry, SessionPolicyEntry, UserDetail, UserListEntry, }; use futures::future::BoxFuture; use time::OffsetDateTime; @@ -581,6 +581,69 @@ impl ManagementStore for SqliteCatalogStore { }) } + // ── Session policies (resource-tag migration) ───────────────── + + fn list_session_policies(&self) -> BoxFuture<'_, OpResult>> { + Box::pin(async move { + let rows: Vec<(String, String, String, String, Option)> = sqlx::query_as( + "SELECT session_token, account_id, role_name, session_name, session_policy \ + FROM iam_sessions WHERE session_policy IS NOT NULL", + ) + .fetch_all(self.pool()) + .await + .map_err(|e| { + tracing::error!("list_session_policies: {e}"); + OpError::Internal("Database error".to_owned()) + })?; + + let mut out = Vec::new(); + for (session_token, account_id, role_name, session_name, policy) in rows { + let Some(text) = policy else { continue }; + // A session policy that does not parse is a corrupt row. It + // cannot be evaluated under either key form, so no decision + // depends on the migration; report it and carry on. + match serde_json::from_str::(&text) { + Ok(policy) => out.push(SessionPolicyEntry { + session_token, + account_id, + role_name, + session_name, + policy, + }), + Err(_) => tracing::warn!( + account_id = %account_id, + role_name = %role_name, + session_name = %session_name, + "stored session policy does not parse; skipped by the resource-tag migration" + ), + } + } + Ok(out) + }) + } + + fn update_session_policy( + &self, + session_token: &str, + document: &serde_json::Value, + ) -> BoxFuture<'_, OpResult<()>> { + let session_token = session_token.to_owned(); + let document = document.to_string(); + Box::pin(async move { + let _writer = self.lock_writes().await; + sqlx::query("UPDATE iam_sessions SET session_policy = ? WHERE session_token = ?") + .bind(&document) + .bind(&session_token) + .execute(self.pool()) + .await + .map(|_| ()) + .map_err(|e| { + tracing::error!("update_session_policy: {e}"); + OpError::Internal("Database error".to_owned()) + }) + }) + } + // ── Caller tags (trust-policy evaluation) ────────────────────── fn fetch_caller_tags( @@ -637,3 +700,183 @@ impl ManagementStore for SqliteCatalogStore { }) } } + +#[cfg(test)] +mod migration_tests { + use super::*; + use extenddb_storage::management_store::ManagementStore; + use extenddb_storage::policy_migration::migrate_resource_tag_condition_keys; + use serde_json::json; + use sqlx::sqlite::SqlitePoolOptions; + + async fn store() -> SqliteCatalogStore { + // One connection keeps the in-memory database alive for the test. + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .expect("open in-memory sqlite"); + sqlx::query("PRAGMA foreign_keys = ON") + .execute(&pool) + .await + .unwrap(); + crate::schema::apply(&pool).await.expect("apply schema"); + SqliteCatalogStore::new(pool) + } + + fn policy(resource_tag_key: &str) -> serde_json::Value { + json!({ + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Allow", + "Action": "dynamodb:GetItem", + "Resource": "*", + "Condition": {"StringEquals": {resource_tag_key: "prod"}} + }] + }) + } + + #[tokio::test] + async fn rewrites_inline_boundary_and_session_policies() { + let s = store().await; + s.create_account("111111111111", "acct").await.unwrap(); + s.create_user("111111111111", "alice", None).await.unwrap(); + s.put_policy( + "111111111111", + "user", + "alice", + "p1", + &policy("dynamodb:ResourceTag/env"), + ) + .await + .unwrap(); + s.set_user_boundary("111111111111", "alice", &policy("dynamodb:ResourceTag/env")) + .await + .unwrap(); + s.create_role( + "111111111111", + "r1", + &json!({"Version": "2012-10-17", "Statement": []}), + ) + .await + .unwrap(); + s.store_session( + "tok-1", + "AKIAEXAMPLE000000001", + b"enc", + "111111111111", + "r1", + "sess", + &None, + &Some(policy("dynamodb:ResourceTag/env")), + time::OffsetDateTime::now_utc() + time::Duration::hours(1), + ) + .await + .unwrap(); + + // Detect mode writes nothing. + let detect = migrate_resource_tag_condition_keys(&s, false).await; + assert_eq!(detect.errors, 0); + assert_eq!(detect.rewritten.len(), 3, "inline + boundary + session"); + assert!(detect.needs_attention.is_empty()); + // Still on the old key. + let before = s + .list_policies("111111111111", "user", "alice") + .await + .unwrap(); + let p1_before = before.iter().find(|(n, ..)| n == "p1").unwrap(); + assert!( + p1_before + .1 + .pointer("/Statement/0/Condition/StringEquals") + .unwrap() + .as_object() + .unwrap() + .contains_key("dynamodb:ResourceTag/env") + ); + + // Apply mode writes. + let applied = migrate_resource_tag_condition_keys(&s, true).await; + assert_eq!(applied.errors, 0); + assert_eq!(applied.rewritten.len(), 3); + + let after = s + .list_policies("111111111111", "user", "alice") + .await + .unwrap(); + let p1_after = after.iter().find(|(n, ..)| n == "p1").unwrap(); + assert!( + p1_after + .1 + .pointer("/Statement/0/Condition/StringEquals") + .unwrap() + .as_object() + .unwrap() + .contains_key("aws:ResourceTag/env") + ); + let boundary = s + .get_user_boundary("111111111111", "alice") + .await + .unwrap() + .unwrap(); + assert!( + boundary + .pointer("/Statement/0/Condition/StringEquals") + .unwrap() + .as_object() + .unwrap() + .contains_key("aws:ResourceTag/env") + ); + let sessions = s.list_session_policies().await.unwrap(); + assert_eq!(sessions.len(), 1); + assert!( + sessions[0] + .policy + .pointer("/Statement/0/Condition/StringEquals") + .unwrap() + .as_object() + .unwrap() + .contains_key("aws:ResourceTag/env") + ); + + // Idempotent: a second pass changes nothing. + let again = migrate_resource_tag_condition_keys(&s, true).await; + assert_eq!(again.errors, 0); + assert!(again.rewritten.is_empty()); + } + + #[tokio::test] + async fn collision_is_reported_and_left_unchanged() { + let s = store().await; + s.create_account("222222222222", "acct").await.unwrap(); + s.create_user("222222222222", "bob", None).await.unwrap(); + let collide = json!({ + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Allow", + "Action": "dynamodb:GetItem", + "Resource": "*", + "Condition": {"StringEquals": { + "aws:ResourceTag/env": "prod", + "dynamodb:ResourceTag/env": "dev" + }} + }] + }); + s.put_policy("222222222222", "user", "bob", "p", &collide) + .await + .unwrap(); + + let report = migrate_resource_tag_condition_keys(&s, true).await; + assert_eq!(report.errors, 0); + assert!(report.rewritten.is_empty()); + assert_eq!(report.needs_attention.len(), 1); + assert_eq!(report.needs_attention[0].principal, "user/bob"); + // Unchanged on disk. + let stored = s + .list_policies("222222222222", "user", "bob") + .await + .unwrap(); + let p = stored.iter().find(|(n, ..)| n == "p").unwrap(); + assert_eq!(p.1, collide); + } +} diff --git a/crates/storage-sqlite/src/schema.rs b/crates/storage-sqlite/src/schema.rs index f28532e8e..0000689d6 100644 --- a/crates/storage-sqlite/src/schema.rs +++ b/crates/storage-sqlite/src/schema.rs @@ -430,6 +430,7 @@ INSERT INTO settings (key, value) VALUES ('catalog_version', '0.0.3') ON CONFLICT(key) DO UPDATE SET value = excluded.value; INSERT OR IGNORE INTO settings (key, value) VALUES ('control_plane_delay_seconds', '0.25'); INSERT OR IGNORE INTO settings (key, value) VALUES ('index_propagation_delay_ms', '10'); +INSERT OR IGNORE INTO settings (key, value) VALUES ('resource_tag_migration', 'applied'); "#; /// Apply the full catalog schema to a fresh (or existing) database. @@ -486,4 +487,37 @@ mod tests { "catalog_version must not be seeded with INSERT OR IGNORE" ); } + + /// A fresh catalog seeds the resource-tag migration marker, so a + /// newly-initialised deployment serves without a separate migrate step. + /// An older catalog that lacks it is caught by the server's startup guard. + #[test] + fn the_schema_seeds_the_resource_tag_migration_marker() { + assert!( + SCHEMA_SQL.contains( + "INSERT OR IGNORE INTO settings (key, value) VALUES ('resource_tag_migration'" + ), + "schema must seed the resource_tag_migration marker on a fresh catalog" + ); + } + + #[tokio::test] + async fn applying_the_schema_records_the_marker_row() { + use sqlx::sqlite::SqlitePoolOptions; + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .unwrap(); + super::apply(&pool).await.unwrap(); + let value: Option<(String,)> = + sqlx::query_as("SELECT value FROM settings WHERE key = 'resource_tag_migration'") + .fetch_optional(&pool) + .await + .unwrap(); + assert_eq!( + value.map(|(v,)| v).as_deref(), + Some(extenddb_core::settings_keys::RESOURCE_TAG_MIGRATION_DONE) + ); + } } diff --git a/crates/storage/src/backend.rs b/crates/storage/src/backend.rs index 8eca6603f..54634257e 100644 --- a/crates/storage/src/backend.rs +++ b/crates/storage/src/backend.rs @@ -51,6 +51,8 @@ pub struct Backend { pub operations: &'static dyn OperationsEngine, /// Creates the runtime settings store. pub settings_store: SettingsStoreFactory, + /// Creates a catalog-only management store (used by `migrate`). + pub management_store: crate::management_store_factory::ManagementStoreFactory, /// Creates the diagnostics store (`catalog-check`, `verify`). pub diagnostics_store: DiagnosticsStoreFactory, /// Creates the assembled server components for `serve`. diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index 74ff83f4e..f83b42c64 100755 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -16,7 +16,9 @@ pub mod diagnostics_store; pub mod error; pub mod hooks; pub mod management_store; +pub mod management_store_factory; pub mod operations; +pub mod policy_migration; pub mod server_components; pub mod settings_store; pub mod transact; diff --git a/crates/storage/src/management_store/mod.rs b/crates/storage/src/management_store/mod.rs index 789c9b18c..c0b753845 100755 --- a/crates/storage/src/management_store/mod.rs +++ b/crates/storage/src/management_store/mod.rs @@ -26,7 +26,7 @@ mod types; pub use types::{ AccessKeyCreated, AccountDetail, AdminEntry, GroupDetail, MetricsRow, OpError, OpResult, - RoleDetail, UserDetail, + RoleDetail, SessionPolicyEntry, UserDetail, }; use futures::future::BoxFuture; @@ -447,6 +447,28 @@ pub trait ManagementStore: Send + Sync { expires_at: time::OffsetDateTime, ) -> BoxFuture<'_, OpResult<()>>; + /// List every stored session that carries a session policy, for the + /// resource-tag policy migration. + /// + /// The default returns an empty list, which is correct for a store that + /// keeps no session policies. Backends that store session policies + /// override this so the migration can rewrite them. + fn list_session_policies(&self) -> BoxFuture<'_, OpResult>> { + Box::pin(async { Ok(Vec::new()) }) + } + + /// Replace a stored session's policy document, keyed by `session_token`. + /// Used only by the resource-tag policy migration. The default is a no-op + /// for a store that keeps no session policies. + fn update_session_policy( + &self, + session_token: &str, + document: &serde_json::Value, + ) -> BoxFuture<'_, OpResult<()>> { + let _ = (session_token, document); + Box::pin(async { Ok(()) }) + } + // ── Caller tags (for trust policy evaluation) ────────────────── /// Fetch tags for a caller principal identified by ARN. diff --git a/crates/storage/src/management_store/types.rs b/crates/storage/src/management_store/types.rs index 1c39d8995..a8aab735a 100755 --- a/crates/storage/src/management_store/types.rs +++ b/crates/storage/src/management_store/types.rs @@ -23,6 +23,20 @@ pub enum OpError { /// Result alias for management operations. pub type OpResult = Result; +/// A stored session's policy, returned by +/// [`super::ManagementStore::list_session_policies`] for the policy migration. +/// +/// Only sessions that carry a session policy are returned. Carries the identity +/// needed to rewrite the policy and report it, keyed by `session_token`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SessionPolicyEntry { + pub session_token: String, + pub account_id: String, + pub role_name: String, + pub session_name: String, + pub policy: serde_json::Value, +} + // ── Detail structs ───────────────────────────────────────────────────── /// Account detail returned by [`super::ManagementStore::get_account_detail`]. diff --git a/crates/storage/src/management_store_factory.rs b/crates/storage/src/management_store_factory.rs new file mode 100644 index 000000000..a681b0dd3 --- /dev/null +++ b/crates/storage/src/management_store_factory.rs @@ -0,0 +1,54 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Management store factory registry for backend-agnostic instantiation. +//! +//! The `migrate` command drives the resource-tag policy rewrite over a +//! [`ManagementStore`] without standing up a full server. This factory opens a +//! catalog-only store from a connection string, mirroring +//! [`crate::settings_store`]. + +use crate::management_store::ManagementStore; +use futures::future::BoxFuture; + +/// Error type for management store creation. +#[derive(Debug)] +pub enum ManagementStoreError { + /// No storage backend has been installed (`set_backend` was not called). + BackendNotInstalled, + ConnectionFailed(String), +} + +impl std::fmt::Display for ManagementStoreError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::BackendNotInstalled => { + write!( + f, + "no storage backend installed (set_backend was not called)" + ) + } + Self::ConnectionFailed(msg) => write!(f, "Failed to connect: {msg}"), + } + } +} + +impl std::error::Error for ManagementStoreError {} + +/// Factory function type for creating management stores. +pub type ManagementStoreFactory = + fn(&str) -> BoxFuture<'static, Result, ManagementStoreError>>; + +/// Create a management store for the installed backend. +/// +/// # Errors +/// +/// Returns [`ManagementStoreError::BackendNotInstalled`] when no backend has +/// been installed, or [`ManagementStoreError::ConnectionFailed`] on a +/// connection error. +pub async fn create_management_store( + connection_string: &str, +) -> Result, ManagementStoreError> { + let backend = crate::backend::try_backend().ok_or(ManagementStoreError::BackendNotInstalled)?; + (backend.management_store)(connection_string).await +} diff --git a/crates/storage/src/policy_migration.rs b/crates/storage/src/policy_migration.rs new file mode 100644 index 000000000..3ba62d979 --- /dev/null +++ b/crates/storage/src/policy_migration.rs @@ -0,0 +1,1004 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Rewrite stored IAM policies from `dynamodb:ResourceTag/` to +//! `aws:ResourceTag/`. +//! +//! The live service evaluates resource tags only as `aws:ResourceTag/` and +//! leaves `dynamodb:ResourceTag/` absent, so ExtendDB no longer resolves +//! the `dynamodb:` form (see `extenddb_auth::policy::context`). A policy that +//! was written against the `dynamodb:` form therefore changed meaning on +//! upgrade. The rewrite restores the author's intent: the `dynamodb:` form and +//! the `aws:` form resolved identically before the evaluator change, so +//! replacing the former with the latter keeps every policy's decision while +//! moving it onto the key the service (and this build) actually evaluates. +//! +//! This module has two layers: +//! - [`rewrite_policy_document`], a pure transform over a single policy +//! document (unit-tested per operator form), and +//! - [`migrate_resource_tag_condition_keys`], which drives it over every stored +//! policy reachable through [`ManagementStore`], backend-independently. + +use serde_json::Value; + +use crate::management_store::{ManagementStore, SessionPolicyEntry}; + +/// The condition-key prefix the live service does not populate. +const DYNAMODB_PREFIX: &str = "dynamodb:ResourceTag/"; +/// The condition-key prefix the live service evaluates resource tags as. +const AWS_PREFIX: &str = "aws:ResourceTag/"; + +/// ASCII-case-insensitive prefix strip, returning the untouched remainder. +fn strip_prefix_ci<'a>(s: &'a str, prefix: &str) -> Option<&'a str> { + let head = s.get(..prefix.len())?; + head.eq_ignore_ascii_case(prefix) + .then(|| &s[prefix.len()..]) +} + +/// Outcome of rewriting one policy document. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RewriteOutcome { + /// The rewritten document. Equal to the input when nothing changed or a + /// collision was found. + pub document: Value, + /// Number of rewrites applied (condition-key renames plus policy-variable + /// prefix rewrites). Zero when nothing changed or on a collision. + pub rewrites: usize, + /// A `Condition` operator block held both `aws:ResourceTag/k` and a + /// `dynamodb:ResourceTag/k` that would rewrite onto the same JSON key. + /// The two cannot share one JSON object, so the document is left unchanged + /// and the caller reports it for manual attention. + pub collision: bool, +} + +impl RewriteOutcome { + /// Whether the document was rewritten (changed and no collision). + #[must_use] + pub fn changed(&self) -> bool { + self.rewrites > 0 && !self.collision + } +} + +/// Rewrite every `dynamodb:ResourceTag/` condition key and +/// `${dynamodb:ResourceTag/...}` policy variable (any ASCII case) in a policy +/// document to the `aws:ResourceTag/` form, inside `Condition` blocks only. +/// +/// Idempotent: a document with no `dynamodb:ResourceTag` reference is returned +/// unchanged with `rewrites == 0`. On a collision the input is returned +/// unchanged with `collision == true`. +#[must_use] +pub fn rewrite_policy_document(document: &Value) -> RewriteOutcome { + let mut doc = document.clone(); + let mut rewrites = 0usize; + let mut collision = false; + + // Statement may be a single object or an array of objects. + match doc.get_mut("Statement") { + Some(Value::Array(statements)) => { + for stmt in statements.iter_mut() { + rewrite_statement(stmt, &mut rewrites, &mut collision); + } + } + Some(stmt @ Value::Object(_)) => { + rewrite_statement(stmt, &mut rewrites, &mut collision); + } + _ => {} + } + + if collision { + // Leave the stored policy exactly as it was. + return RewriteOutcome { + document: document.clone(), + rewrites: 0, + collision: true, + }; + } + + RewriteOutcome { + document: if rewrites > 0 { doc } else { document.clone() }, + rewrites, + collision: false, + } +} + +/// Rewrite the `Condition` block of one statement in place. +fn rewrite_statement(stmt: &mut Value, rewrites: &mut usize, collision: &mut bool) { + let Some(condition) = stmt.get_mut("Condition").and_then(Value::as_object_mut) else { + return; + }; + for (_operator, block) in condition.iter_mut() { + let Some(map) = block.as_object_mut() else { + continue; + }; + rewrite_operator_block(map, rewrites, collision); + } +} + +/// Rewrite one operator block: its condition keys (object keys) and the policy +/// variables in its values. +fn rewrite_operator_block( + block: &mut serde_json::Map, + rewrites: &mut usize, + collision: &mut bool, +) { + // First rewrite the variables inside the values, keyed by the current + // (pre-rename) condition key, so a value rewrite is not lost by the rename. + for value in block.values_mut() { + rewrite_variables_in_json(value, rewrites); + } + + // Collect the condition-key renames. + let mut renames: Vec<(String, String)> = Vec::new(); + for key in block.keys() { + if let Some(rest) = strip_prefix_ci(key, DYNAMODB_PREFIX) { + renames.push((key.clone(), format!("{AWS_PREFIX}{rest}"))); + } + } + if renames.is_empty() { + return; + } + // A rename onto a key the block already has (an aws:ResourceTag key, or + // an earlier rename) is a duplicate when the two values are equal: both + // prefixes resolve from the same table tags, so the two conditions are + // the same condition and the dynamodb: one is dropped. Different values + // cannot share one JSON key, so the block is a collision. + let mut moves: Vec<(String, String)> = Vec::new(); + let mut drops: Vec = Vec::new(); + for (old_key, new_key) in &renames { + let value = &block[old_key]; + let existing = block.get(new_key).or_else(|| { + moves + .iter() + .find(|(_, n)| n == new_key) + .map(|(o, _)| &block[o]) + }); + match existing { + Some(v) if v == value => drops.push(old_key.clone()), + Some(_) => { + *collision = true; + return; + } + None => moves.push((old_key.clone(), new_key.clone())), + } + } + for old_key in drops { + block.remove(&old_key); + *rewrites += 1; + } + for (old_key, new_key) in moves { + if let Some(value) = block.remove(&old_key) { + block.insert(new_key, value); + *rewrites += 1; + } + } +} + +/// Rewrite `${dynamodb:ResourceTag/...}` policy variables inside a value, which +/// is a string or an array of strings. +fn rewrite_variables_in_json(value: &mut Value, rewrites: &mut usize) { + match value { + Value::String(s) => { + if let Some(new) = rewrite_variables_in_str(s, rewrites) { + *s = new; + } + } + Value::Array(items) => { + for item in items.iter_mut() { + rewrite_variables_in_json(item, rewrites); + } + } + _ => {} + } +} + +/// Rewrite the prefix of every `${dynamodb:ResourceTag/...}` variable in `s`, +/// leaving all other text (including a literal `dynamodb:ResourceTag/...` not +/// wrapped in `${...}`) untouched. Returns `None` when nothing changed. +fn rewrite_variables_in_str(s: &str, rewrites: &mut usize) -> Option { + if !s.contains("${") { + return None; + } + let mut out = String::with_capacity(s.len()); + let mut rest = s; + let mut changed = false; + while let Some(open) = rest.find("${") { + out.push_str(&rest[..open + 2]); + let after_open = &rest[open + 2..]; + let Some(close) = after_open.find('}') else { + // No closing brace: copy the remainder and stop. + out.push_str(after_open); + rest = ""; + break; + }; + let inner = &after_open[..close]; + if let Some(tag_rest) = strip_prefix_ci(inner, DYNAMODB_PREFIX) { + out.push_str(AWS_PREFIX); + out.push_str(tag_rest); + *rewrites += 1; + changed = true; + } else { + out.push_str(inner); + } + out.push('}'); + rest = &after_open[close + 1..]; + } + out.push_str(rest); + changed.then_some(out) +} + +/// A policy that the migration could not rewrite automatically, with enough +/// identity for an operator to find and fix it by hand. Carries no policy body. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NeedsAttention { + pub account_id: String, + pub principal: String, + pub policy_name: String, +} + +/// A policy the migration would rewrite (detect mode) or did rewrite (apply +/// mode), with its identity and the number of rewrites. Carries no policy body. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Rewritten { + pub account_id: String, + pub principal: String, + pub policy_name: String, + pub rewrites: usize, +} + +/// Result of a migration pass over every stored policy. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct MigrationReport { + /// Policies that were (or, in detect mode, would be) rewritten. + pub rewritten: Vec, + /// Policies left unchanged because of a collision. + pub needs_attention: Vec, + /// Count of storage errors encountered. The pass keeps going past an + /// error so a single bad row does not hide the rest; the caller treats a + /// non-zero count as a failed migration. + pub errors: u64, +} + +impl MigrationReport { + /// Total rewrites across all policies (for the summary line). + #[must_use] + pub fn total_rewrites(&self) -> usize { + self.rewritten.iter().map(|r| r.rewrites).sum() + } +} + +/// Rewrite `dynamodb:ResourceTag/` condition keys to `aws:ResourceTag/` across +/// every stored policy reachable through `store`, backend-independently. +/// +/// Scope: inline user, group and role policies, user and role permissions +/// boundaries, and stored session policies. Role trust policies are not +/// rewritten: the `AssumeRole` path never resolved the key, so their behaviour +/// does not change. +/// +/// With `apply` false nothing is written: the report lists what would change. +/// With `apply` true each changed policy is written back. Every storage error +/// is counted and the pass continues, so an interrupted or partial run is +/// finished by the next `migrate`. Idempotent: a second pass finds nothing to +/// rewrite. No policy body is ever logged or returned. +pub async fn migrate_resource_tag_condition_keys( + store: &dyn ManagementStore, + apply: bool, +) -> MigrationReport { + let mut report = MigrationReport::default(); + + let accounts = match store.list_all_accounts().await { + Ok(a) => a, + Err(_) => { + report.errors += 1; + return report; + } + }; + + for (account_id, _name) in &accounts { + migrate_account(store, account_id, apply, &mut report).await; + } + + migrate_sessions(store, apply, &mut report).await; + + report +} + +/// Classify a policy, record it in the report, and return the rewritten +/// document when it should be written back. `principal` and `policy_name` are +/// for reporting only; no policy body is recorded. +fn record_outcome( + report: &mut MigrationReport, + account_id: &str, + principal: &str, + policy_name: &str, + document: &Value, +) -> Option { + let outcome = rewrite_policy_document(document); + if outcome.collision { + report.needs_attention.push(NeedsAttention { + account_id: account_id.to_owned(), + principal: principal.to_owned(), + policy_name: policy_name.to_owned(), + }); + return None; + } + if !outcome.changed() { + return None; + } + report.rewritten.push(Rewritten { + account_id: account_id.to_owned(), + principal: principal.to_owned(), + policy_name: policy_name.to_owned(), + rewrites: outcome.rewrites, + }); + Some(outcome.document) +} + +async fn migrate_account( + store: &dyn ManagementStore, + account_id: &str, + apply: bool, + report: &mut MigrationReport, +) { + // Inline policies on users, groups and roles. + match store.list_users(account_id).await { + Ok(users) => { + for (_a, user_name, ..) in users { + migrate_inline(store, account_id, "user", &user_name, apply, report).await; + migrate_user_boundary(store, account_id, &user_name, apply, report).await; + } + } + Err(_) => report.errors += 1, + } + match store.list_groups(account_id).await { + Ok(groups) => { + for (_a, group_name, ..) in groups { + migrate_inline(store, account_id, "group", &group_name, apply, report).await; + } + } + Err(_) => report.errors += 1, + } + match store.list_roles(account_id).await { + Ok(roles) => { + for (_a, role_name, ..) in roles { + migrate_inline(store, account_id, "role", &role_name, apply, report).await; + migrate_role_boundary(store, account_id, &role_name, apply, report).await; + } + } + Err(_) => report.errors += 1, + } +} + +async fn migrate_inline( + store: &dyn ManagementStore, + account_id: &str, + principal_type: &str, + principal_name: &str, + apply: bool, + report: &mut MigrationReport, +) { + let policies = match store + .list_policies(account_id, principal_type, principal_name) + .await + { + Ok(p) => p, + Err(_) => { + report.errors += 1; + return; + } + }; + let principal = format!("{principal_type}/{principal_name}"); + for (policy_name, document, _created) in policies { + if let Some(new_doc) = + record_outcome(report, account_id, &principal, &policy_name, &document) + && apply + && store + .put_policy( + account_id, + principal_type, + principal_name, + &policy_name, + &new_doc, + ) + .await + .is_err() + { + report.errors += 1; + } + } +} + +async fn migrate_user_boundary( + store: &dyn ManagementStore, + account_id: &str, + user_name: &str, + apply: bool, + report: &mut MigrationReport, +) { + match store.get_user_boundary(account_id, user_name).await { + Ok(Some(document)) => { + let principal = format!("user/{user_name}"); + if let Some(new_doc) = record_outcome( + report, + account_id, + &principal, + "", + &document, + ) && apply + && store + .set_user_boundary(account_id, user_name, &new_doc) + .await + .is_err() + { + report.errors += 1; + } + } + Ok(None) => {} + Err(_) => report.errors += 1, + } +} + +async fn migrate_role_boundary( + store: &dyn ManagementStore, + account_id: &str, + role_name: &str, + apply: bool, + report: &mut MigrationReport, +) { + match store.get_role_boundary(account_id, role_name).await { + Ok(Some(document)) => { + let principal = format!("role/{role_name}"); + if let Some(new_doc) = record_outcome( + report, + account_id, + &principal, + "", + &document, + ) && apply + && store + .set_role_boundary(account_id, role_name, &new_doc) + .await + .is_err() + { + report.errors += 1; + } + } + Ok(None) => {} + Err(_) => report.errors += 1, + } +} + +async fn migrate_sessions(store: &dyn ManagementStore, apply: bool, report: &mut MigrationReport) { + let sessions = match store.list_session_policies().await { + Ok(s) => s, + Err(_) => { + report.errors += 1; + return; + } + }; + for SessionPolicyEntry { + session_token, + account_id, + role_name, + session_name, + policy, + } in sessions + { + let principal = format!("assumed-role/{role_name}/{session_name}"); + if let Some(new_doc) = + record_outcome(report, &account_id, &principal, "", &policy) + && apply + && store + .update_session_policy(&session_token, &new_doc) + .await + .is_err() + { + report.errors += 1; + } + } +} +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn stmt_cond(cond: Value) -> Value { + json!({ + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Allow", + "Action": "dynamodb:GetItem", + "Resource": "*", + "Condition": cond + }] + }) + } + + // ── Condition-key rewrites, per operator form ──────────────────────── + + #[test] + fn bare_key_rewritten() { + let doc = stmt_cond(json!({"StringEquals": {"dynamodb:ResourceTag/env": "prod"}})); + let out = rewrite_policy_document(&doc); + assert!(out.changed()); + assert_eq!(out.rewrites, 1); + assert_eq!( + out.document, + stmt_cond(json!({"StringEquals": {"aws:ResourceTag/env": "prod"}})) + ); + } + + #[test] + fn negated_key_rewritten() { + let doc = stmt_cond(json!({"StringNotEquals": {"dynamodb:ResourceTag/env": "prod"}})); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 1); + assert_eq!( + out.document, + stmt_cond(json!({"StringNotEquals": {"aws:ResourceTag/env": "prod"}})) + ); + } + + #[test] + fn if_exists_key_rewritten() { + let doc = stmt_cond(json!({"StringEqualsIfExists": {"dynamodb:ResourceTag/env": "prod"}})); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 1); + assert_eq!( + out.document, + stmt_cond(json!({"StringEqualsIfExists": {"aws:ResourceTag/env": "prod"}})) + ); + } + + #[test] + fn for_all_values_key_rewritten() { + let doc = stmt_cond( + json!({"ForAllValues:StringEquals": {"dynamodb:ResourceTag/team": ["a", "b"]}}), + ); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 1); + assert_eq!( + out.document, + stmt_cond(json!({"ForAllValues:StringEquals": {"aws:ResourceTag/team": ["a", "b"]}})) + ); + } + + #[test] + fn for_any_value_if_exists_key_rewritten() { + let doc = stmt_cond( + json!({"ForAnyValue:StringEqualsIfExists": {"dynamodb:ResourceTag/team": ["a"]}}), + ); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 1); + assert_eq!( + out.document, + stmt_cond(json!({"ForAnyValue:StringEqualsIfExists": {"aws:ResourceTag/team": ["a"]}})) + ); + } + + #[test] + fn null_key_rewritten() { + let doc = stmt_cond(json!({"Null": {"dynamodb:ResourceTag/env": "true"}})); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 1); + assert_eq!( + out.document, + stmt_cond(json!({"Null": {"aws:ResourceTag/env": "true"}})) + ); + } + + #[test] + fn mixed_case_prefix_rewritten_tag_key_preserved() { + let doc = stmt_cond(json!({"StringEquals": {"DynamoDB:ResourceTag/Env": "prod"}})); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 1); + // Prefix normalised to aws:ResourceTag/, tag key "Env" case preserved. + assert_eq!( + out.document, + stmt_cond(json!({"StringEquals": {"aws:ResourceTag/Env": "prod"}})) + ); + } + + // ── Policy-variable rewrites ───────────────────────────────────────── + + #[test] + fn policy_variable_rewritten() { + let doc = stmt_cond(json!({ + "StringEquals": {"aws:PrincipalTag/team": "${dynamodb:ResourceTag/team}"} + })); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 1); + assert_eq!( + out.document, + stmt_cond(json!({ + "StringEquals": {"aws:PrincipalTag/team": "${aws:ResourceTag/team}"} + })) + ); + } + + #[test] + fn policy_variable_mixed_case_rewritten() { + let doc = stmt_cond(json!({ + "StringEquals": {"aws:PrincipalTag/team": "${DYNAMODB:resourcetag/team}"} + })); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 1); + assert_eq!( + out.document, + stmt_cond(json!({ + "StringEquals": {"aws:PrincipalTag/team": "${aws:ResourceTag/team}"} + })) + ); + } + + #[test] + fn policy_variable_in_array_and_amongst_other_text() { + let doc = stmt_cond(json!({ + "StringLike": {"aws:PrincipalTag/x": [ + "pre-${dynamodb:ResourceTag/a}-post", + "${aws:PrincipalTag/y}" + ]} + })); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 1); + assert_eq!( + out.document, + stmt_cond(json!({ + "StringLike": {"aws:PrincipalTag/x": [ + "pre-${aws:ResourceTag/a}-post", + "${aws:PrincipalTag/y}" + ]} + })) + ); + } + + #[test] + fn literal_value_not_wrapped_in_variable_is_left_alone() { + // The key name appearing as a bare value (not inside ${...}) is a + // literal comparison value, not a variable, and must not be rewritten. + let doc = stmt_cond(json!({ + "StringEquals": {"aws:PrincipalTag/x": "dynamodb:ResourceTag/env"} + })); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 0); + assert!(!out.changed()); + assert_eq!(out.document, doc); + } + + #[test] + fn key_and_variable_both_rewritten_counts_two() { + let doc = stmt_cond(json!({ + "StringEquals": {"dynamodb:ResourceTag/env": "${dynamodb:ResourceTag/env}"} + })); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 2); + assert_eq!( + out.document, + stmt_cond(json!({ + "StringEquals": {"aws:ResourceTag/env": "${aws:ResourceTag/env}"} + })) + ); + } + + // ── Collision ──────────────────────────────────────────────────────── + + #[test] + fn collision_leaves_document_unchanged() { + let doc = stmt_cond(json!({ + "StringEquals": { + "aws:ResourceTag/env": "prod", + "dynamodb:ResourceTag/env": "dev" + } + })); + let out = rewrite_policy_document(&doc); + assert!(out.collision); + assert!(!out.changed()); + assert_eq!(out.rewrites, 0); + assert_eq!( + out.document, doc, + "collision must leave the policy untouched" + ); + } + + #[test] + fn same_value_duplicate_drops_the_dynamodb_key() { + // Both prefixes resolved from the same table tags, so the two + // conditions are one condition; the dynamodb: key is dropped. + let doc = stmt_cond(json!({ + "StringEquals": { + "aws:ResourceTag/env": "prod", + "dynamodb:ResourceTag/env": "prod" + } + })); + let out = rewrite_policy_document(&doc); + assert!(!out.collision); + assert_eq!(out.rewrites, 1); + assert_eq!( + out.document, + stmt_cond(json!({"StringEquals": {"aws:ResourceTag/env": "prod"}})) + ); + } + + #[test] + fn two_dynamodb_spellings_of_one_key() { + // Same value: one aws: key remains. + let same = stmt_cond(json!({ + "StringEquals": { + "dynamodb:ResourceTag/env": "prod", + "DynamoDB:resourcetag/env": "prod" + } + })); + let out = rewrite_policy_document(&same); + assert!(!out.collision); + assert_eq!(out.rewrites, 2); + assert_eq!( + out.document, + stmt_cond(json!({"StringEquals": {"aws:ResourceTag/env": "prod"}})) + ); + // Different values: a collision, left unchanged. + let different = stmt_cond(json!({ + "StringEquals": { + "dynamodb:ResourceTag/env": "prod", + "DynamoDB:resourcetag/env": "dev" + } + })); + let out = rewrite_policy_document(&different); + assert!(out.collision); + assert_eq!(out.document, different); + } + + #[test] + fn text_outside_condition_blocks_is_never_rewritten() { + let doc = json!({ + "Version": "2012-10-17", + "Statement": [{ + "Sid": "dynamodbResourceTagenv", + "Effect": "Allow", + "Action": "dynamodb:ResourceTag/env", + "Resource": "arn:aws:dynamodb:us-east-1:123456789012:table/dynamodb:ResourceTag/env", + "Condition": {"StringEquals": {"dynamodb:ResourceTag/env": "prod"}} + }] + }); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 1); + let stmt = &out.document["Statement"][0]; + assert_eq!(stmt["Sid"], doc["Statement"][0]["Sid"]); + assert_eq!(stmt["Action"], doc["Statement"][0]["Action"]); + assert_eq!(stmt["Resource"], doc["Statement"][0]["Resource"]); + assert_eq!( + stmt["Condition"], + json!({"StringEquals": {"aws:ResourceTag/env": "prod"}}) + ); + } + + #[test] + fn no_collision_when_aws_key_is_in_a_different_operator_block() { + // aws:ResourceTag/env under StringEquals and dynamodb:ResourceTag/env + // under StringNotEquals are different JSON objects, so no collision. + let doc = stmt_cond(json!({ + "StringEquals": {"aws:ResourceTag/env": "prod"}, + "StringNotEquals": {"dynamodb:ResourceTag/env": "dev"} + })); + let out = rewrite_policy_document(&doc); + assert!(!out.collision); + assert_eq!(out.rewrites, 1); + } + + // ── Idempotency and untouched shapes ───────────────────────────────── + + #[test] + fn idempotent_second_pass_is_a_noop() { + let doc = stmt_cond(json!({"StringEquals": {"dynamodb:ResourceTag/env": "prod"}})); + let once = rewrite_policy_document(&doc); + let twice = rewrite_policy_document(&once.document); + assert_eq!(twice.rewrites, 0); + assert!(!twice.changed()); + assert_eq!(twice.document, once.document); + } + + #[test] + fn single_statement_object_form_supported() { + let doc = json!({ + "Version": "2012-10-17", + "Statement": { + "Effect": "Allow", + "Action": "dynamodb:GetItem", + "Resource": "*", + "Condition": {"StringEquals": {"dynamodb:ResourceTag/env": "prod"}} + } + }); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 1); + assert_eq!( + out.document + .pointer("/Statement/Condition/StringEquals") + .and_then(|v| v.as_object()) + .map(|m| m.contains_key("aws:ResourceTag/env")), + Some(true) + ); + } + + #[test] + fn policy_without_resource_tag_is_untouched() { + let doc = stmt_cond(json!({"StringEquals": {"aws:PrincipalTag/team": "eng"}})); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 0); + assert!(!out.changed()); + assert_eq!(out.document, doc); + } + + #[test] + fn statement_without_condition_is_untouched() { + let doc = json!({ + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": "dynamodb:GetItem", "Resource": "*"}] + }); + let out = rewrite_policy_document(&doc); + assert_eq!(out.rewrites, 0); + assert_eq!(out.document, doc); + } + + #[test] + fn total_rewrites_sums_per_policy() { + let report = MigrationReport { + rewritten: vec![ + Rewritten { + account_id: "a".into(), + principal: "user/u".into(), + policy_name: "p".into(), + rewrites: 2, + }, + Rewritten { + account_id: "a".into(), + principal: "role/r".into(), + policy_name: "q".into(), + rewrites: 3, + }, + ], + needs_attention: vec![], + errors: 0, + }; + assert_eq!(report.total_rewrites(), 5); + } + + // ── Equivalence: rewrite preserves the decision ────────────────────── + // + // On main both `aws:ResourceTag/*` and `dynamodb:ResourceTag/*` resolve + // from the resource-tag map; after the evaluator change only `aws:` does. + // The rewrite swaps the prefix, so a stored `dynamodb:` policy evaluated on + // main must reach the same decision as the rewritten `aws:` policy + // evaluated on the new semantics. These contexts model the two evaluators. + + use extenddb_auth::policy::condition::evaluate_condition; + use extenddb_auth::policy::context::ConditionContext; + use extenddb_auth::policy::document::{Condition, ConditionOperator}; + use std::collections::HashMap; + + /// Main semantics: resource tags resolve under both prefixes. + struct MainCtx { + tags: HashMap, + } + /// New semantics: resource tags resolve only as `aws:ResourceTag/*`. + struct NewCtx { + tags: HashMap, + } + + fn resolve_tag<'a>(tags: &'a HashMap, sub: &str) -> Option> { + tags.get(sub).map(|v| vec![v.as_str()]) + } + + impl ConditionContext for MainCtx { + fn resolve_key(&self, key: &str) -> Option> { + if let Some(sub) = strip_prefix_ci(key, AWS_PREFIX) { + resolve_tag(&self.tags, sub) + } else if let Some(sub) = strip_prefix_ci(key, DYNAMODB_PREFIX) { + resolve_tag(&self.tags, sub) + } else { + None + } + } + fn is_multivalued_key(&self, _key: &str) -> bool { + false + } + } + impl ConditionContext for NewCtx { + fn resolve_key(&self, key: &str) -> Option> { + strip_prefix_ci(key, AWS_PREFIX).and_then(|sub| resolve_tag(&self.tags, sub)) + } + fn is_multivalued_key(&self, _key: &str) -> bool { + false + } + } + + /// Pull the single (operator-name, key, values) out of a one-statement, + /// one-operator, one-key policy document. + fn sole_condition(doc: &Value) -> (String, String, Vec) { + let cond = doc + .pointer("/Statement/0/Condition") + .and_then(Value::as_object) + .expect("one condition block"); + let (op, block) = cond.iter().next().expect("one operator"); + let map = block.as_object().expect("operator block object"); + let (key, val) = map.iter().next().expect("one condition key"); + let values = match val { + Value::String(s) => vec![s.clone()], + Value::Array(a) => a + .iter() + .filter_map(|v| v.as_str().map(str::to_owned)) + .collect(), + _ => vec![], + }; + (op.clone(), key.clone(), values) + } + + fn build_condition(op_name: &str, key: &str, values: &[String]) -> Condition { + let base = |n: &str| match n { + "StringEquals" => ConditionOperator::StringEquals, + "StringNotEquals" => ConditionOperator::StringNotEquals, + "Null" => ConditionOperator::Null, + other => panic!("unexpected operator {other}"), + }; + let operator = if let Some(inner) = op_name.strip_prefix("ForAllValues:") { + ConditionOperator::ForAllValues(Box::new(base(inner))) + } else if let Some(inner) = op_name.strip_prefix("ForAnyValue:") { + ConditionOperator::ForAnyValue(Box::new(base(inner))) + } else if let Some(inner) = op_name.strip_suffix("IfExists") { + ConditionOperator::IfExists(Box::new(base(inner))) + } else { + base(op_name) + }; + Condition { + operator, + key: key.to_owned(), + values: values.to_vec(), + } + } + + #[test] + fn rewrite_preserves_decision_across_shapes_and_tag_states() { + // (operator block in a dynamodb:ResourceTag policy) + let shapes = [ + json!({"StringEquals": {"dynamodb:ResourceTag/env": "prod"}}), + json!({"StringNotEquals": {"dynamodb:ResourceTag/env": "prod"}}), + json!({"StringEqualsIfExists": {"dynamodb:ResourceTag/env": "prod"}}), + json!({"ForAllValues:StringEquals": {"dynamodb:ResourceTag/env": ["prod"]}}), + json!({"ForAnyValue:StringEquals": {"dynamodb:ResourceTag/env": ["prod"]}}), + json!({"Null": {"dynamodb:ResourceTag/env": "true"}}), + json!({"Null": {"dynamodb:ResourceTag/env": "false"}}), + ]; + // tag states: matching, mismatching, absent + let tag_states = [ + HashMap::from([("env".to_owned(), "prod".to_owned())]), + HashMap::from([("env".to_owned(), "dev".to_owned())]), + HashMap::new(), + ]; + + for shape in &shapes { + let original = stmt_cond(shape.clone()); + let (op_dyn, key_dyn, vals) = sole_condition(&original); + assert!(key_dyn.starts_with("dynamodb:ResourceTag/")); + + // Run the real rewrite and confirm it produced the aws: key, same + // operator and same values, and dropped the dynamodb: key. + let out = rewrite_policy_document(&original); + assert!(out.changed()); + let (op_aws, key_aws, vals_aws) = sole_condition(&out.document); + assert_eq!(op_aws, op_dyn); + assert_eq!(key_aws, "aws:ResourceTag/env"); + assert_eq!(vals_aws, vals); + + let cond_dyn = build_condition(&op_dyn, &key_dyn, &vals); + let cond_aws = build_condition(&op_aws, &key_aws, &vals_aws); + + for tags in &tag_states { + let main_ctx = MainCtx { tags: tags.clone() }; + let new_ctx = NewCtx { tags: tags.clone() }; + let before = evaluate_condition(&cond_dyn, &main_ctx); + let after = evaluate_condition(&cond_aws, &new_ctx); + assert_eq!( + before, after, + "shape {op_dyn} with tags {tags:?}: main={before} new={after}" + ); + } + } + } +} diff --git a/docs/adr/0022-trust-policy-deny-over-allow.md b/docs/adr/0022-trust-policy-deny-over-allow.md index 0c789036a..935a6fe23 100644 --- a/docs/adr/0022-trust-policy-deny-over-allow.md +++ b/docs/adr/0022-trust-policy-deny-over-allow.md @@ -118,7 +118,7 @@ implicitly denied. `User-Agent` header), `aws:ViaAWSService` and `aws:PrincipalIsAWSService` (both false for a direct caller), the tag keys (`aws:PrincipalTag/*`, `aws:RequestTag/*`, `aws:ResourceTag/*` on the identity path, - `dynamodb:ResourceTag/*`, each matched case-insensitively on the key), + each matched case-insensitively on the key), `sts:ExternalId` and `aws:TagKeys`, and, for a role-session caller only, `aws:TokenIssueTime` (the session issue time) and `aws:MultiFactorAuthPresent` (`false` for a session assumed without MFA), both absent for an IAM user. diff --git a/docs/design/01-requirements.md b/docs/design/01-requirements.md index e27724f63..394412bd2 100755 --- a/docs/design/01-requirements.md +++ b/docs/design/01-requirements.md @@ -307,7 +307,7 @@ extenddb implements a local IAM identity model that mirrors the structure of AWS ### 4.4 Authorization — Tag-Based Access Control (ABAC) -- REQ-ABAC-001: Support **resource tag conditions** in policies via `dynamodb:ResourceTag/{key}`. The policy engine resolves resource tags by querying the tag store for the target table's ARN. +- REQ-ABAC-001: Support **resource tag conditions** in policies via `aws:ResourceTag/{key}`. The policy engine resolves resource tags by querying the tag store for the target table's ARN. - REQ-ABAC-002: Support **principal tag conditions** via `aws:PrincipalTag/{key}`. Principal tags come from the authenticated identity (user tags, role tags, or session tags). - REQ-ABAC-003: Support the following condition operators for ABAC: - String operators: `StringEquals`, `StringNotEquals`, `StringLike`, `StringNotLike`, `StringEqualsIgnoreCase` diff --git a/docs/design/05-component-auth-v3.md b/docs/design/05-component-auth-v3.md index 341d2e0b8..7a920770b 100755 --- a/docs/design/05-component-auth-v3.md +++ b/docs/design/05-component-auth-v3.md @@ -361,7 +361,7 @@ Full set from v2.0 §6.5: StringEquals, StringLike, NumericEquals, DateEquals, B ### 8.3 DynamoDB Condition Keys - `aws:PrincipalTag/*` — from authenticated identity's tags -- `dynamodb:ResourceTag/*` — from target table's tags +- `aws:ResourceTag/*`, from target table's tags - `dynamodb:LeadingKeys` — partition key values being accessed - `dynamodb:Attributes` — attribute names being read/written - `dynamodb:Select`, `dynamodb:ReturnValues`, `dynamodb:ReturnConsumedCapacity` diff --git a/docs/design/05-component-auth.md b/docs/design/05-component-auth.md index fa5bfb10c..fd65f9bf4 100755 --- a/docs/design/05-component-auth.md +++ b/docs/design/05-component-auth.md @@ -144,7 +144,7 @@ pub enum AuthzDecision { pub struct RequestContext { /// aws:PrincipalTag/* — from the authenticated identity's tags. pub principal_tags: HashMap, - /// dynamodb:ResourceTag/* — from the target table's tags. + /// aws:ResourceTag/*, from the target table's tags. pub resource_tags: HashMap, /// dynamodb:LeadingKeys — partition key values being accessed. /// `None` for table-level operations (CreateTable, DeleteTable, etc.) @@ -600,7 +600,7 @@ impl RequestContext { fn resolve_key_inner(&self, key: &str) -> Option> { if let Some(tag_key) = key.strip_prefix("aws:PrincipalTag/") { self.principal_tags.get(tag_key).map(|v| vec![v.as_str()]) - } else if let Some(tag_key) = key.strip_prefix("dynamodb:ResourceTag/") { + } else if let Some(tag_key) = key.strip_prefix("aws:ResourceTag/") { self.resource_tags.get(tag_key).map(|v| vec![v.as_str()]) } else { match key { diff --git a/docs/manuals/02-design-guide.md b/docs/manuals/02-design-guide.md index 00d5541a0..a841b827d 100755 --- a/docs/manuals/02-design-guide.md +++ b/docs/manuals/02-design-guide.md @@ -120,7 +120,7 @@ After authentication, the authorization layer evaluates IAM policies using a 5-p 4. **Identity Allow** — scan identity policies (user, group, role). Any matching Allow → access granted. 5. **Implicit Deny** — no matching Allow → access denied. -Policy conditions support all IAM condition operators: `StringEquals`, `StringNotEquals`, `StringEqualsIgnoreCase`, `StringLike`, `StringNotLike`, `NumericEquals`, `NumericNotEquals`, `NumericLessThan`, `NumericLessThanEquals`, `NumericGreaterThan`, `NumericGreaterThanEquals`, `DateEquals`, `DateNotEquals`, `DateLessThan`, `DateLessThanEquals`, `DateGreaterThan`, `DateGreaterThanEquals`, `Bool`, `Null`, `ArnEquals`, `ArnNotEquals`, `ArnLike`, `ArnNotLike`, plus `ForAllValues`, `ForAnyValue`, and `IfExists` modifiers. Supported condition keys include `aws:PrincipalTag/*`, `dynamodb:ResourceTag/*`, `dynamodb:LeadingKeys`, `dynamodb:Attributes`, `dynamodb:Select`, `dynamodb:ReturnValues`, `dynamodb:ReturnConsumedCapacity`, `dynamodb:FullTableScan`, and `dynamodb:EnclosingOperation`. +Policy conditions support all IAM condition operators: `StringEquals`, `StringNotEquals`, `StringEqualsIgnoreCase`, `StringLike`, `StringNotLike`, `NumericEquals`, `NumericNotEquals`, `NumericLessThan`, `NumericLessThanEquals`, `NumericGreaterThan`, `NumericGreaterThanEquals`, `DateEquals`, `DateNotEquals`, `DateLessThan`, `DateLessThanEquals`, `DateGreaterThan`, `DateGreaterThanEquals`, `Bool`, `Null`, `ArnEquals`, `ArnNotEquals`, `ArnLike`, `ArnNotLike`, plus `ForAllValues`, `ForAnyValue`, and `IfExists` modifiers. Supported condition keys include `aws:PrincipalTag/*`, `aws:ResourceTag/*`, `dynamodb:LeadingKeys`, `dynamodb:Attributes`, `dynamodb:Select`, `dynamodb:ReturnValues`, `dynamodb:ReturnConsumedCapacity`, `dynamodb:FullTableScan`, and `dynamodb:EnclosingOperation`. ### Credential Storage @@ -198,7 +198,7 @@ Catalog state is never cached because correctness requires every request to see - **Table metadata** (key schema, attribute definitions, status, billing mode): A stale cache could serve the wrong key schema after a table is deleted and recreated with the same name but different schema. The new table has a different `table_id`, different key schema, and different indexes — stale cache serves wrong schema, writes corrupt data, reads return garbage. - **IAM policies and credentials**: A revoked Deny policy still cached as absent creates a security gap. A deleted access key still cached as valid allows unauthorized access. -- **Tags**: Tag-based authorization (`dynamodb:ResourceTag/*`) requires current tag values. +- **Tags**: Tag-based authorization (`aws:ResourceTag/*`) requires current tag values. - **GSI definitions**: Stale GSI metadata could route writes to wrong index tables. ### The Table-Name-Reuse Problem diff --git a/docs/manuals/05-admin-guide.md b/docs/manuals/05-admin-guide.md index f3191d21c..b76cf842c 100755 --- a/docs/manuals/05-admin-guide.md +++ b/docs/manuals/05-admin-guide.md @@ -438,8 +438,8 @@ caller omits the key. ExtendDB resolves `aws:PrincipalArn`, `aws:PrincipalAccoun `aws:ResourceAccount`, `aws:PrincipalType`, `aws:username` (for a user caller), `aws:userid`, `aws:SourceIp`, `aws:UserAgent`, `aws:SecureTransport`, `aws:RequestedRegion`, `aws:CurrentTime`, `aws:EpochTime`, `aws:ViaAWSService`, -`aws:PrincipalIsAWSService`, `aws:PrincipalTag/`, `aws:ResourceTag/` -and `dynamodb:ResourceTag/` (tag keys matched case-insensitively) on a +`aws:PrincipalIsAWSService`, `aws:PrincipalTag/` and `aws:ResourceTag/` +(tag keys matched case-insensitively) on a data-plane request. Every other `aws:` global key is absent, which is what the service does for a key it does not populate for this caller and for an unknown key. @@ -468,7 +468,7 @@ Resolved (evaluated against the request value): - `aws:ViaAWSService` and `aws:PrincipalIsAWSService` (both false for a direct caller). - `aws:PrincipalTag/`, `aws:RequestTag/` (trust path), - `aws:ResourceTag/`, `dynamodb:ResourceTag/`, `aws:TagKeys`, + `aws:ResourceTag/`, `aws:TagKeys`, `sts:ExternalId` (trust path). - Role-session only (absent for an IAM user, matching the service): `aws:TokenIssueTime` (the session issue time) and diff --git a/docs/manuals/07-upgrade-manual.md b/docs/manuals/07-upgrade-manual.md index bd98e8b7a..15af59c32 100755 --- a/docs/manuals/07-upgrade-manual.md +++ b/docs/manuals/07-upgrade-manual.md @@ -346,6 +346,37 @@ for any of these changes. --- +### v0.1.14: IAM resource tags evaluate only as `aws:ResourceTag/` + +**What changed.** ExtendDB now resolves resource tags only as `aws:ResourceTag/`, matching the live service. `dynamodb:ResourceTag/` is left absent (the live service never populates it). Before v0.1.14, ExtendDB filled `dynamodb:ResourceTag/` from the target table's tags, so a condition on that key evaluated against the tag value instead of being treated as absent. `aws:ResourceTag/` is unaffected and keeps resolving from the table's tags, case-insensitively on the key. + +**Who is affected.** Any deployment with an IAM policy that uses a `dynamodb:ResourceTag/` condition key or a `${dynamodb:ResourceTag/...}` policy variable: inline user, group and role policies, user and role permissions boundaries, and session policies. A policy that uses `aws:ResourceTag/` is not affected. Role trust policies are not affected: the AssumeRole path never resolved `dynamodb:ResourceTag/`, so a trust policy that names it behaved then as it does now. + +**Migration is required.** `extenddb serve` refuses to start on a catalog that has not been migrated, with an error that names `extenddb migrate --yes`. Run `extenddb migrate --yes` once per deployment before starting the new binary. On PostgreSQL, `migrate` connects as the admin user, so pass `--pg-user ` and set `EXTENDDB_PG_PASSWORD`, as for `init`. The migration rewrites every stored `dynamodb:ResourceTag/` condition key and `${dynamodb:ResourceTag/...}` policy variable to the `aws:ResourceTag/` form, keeping the operator, the tag key and the value unchanged. The rewrite preserves each policy's decision, because `aws:ResourceTag/` resolves from the same table tags the old `dynamodb:ResourceTag/` key read. The one difference is a `${dynamodb:ResourceTag/}` policy variable on a table without that tag: neither form expands, and the literal text in the comparison changes from `${dynamodb:ResourceTag/}` to `${aws:ResourceTag/}`. The decision changes only for a request value equal to one of those literals. A freshly initialised deployment needs no migrate step; its catalog is already recorded as migrated. + +**What `migrate` prints.** Without `--yes`, `migrate` lists each policy it would rewrite, one line per policy with the account, the principal, the policy name and the number of condition keys, and writes nothing. With `--yes`, it rewrites and records that the migration is complete. It runs on every `migrate`, so an interrupted run is finished by the next one. If a policy lookup fails, or a policy needs attention (below), `migrate` reports it, exits non-zero, and does not record completion, so a later run retries. Policy bodies are never printed. + +**Both forms on one key.** If one operator block in a policy holds both `aws:ResourceTag/` and `dynamodb:ResourceTag/` for the same tag key with the same value, the two were one condition before the upgrade, and `migrate` drops the `dynamodb:` key. With different values they cannot share one JSON key. `migrate` then leaves the policy unchanged, lists it as `NEEDS ATTENTION` with the account, the principal and the policy name, exits non-zero, and does not record the migration as complete, so `serve` keeps refusing the catalog. Fix the policy with the previous version still serving the catalog (keep one of the two forms), then run `migrate --yes` with the new binary. A session policy that does not parse is reported in the log and skipped; it cannot be evaluated under either form. + +**What each statement shape does for a policy that was not migrated.** An operator who writes a policy by hand after the upgrade, for example from an old template, needs to know that the `dynamodb:ResourceTag/` form has no effect. With the key absent, standard IAM absent-key semantics apply: + +| Statement shape on `dynamodb:ResourceTag/env` | Outcome for an unmigrated policy | +|---|---| +| `StringEquals` (bare positive) | Never matches; an `Allow` stops granting, a `Deny` stops blocking | +| `StringNotEquals` (bare negated) | Always matches on the absent single-valued key; an `Allow` grants, a `Deny` blocks | +| `StringEqualsIfExists` | Passes (the key is absent), so the statement applies on the key alone | +| `ForAllValues:StringEquals` | Vacuously true on the absent key, so the statement applies on the key alone | +| `ForAnyValue:StringEquals` | False on the absent key, so an `Allow` does not grant | +| `Null` with value `true` | Matches (the key is absent) | +| `Null` with value `false` | Does not match (the key is absent) | +| `${dynamodb:ResourceTag/env}` policy variable in a value | Does not expand; the literal text `${dynamodb:ResourceTag/env}` is compared, which no request value equals | + +To get the intended behaviour, write the condition on `aws:ResourceTag/`. + +**Sessions issued before the migration.** Session policies are stored in the catalog, so `migrate` rewrites them along with the identity policies. A session issued before the upgrade is evaluated with its rewritten policy once `migrate --yes` has run, and it does not need to be issued again. The new binary does not serve an unmigrated catalog, so no request is evaluated against the old policy text under the new rules. + +--- + ## License Copyright 2026 ExtendDB contributors. Licensed under the Apache License, Version 2.0. diff --git a/docs/manuals/10-security-model.md b/docs/manuals/10-security-model.md index 7eb64fe2d..119459fed 100755 --- a/docs/manuals/10-security-model.md +++ b/docs/manuals/10-security-model.md @@ -108,7 +108,7 @@ Set operators and `IfExists` can be combined: `ForAllValues:StringEqualsIfExists | Key | Type | Description | |-----|------|-------------| | `aws:PrincipalTag/` | String | Tag on the authenticated principal (user or role) | -| `dynamodb:ResourceTag/` | String | Tag on the target DynamoDB table | +| `aws:ResourceTag/` | String | Tag on the target DynamoDB table | | `dynamodb:LeadingKeys` | Multi-valued string | Partition key values being accessed | | `dynamodb:Attributes` | Multi-valued string | Attribute names being read or written | | `dynamodb:Select` | String | The `Select` parameter value | @@ -127,7 +127,7 @@ extenddb supports the same access control patterns as real AWS IAM: **Role-Based Access Control (RBAC)**: Create IAM groups with policies, add users to groups. Users inherit group policies. Create IAM roles for cross-account or service-to-service access. -**Attribute-Based Access Control (ABAC)**: Use `aws:PrincipalTag/*` and `dynamodb:ResourceTag/*` condition keys to make access decisions based on tags. Example: allow users tagged `Department=Engineering` to access tables tagged `Department=Engineering`. +**Attribute-Based Access Control (ABAC)**: Use `aws:PrincipalTag/*` and `aws:ResourceTag/*` condition keys to make access decisions based on tags. Example: allow users tagged `Department=Engineering` to access tables tagged `Department=Engineering`. **Fine-Grained Access Control (FGAC)**: Use `dynamodb:LeadingKeys` to restrict access to specific partition key values. Use `dynamodb:Attributes` to restrict which attributes can be read or written. Example: allow a user to access only items where the partition key matches their user ID. diff --git a/tests/test_abac.py b/tests/test_abac.py index 62f4d98ef..140ae9169 100755 --- a/tests/test_abac.py +++ b/tests/test_abac.py @@ -91,7 +91,7 @@ def _create_user_with_key(mgmt: ManagementClient, account_id: str, # --------------------------------------------------------------------------- class TestResourceTagCondition: - """Policies with dynamodb:ResourceTag/* conditions.""" + """Policies with aws:ResourceTag/* conditions.""" @pytest.fixture(autouse=True) def setup(self, auth_env, mgmt, account_id, region): @@ -130,7 +130,7 @@ def test_resource_tag_allows_matching_table(self): "Resource": "*", "Condition": { "StringEquals": { - "dynamodb:ResourceTag/Env": "dev", + "aws:ResourceTag/Env": "dev", }, }, }], @@ -206,7 +206,7 @@ def test_resource_tag_denies_non_matching_table(self): "Resource": "*", "Condition": { "StringEquals": { - "dynamodb:ResourceTag/Env": "dev", + "aws:ResourceTag/Env": "dev", }, }, }], @@ -692,6 +692,48 @@ def test_unknown_key_deny_negated_blocks(self, table): {"StringNotEquals": {"aws:NoSuchGlobalKey": "x"}})) == "DENY" +class TestResourceTagConditionKeyForms(_TaggedTableBase): + """Resource tags evaluate only as aws:ResourceTag/*, matching live DynamoDB. + aws:ResourceTag/env resolves to the table's tag value; dynamodb:ResourceTag/env + is absent and keeps IAM absent-key semantics. Table is tagged env=prod. Every + expected outcome was measured live on 2026-10-21.""" + + def test_aws_resourcetag_eq_grants(self, table): + # aws:ResourceTag/env resolves to prod, so StringEquals matches. Live: ALLOW. + assert self._get(table, self._allow_get_if( + {"StringEquals": {"aws:ResourceTag/env": "prod"}})) == "ALLOW" + + def test_aws_resourcetag_ne_denies(self, table): + # aws:ResourceTag/env resolves to prod, so StringNotEquals is false and the + # Allow does not grant. Live: DENY. + assert self._get(table, self._allow_get_if( + {"StringNotEquals": {"aws:ResourceTag/env": "prod"}})) == "DENY" + + def test_aws_resourcetag_deny_eq_blocks(self, table): + # A Deny gated on StringEquals aws:ResourceTag/env=prod fires on the prod + # table. Live: DENY. + assert self._get(table, self._allow_all_deny_get_if( + {"StringEquals": {"aws:ResourceTag/env": "prod"}})) == "DENY" + + def test_dynamodb_resourcetag_eq_denies_absent_key(self, table): + # dynamodb:ResourceTag/env is absent, so StringEquals never matches and the + # Allow does not grant. Live: DENY. + assert self._get(table, self._allow_get_if( + {"StringEquals": {"dynamodb:ResourceTag/env": "prod"}})) == "DENY" + + def test_dynamodb_resourcetag_ne_grants_absent_key(self, table): + # dynamodb:ResourceTag/env is absent, so StringNotEquals on an absent + # single-valued key matches and the Allow grants. Live: ALLOW. + assert self._get(table, self._allow_get_if( + {"StringNotEquals": {"dynamodb:ResourceTag/env": "prod"}})) == "ALLOW" + + def test_dynamodb_resourcetag_deny_eq_does_not_block_absent_key(self, table): + # A Deny gated on StringEquals dynamodb:ResourceTag/env=prod does not fire, + # because the key is absent, so the base Allow stands. Live: ALLOW. + assert self._get(table, self._allow_all_deny_get_if( + {"StringEquals": {"dynamodb:ResourceTag/env": "prod"}})) == "ALLOW" + + class TestRound3GlobalKeys(_TaggedTableBase): """Round three keys on the identity path: aws:userid, aws:SourceIp, aws:UserAgent, aws:ResourceAccount, and the organization keys absent for a @@ -786,9 +828,11 @@ def test_resourcetag_lowercase_prefix_grants(self, table): assert self._get(table, self._allow_get_if( {"StringEquals": {"aws:resourcetag/env": "prod"}})) == "ALLOW" - def test_dynamodb_resourcetag_lowercase_prefix_grants(self, table): + def test_dynamodb_resourcetag_lowercase_prefix_is_absent(self, table): + # dynamodb:ResourceTag/* is absent in any case, as on DynamoDB, so a + # positive operator on it never matches and the Allow does not grant. assert self._get(table, self._allow_get_if( - {"StringEquals": {"dynamodb:resourcetag/env": "prod"}})) == "ALLOW" + {"StringEquals": {"dynamodb:resourcetag/env": "prod"}})) == "DENY" def test_principaltype_deny_lowercase_key_blocks(self, table): # A Deny gated on a miscased key must still be enforced. aws:principaltype diff --git a/tests/test_resource_tag_migration.py b/tests/test_resource_tag_migration.py new file mode 100644 index 000000000..031eae8a8 --- /dev/null +++ b/tests/test_resource_tag_migration.py @@ -0,0 +1,574 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""End-to-end wire test for the resource-tag condition-key migration (SQLite). + +Drives a real SQLite deployment through its lifecycle on its own data +directory and port, independent of the server the integration job already +started: + + 1. init (a fresh catalog carries the migration marker) and serve; the first + start of a fresh catalog logs no migration line. + 2. store an inline user policy, a group policy, a permissions boundary and a + role trust policy using every operator shape of dynamodb:ResourceTag, + plus a conditional allow used for the decision check, through the + management API. + 3. stop, run `migrate` without --yes: it lists each rewrite with a count and + exits non-zero, and nothing on disk changes (read back with the sqlite3 + module). + 4. run `migrate --yes`, restart, and confirm a GetItem gated on the resource + tag that was denied before the migration is allowed after it. + 5. delete the marker row with the sqlite3 module and confirm serve refuses to + start, naming `extenddb migrate --yes`. + +The binary is found the way tests/lifecycle_helpers.py does, falling back to +target/release/extenddb, so the SQLite integration job runs it against the +release binary it builds. On a non-SQLite binary (the PostgreSQL job) the test +skips, because `init --backend sqlite` is rejected there. + +This file is NOT in devtools/run-tests' ignore list, so the main pytest run of +both integration jobs collects it; only the SQLite job's binary makes it run. +""" + +from __future__ import annotations + +import json +import os +import re +import signal +import socket +import sqlite3 +import subprocess +import time +import uuid + +import boto3 +import pytest +import requests +from botocore.config import Config as BotoConfig +from botocore.exceptions import ClientError + +from lifecycle_helpers import EXTENDDB_BINARY + +MIGRATION_MARKER_KEY = "resource_tag_migration" + +# This test stands up its own deployment, so it must not inherit admin +# credentials or an endpoint from an outer harness (devtools/run-tests and the +# SQLite job export EXTENDDB_ADMIN_USER / EXTENDDB_ADMIN_PASSWORD / +# EXTENDDB_TEST_ENDPOINT for the already-running server). With those set, `init` +# would adopt the ambient admin credential and print no generated password, and +# the test could not authenticate to its own server. Scrub them for every +# subprocess this test launches. +_AMBIENT_KEYS = ( + "EXTENDDB_ADMIN_USER", + "EXTENDDB_ADMIN_PASSWORD", + "EXTENDDB_TEST_ENDPOINT", + "REQUESTS_CA_BUNDLE", + "SSL_CERT_FILE", + "CURL_CA_BUNDLE", + "AWS_CA_BUNDLE", +) + + +def _clean_env() -> dict: + env = os.environ.copy() + for k in _AMBIENT_KEYS: + env.pop(k, None) + return env + + +def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _run(*args, timeout=60, env=None): + return subprocess.run( + [EXTENDDB_BINARY, *args], + capture_output=True, + text=True, + timeout=timeout, + stdin=subprocess.DEVNULL, + env=env if env is not None else _clean_env(), + ) + + +def _binary_is_sqlite(config_path: str, sqlite_path: str) -> bool: + """A SQLite init succeeds only on a SQLite-backend binary; the PostgreSQL + binary rejects --backend sqlite. Use that to skip on the wrong binary.""" + proc = _run( + "init", + "--backend", + "sqlite", + "--sqlite-path", + sqlite_path, + "--config", + config_path, + timeout=60, + ) + if proc.returncode == 0: + return True + combined = proc.stdout + proc.stderr + if "backend" in combined.lower(): + return False + raise AssertionError(f"unexpected init failure: {combined}") + + +def _admin_password(init_output: str) -> str: + m = re.search(r"Password:\s*(\S+)", init_output) + assert m, f"no admin password in init output: {init_output}" + return m.group(1) + + +def _patch_port(config_path: str, port: int) -> None: + with open(config_path) as f: + content = f.read() + new_content, n = re.subn( + r"(?m)^[ \t]*#?[ \t]*port[ \t]*=[ \t]*\d+.*$", + f"port = {port}", + content, + count=1, + ) + if n == 0: + new_content = content.replace( + 'bind_addr = "127.0.0.1"', + f'bind_addr = "127.0.0.1"\nport = {port}', + 1, + ) + with open(config_path, "w") as f: + f.write(new_content) + + +class _Server: + """A foreground extenddb server subprocess with its own log file.""" + + def __init__(self, config_path: str, port: int, log_path: str): + self.config_path = config_path + self.port = port + self.log_path = log_path + self.proc: subprocess.Popen | None = None + + def start(self, wait_healthy=True, timeout=60) -> None: + log = open(self.log_path, "w") + self.proc = subprocess.Popen( + [EXTENDDB_BINARY, "serve", "--config", self.config_path, "--foreground"], + stdout=log, + stderr=subprocess.STDOUT, + stdin=subprocess.DEVNULL, + env=_clean_env(), + ) + if not wait_healthy: + return + url = f"https://127.0.0.1:{self.port}/health" + deadline = time.time() + timeout + while time.time() < deadline: + if self.proc.poll() is not None: + raise AssertionError(f"server exited early: {self.read_log()}") + try: + r = requests.get(url, verify=False, timeout=2) + if r.ok and "healthy" in r.text: + return + except requests.RequestException: + pass + time.sleep(0.5) + raise AssertionError(f"server did not become healthy: {self.read_log()}") + + def stop(self) -> None: + if self.proc is None: + return + self.proc.send_signal(signal.SIGTERM) + try: + self.proc.wait(timeout=20) + except subprocess.TimeoutExpired: + self.proc.kill() + self.proc.wait(timeout=10) + self.proc = None + + def read_log(self) -> str: + try: + with open(self.log_path) as f: + return f.read() + except FileNotFoundError: + return "" + + +def _client(endpoint: str, ak: str, sk: str, region: str): + return boto3.client( + "dynamodb", + endpoint_url=endpoint, + aws_access_key_id=ak, + aws_secret_access_key=sk, + region_name=region, + config=BotoConfig(retries={"max_attempts": 0}), + verify=False, + ) + + +def _policy(resource_tag_key: str) -> dict: + """An allow gated on = prod.""" + return { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Action": ["dynamodb:GetItem", "dynamodb:DescribeTable"], + "Resource": "*", + "Condition": {"StringEquals": {resource_tag_key: "prod"}}, + } + ], + } + + +def _every_shape_policy() -> dict: + """One statement per operator shape, all on dynamodb:ResourceTag/*.""" + return { + "Version": "2012-10-17", + "Statement": [ + {"Effect": "Allow", "Action": "dynamodb:GetItem", "Resource": "*", + "Condition": {"StringEquals": {"dynamodb:ResourceTag/env": "prod"}}}, + {"Effect": "Deny", "Action": "dynamodb:GetItem", "Resource": "*", + "Condition": {"StringNotEquals": {"dynamodb:ResourceTag/env": "prod"}}}, + {"Effect": "Allow", "Action": "dynamodb:PutItem", "Resource": "*", + "Condition": {"StringEqualsIfExists": {"dynamodb:ResourceTag/env": "prod"}}}, + {"Effect": "Allow", "Action": "dynamodb:Query", "Resource": "*", + "Condition": {"ForAllValues:StringEquals": {"dynamodb:ResourceTag/env": ["prod"]}}}, + {"Effect": "Allow", "Action": "dynamodb:Scan", "Resource": "*", + "Condition": {"ForAnyValue:StringEquals": {"dynamodb:ResourceTag/env": ["prod"]}}}, + {"Effect": "Allow", "Action": "dynamodb:BatchGetItem", "Resource": "*", + "Condition": {"Null": {"dynamodb:ResourceTag/env": "false"}}}, + {"Effect": "Allow", "Action": "dynamodb:UpdateItem", "Resource": "*", + "Condition": {"StringEquals": {"aws:PrincipalTag/team": "${dynamodb:ResourceTag/team}"}}}, + ], + } + + +@pytest.fixture() +def deployment(tmp_path): + requests.packages.urllib3.disable_warnings() # self-signed cert + if not os.path.isfile(EXTENDDB_BINARY): + pytest.skip(f"extenddb binary not built at {EXTENDDB_BINARY}") + + data_dir = tmp_path / "data" + data_dir.mkdir() + sqlite_path = str(data_dir / "data.sqlite") + config_path = str(tmp_path / "extenddb.toml") + + init = _run( + "init", "--backend", "sqlite", + "--sqlite-path", sqlite_path, "--config", config_path, + ) + if init.returncode != 0: + combined = init.stdout + init.stderr + if "backend" in combined.lower(): + pytest.skip("binary is not a SQLite build") + raise AssertionError(f"sqlite init failed: {combined}") + + admin_pw = _admin_password(init.stdout + init.stderr) + port = _free_port() + _patch_port(config_path, port) + endpoint = f"https://127.0.0.1:{port}" + log_path = str(tmp_path / "server.log") + server = _Server(config_path, port, log_path) + + yield { + "config": config_path, + "sqlite_path": sqlite_path, + "port": port, + "endpoint": endpoint, + "admin_user": "admin", + "admin_pw": admin_pw, + "server": server, + "init_output": init.stdout + init.stderr, + } + server.stop() + + +def _sqlite_policy_docs(sqlite_path: str) -> list[str]: + conn = sqlite3.connect(sqlite_path) + try: + rows = conn.execute("SELECT policy_document FROM iam_policies").fetchall() + finally: + conn.close() + return [r[0] for r in rows] + + +def test_resource_tag_migration_wire(deployment): + from management_helpers import ManagementClient + + endpoint = deployment["endpoint"] + server = deployment["server"] + sqlite_path = deployment["sqlite_path"] + region = "us-east-1" + + # 1. First start of a fresh catalog: marker is seeded, so serve starts and + # logs nothing about the migration. + server.start() + log = server.read_log() + assert "resource-tag condition-key migration" not in log, log + assert "dynamodb:ResourceTag" not in log, log + + mgmt = ManagementClient(endpoint, deployment["admin_user"], deployment["admin_pw"]) + account_id = f"{uuid.uuid4().int % 10**12:012d}" + assert mgmt.create_account(account_id, "wire").status_code == 201 + + # Admin user with full access to create and tag a table. + assert mgmt.create_user(account_id, "fulladmin", "TestPass123!").status_code == 201 + creds = mgmt.create_access_key(account_id, "fulladmin").json() + admin_ak, admin_sk = creds["access_key_id"], creds["secret_access_key"] + assert mgmt.put_user_policy( + account_id, "fulladmin", "full", + {"Version": "2012-10-17", "Statement": [ + {"Effect": "Allow", "Action": "dynamodb:*", "Resource": "*"}]}, + ).status_code == 204 + + # Restricted user: GetItem allowed only when the resource tag env=prod, + # written on the dynamodb:ResourceTag form (the pre-migration shape). + assert mgmt.create_user(account_id, "scoped", "TestPass123!").status_code == 201 + scoped = mgmt.create_access_key(account_id, "scoped").json() + scoped_ak, scoped_sk = scoped["access_key_id"], scoped["secret_access_key"] + assert mgmt.put_user_policy( + account_id, "scoped", "tagpolicy", _policy("dynamodb:ResourceTag/env"), + ).status_code == 204 + + # Every operator shape, on a group policy, so the migration has to rewrite + # them all. + assert mgmt.create_group(account_id, "grp").status_code == 201 + assert mgmt.put_group_policy( + account_id, "grp", "shapes", _every_shape_policy() + ).status_code == 204 + + # A permissions boundary carrying the key too. + assert mgmt.set_user_boundary( + account_id, "scoped", _policy("dynamodb:ResourceTag/env") + ).status_code in (200, 204) + + # Create and tag the table (env=prod) and seed an item. + admin_client = _client(endpoint, admin_ak, admin_sk, region) + table = f"wire-{uuid.uuid4().hex[:8]}" + admin_client.create_table( + TableName=table, + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], + BillingMode="PAY_PER_REQUEST", + Tags=[{"Key": "env", "Value": "prod"}], + ) + from conftest import wait_for_active + wait_for_active(admin_client, table) + admin_client.put_item(TableName=table, Item={"pk": {"S": "k1"}, "v": {"S": "hi"}}) + + # Before the migration the scoped policy gates on dynamodb:ResourceTag, + # which this build leaves absent, so the positive condition never matches + # and GetItem is denied. + scoped_client = _client(endpoint, scoped_ak, scoped_sk, region) + with pytest.raises(ClientError) as exc: + scoped_client.get_item(TableName=table, Key={"pk": {"S": "k1"}}) + assert exc.value.response["Error"]["Code"] == "AccessDeniedException" + + # 2. Stop and run migrate WITHOUT --yes: it lists the rewrites with counts + # and exits non-zero, and nothing on disk changes. + server.stop() + before_docs = _sqlite_policy_docs(sqlite_path) + assert any("dynamodb:ResourceTag" in d for d in before_docs) + + dry = _run("migrate", "--config", deployment["config"]) + assert dry.returncode != 0, dry.stdout + dry.stderr + out = dry.stdout + dry.stderr + assert "Rewrite:" in out, out + assert "--yes is required" in out, out + # The listing names policies and key counts, never a policy body. + assert "key(s))" in out, out + # Nothing changed on disk. + assert _sqlite_policy_docs(sqlite_path) == before_docs + + # 3. migrate --yes rewrites and records the marker. + applied = _run("migrate", "--yes", "--config", deployment["config"]) + assert applied.returncode == 0, applied.stdout + applied.stderr + after_docs = _sqlite_policy_docs(sqlite_path) + assert all("dynamodb:ResourceTag" not in d for d in after_docs), after_docs + assert any("aws:ResourceTag" in d for d in after_docs) + + # 4. Restart and confirm the decision flipped: the rewritten aws:ResourceTag + # policy now resolves the tag, so GetItem on the env=prod table is allowed. + server.start() + scoped_client = _client(endpoint, scoped_ak, scoped_sk, region) + got = scoped_client.get_item(TableName=table, Key={"pk": {"S": "k1"}}) + assert got["Item"]["v"]["S"] == "hi" + + # A second migrate is idempotent: nothing left to rewrite, marker present. + server.stop() + again = _run("migrate", "--config", deployment["config"]) + assert again.returncode == 0, again.stdout + again.stderr + assert "No dynamodb:ResourceTag condition keys found." in (again.stdout + again.stderr) + + # 5. Delete the marker row with the sqlite3 module and confirm serve refuses. + conn = sqlite3.connect(sqlite_path) + try: + conn.execute("DELETE FROM settings WHERE key = ?", (MIGRATION_MARKER_KEY,)) + conn.commit() + finally: + conn.close() + refused = _run("serve", "--config", deployment["config"], "--foreground", timeout=60) + assert refused.returncode != 0 + assert "extenddb migrate --yes" in (refused.stdout + refused.stderr), ( + refused.stdout + refused.stderr + ) + + +def _sqlite_marker(sqlite_path: str) -> list: + conn = sqlite3.connect(sqlite_path) + try: + return conn.execute( + "SELECT value FROM settings WHERE key = ?", (MIGRATION_MARKER_KEY,) + ).fetchall() + finally: + conn.close() + + +def _sqlite_policy(sqlite_path: str, policy_name: str) -> dict: + conn = sqlite3.connect(sqlite_path) + try: + row = conn.execute( + "SELECT policy_document FROM iam_policies WHERE policy_name = ?", + (policy_name,), + ).fetchone() + finally: + conn.close() + return json.loads(row[0]) + + +def _sqlite_set_policy(sqlite_path: str, policy_name: str, document: dict) -> None: + conn = sqlite3.connect(sqlite_path) + try: + conn.execute( + "UPDATE iam_policies SET policy_document = ? WHERE policy_name = ?", + (json.dumps(document), policy_name), + ) + conn.commit() + finally: + conn.close() + + +def test_resource_tag_migration_collisions(deployment): + """Both resource-tag forms on one tag key in one operator block. + + Same values: both forms resolved from the same table tags before the + upgrade, so the two conditions are one condition. migrate drops the + dynamodb: key and the Deny keeps applying. + + Different values: the two cannot share one JSON key. migrate leaves the + policy unchanged, exits non-zero and does not record the marker, so serve + keeps refusing the catalog until the policy is fixed and migrate is run + again. Recording the marker here would change the decision: under + StringNotEquals the dynamodb: half, now absent, always matches. + """ + from management_helpers import ManagementClient + + endpoint = deployment["endpoint"] + server = deployment["server"] + sqlite_path = deployment["sqlite_path"] + region = "us-east-1" + + server.start() + mgmt = ManagementClient(endpoint, deployment["admin_user"], deployment["admin_pw"]) + account_id = f"{uuid.uuid4().int % 10**12:012d}" + assert mgmt.create_account(account_id, "collide").status_code == 201 + + assert mgmt.create_user(account_id, "fulladmin", "TestPass123!").status_code == 201 + creds = mgmt.create_access_key(account_id, "fulladmin").json() + assert mgmt.put_user_policy( + account_id, "fulladmin", "full", + {"Version": "2012-10-17", "Statement": [ + {"Effect": "Allow", "Action": "dynamodb:*", "Resource": "*"}]}, + ).status_code == 204 + + # Same-value duplicate on a Deny. + assert mgmt.create_user(account_id, "hedged", "TestPass123!").status_code == 201 + hedged = mgmt.create_access_key(account_id, "hedged").json() + assert mgmt.put_user_policy(account_id, "hedged", "hedged-same", { + "Version": "2012-10-17", + "Statement": [ + {"Effect": "Allow", "Action": "dynamodb:GetItem", "Resource": "*"}, + {"Effect": "Deny", "Action": "dynamodb:GetItem", "Resource": "*", + "Condition": {"StringEquals": { + "aws:ResourceTag/env": "prod", + "dynamodb:ResourceTag/env": "prod", + }}}, + ], + }).status_code == 204 + + # Different values, which cannot be merged. + assert mgmt.create_user(account_id, "split", "TestPass123!").status_code == 201 + split_doc = { + "Version": "2012-10-17", + "Statement": [ + {"Effect": "Allow", "Action": "dynamodb:GetItem", "Resource": "*", + "Condition": {"StringNotEquals": { + "aws:ResourceTag/env": "dev", + "dynamodb:ResourceTag/env": "test", + }}}, + ], + } + assert mgmt.put_user_policy(account_id, "split", "hedged-split", split_doc).status_code == 204 + + admin_client = _client(endpoint, creds["access_key_id"], creds["secret_access_key"], region) + table = f"wire-{uuid.uuid4().hex[:8]}" + admin_client.create_table( + TableName=table, + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], + BillingMode="PAY_PER_REQUEST", + Tags=[{"Key": "env", "Value": "prod"}], + ) + from conftest import wait_for_active + wait_for_active(admin_client, table) + admin_client.put_item(TableName=table, Item={"pk": {"S": "k1"}}) + server.stop() + + # Simulate a catalog created before the migration existed. + conn = sqlite3.connect(sqlite_path) + try: + conn.execute("DELETE FROM settings WHERE key = ?", (MIGRATION_MARKER_KEY,)) + conn.commit() + finally: + conn.close() + + applied = _run("migrate", "--yes", "--config", deployment["config"]) + out = applied.stdout + applied.stderr + assert applied.returncode != 0, out + assert "NEEDS ATTENTION" in out, out + attention = [line for line in out.splitlines() if "NEEDS ATTENTION" in line] + assert any("hedged-split" in line for line in attention), out + assert not any("hedged-same" in line for line in attention), out + assert _sqlite_marker(sqlite_path) == [] + # The different-value policy is unchanged; the same-value one keeps only + # the aws: key. + assert _sqlite_policy(sqlite_path, "hedged-split") == split_doc + same = _sqlite_policy(sqlite_path, "hedged-same") + assert same["Statement"][1]["Condition"] == { + "StringEquals": {"aws:ResourceTag/env": "prod"} + } + + refused = _run("serve", "--config", deployment["config"], "--foreground", timeout=60) + assert refused.returncode != 0 + assert "extenddb migrate --yes" in (refused.stdout + refused.stderr), ( + refused.stdout + refused.stderr + ) + + # Fix the different-value policy by hand (keep the aws: form) and run + # migrate again: it completes and records the marker. + fixed = json.loads(json.dumps(split_doc)) + del fixed["Statement"][0]["Condition"]["StringNotEquals"]["dynamodb:ResourceTag/env"] + _sqlite_set_policy(sqlite_path, "hedged-split", fixed) + again = _run("migrate", "--yes", "--config", deployment["config"]) + assert again.returncode == 0, again.stdout + again.stderr + assert _sqlite_marker(sqlite_path) != [] + + # The same-value Deny still applies after the upgrade. + server.start() + hedged_client = _client( + endpoint, hedged["access_key_id"], hedged["secret_access_key"], region + ) + with pytest.raises(ClientError) as exc: + hedged_client.get_item(TableName=table, Key={"pk": {"S": "k1"}}) + assert exc.value.response["Error"]["Code"] == "AccessDeniedException"