Fix custom role authorization

migrations/mysql/2026-06-30-120000_add_custom_role_permissions/down.sql

@@ -43,6 +43,38 @@ CREATE TABLE IF NOT EXISTS __vw_rollback_manager_allowlist (
     users_organizations_uuid CHAR(36) NOT NULL PRIMARY KEY
 );
 
+-- `IF NOT EXISTS` accepts an already existing table without checking its definition. Validate that
+-- definition before using it for the role mapping: MySQL/MariaDB compare a character UUID with a
+-- numeric allowlist as numbers, so an INT value such as 0 could match unrelated UUIDs. The duplicate
+-- key aborts the revert before any role or permission value is changed.
+CREATE TEMPORARY TABLE __vw_rollback_manager_allowlist_guard (
+    blocked INTEGER NOT NULL PRIMARY KEY
+);
+INSERT INTO __vw_rollback_manager_allowlist_guard (blocked) VALUES (1);
+INSERT INTO __vw_rollback_manager_allowlist_guard (blocked)
+SELECT 1 FROM DUAL
+WHERE (SELECT COUNT(*)
+       FROM information_schema.columns
+       WHERE table_schema = DATABASE()
+         AND table_name = '__vw_rollback_manager_allowlist') <> 1
+   OR (SELECT COUNT(*)
+       FROM information_schema.columns
+       WHERE table_schema = DATABASE()
+         AND table_name = '__vw_rollback_manager_allowlist'
+         AND column_name = 'users_organizations_uuid'
+         AND data_type = 'char'
+         AND character_maximum_length = 36
+         AND is_nullable = 'NO') <> 1
+   OR NOT EXISTS (
+       SELECT 1
+       FROM information_schema.statistics
+       WHERE table_schema = DATABASE()
+         AND table_name = '__vw_rollback_manager_allowlist'
+         AND column_name = 'users_organizations_uuid'
+         AND non_unique = 0
+   );
+DROP TEMPORARY TABLE __vw_rollback_manager_allowlist_guard;
+
 UPDATE users_organizations SET atype = 3
 WHERE atype = 4
   AND uuid IN (SELECT users_organizations_uuid FROM __vw_rollback_manager_allowlist);

src/api/core/ciphers.rs

@@ -393,13 +393,10 @@ async fn enforce_personal_ownership_policy(data: Option<&CipherData>, headers: &
 }
 
 fn has_prevalidated_organization_write_authority(
-    allow_direct_organization_write: bool,
     shared_to_collections: Option<&Vec<CollectionId>>,
     member_has_full_access: bool,
 ) -> bool {
-    allow_direct_organization_write
-        || shared_to_collections.is_some_and(|collections| !collections.is_empty())
-        || member_has_full_access
+    shared_to_collections.is_some_and(|collections| !collections.is_empty()) || member_has_full_access
 }
 
 pub async fn update_cipher_from_data(
@@ -410,23 +407,6 @@ pub async fn update_cipher_from_data(
     conn: &DbConn,
     nt: &Notify<'_>,
     ut: UpdateType,
-) -> EmptyResult {
-    update_cipher_from_data_with_authority(cipher, data, headers, shared_to_collections, false, conn, nt, ut).await
-}
-
-#[expect(
-    clippy::too_many_arguments,
-    reason = "The extra flag is a prevalidated route authority and must remain separate from client data"
-)]
-pub(super) async fn update_cipher_from_data_with_authority(
-    cipher: &mut Cipher,
-    data: CipherData,
-    headers: &Headers,
-    shared_to_collections: Option<Vec<CollectionId>>,
-    allow_direct_organization_write: bool,
-    conn: &DbConn,
-    nt: &Notify<'_>,
-    ut: UpdateType,
 ) -> EmptyResult {
     // Cleanup cipher data, like removing the 'Response' key.
     // This key is somewhere generated during Javascript so no way for us this fix this.
@@ -480,7 +460,6 @@ pub(super) async fn update_cipher_from_data_with_authority(
                 // A non-empty list of collections implies the caller already validated the user's write
                 // access to them, so we can move the cipher into the organization on that basis.
                 if has_prevalidated_organization_write_authority(
-                    allow_direct_organization_write,
                     shared_to_collections.as_ref(),
                     member.has_full_access(),
                 ) || cipher.is_write_accessible_to_user(&headers.user.uuid, conn).await
@@ -611,17 +590,14 @@ mod update_authority_tests {
     use super::has_prevalidated_organization_write_authority;
 
     #[test]
-    fn direct_organization_write_is_an_explicit_import_authority() {
-        // Keep the organization-import shortcut independent from the old non-empty-collection
-        // sentinel. The route may import ciphers without collections when AccessImportExport grants
-        // organization-wide import authority; every other caller passes false.
+    fn organization_write_requires_a_validated_collection_or_full_access() {
         let no_collections: Vec<crate::db::models::CollectionId> = Vec::new();
-        assert!(has_prevalidated_organization_write_authority(true, Some(&no_collections), false));
-        assert!(!has_prevalidated_organization_write_authority(false, Some(&no_collections), false));
+        assert!(!has_prevalidated_organization_write_authority(Some(&no_collections), false));
+        assert!(!has_prevalidated_organization_write_authority(None, false));
 
         let collections = vec!["collection".to_owned().into()];
-        assert!(has_prevalidated_organization_write_authority(false, Some(&collections), false));
-        assert!(has_prevalidated_organization_write_authority(false, None, true));
+        assert!(has_prevalidated_organization_write_authority(Some(&collections), false));
+        assert!(has_prevalidated_organization_write_authority(None, true));
     }
 }
 

src/api/core/organizations.rs

@@ -1116,27 +1116,30 @@ async fn assigned_org_ciphers_json(
     Ok(Value::Array(ciphers_json))
 }
 
-// The organization cipher list the clients use for the admin vault view and for computing every
-// report (Exposed/Reused/Weak Passwords, Unsecured Websites, Inactive 2FA, ...) locally — Vaultwarden
-// has no server-side reports.
-//
-// Bitwarden computes organization reports locally from this list. `accessReports` therefore grants
-// the full organization cipher list, just like Admin/Owner or `editAnyCollection`; limiting it to the
-// caller's assignments makes organization-wide reports silently incomplete.
+// The organization cipher list the clients use for the admin vault view and for computing reports
+// locally. Admins/Owners and Custom members with `editAnyCollection` already reach every cipher.
+// `accessReports` alone only opens the endpoint for the caller's existing assignments: it must not
+// turn permission to compute reports into read access to otherwise inaccessible organization data.
 #[get("/ciphers/organization-details?<data..>")]
 async fn get_org_details(data: OrgIdData, headers: ManagerHeadersLoose, conn: DbConn) -> JsonResult {
     if data.organization_id != headers.membership.org_uuid {
         err_code!("Resource not found.", "Organization id's do not match", rocket::http::Status::NotFound.code);
     }
 
-    let ciphers_json = if may_read_all_organization_ciphers(&headers.membership) {
-        get_org_details_impl(&data.organization_id, &headers.host, &headers.user.uuid, &conn).await?
-    } else {
-        err_code!(
-            "Resource not found.",
-            "User does not have permission to read the organization ciphers",
-            rocket::http::Status::NotFound.code
-        );
+    let ciphers_json = match organization_report_scope(&headers.membership) {
+        OrganizationReportScope::Complete => {
+            get_org_details_impl(&data.organization_id, &headers.host, &headers.user.uuid, &conn).await?
+        }
+        OrganizationReportScope::Assigned => {
+            assigned_org_ciphers_json(&data.organization_id, &headers.host, &headers.user.uuid, &conn).await?
+        }
+        OrganizationReportScope::Denied => {
+            err_code!(
+                "Resource not found.",
+                "User does not have permission to read the organization ciphers",
+                rocket::http::Status::NotFound.code
+            );
+        }
     };
 
     Ok(Json(json!({
@@ -2353,7 +2356,7 @@ async fn bulk_public_keys(
 }
 
 use super::ciphers::CipherData;
-use super::ciphers::update_cipher_from_data_with_authority;
+use super::ciphers::update_cipher_from_data;
 
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -2386,17 +2389,13 @@ async fn post_org_import(
         err!("Organization not found", "Organization id's do not match");
     }
 
-    // Bitwarden authorizes an organization import on `AccessImportExport` *or* the regular
-    // per-collection Create/ImportCiphers authority. Keep the latter path for ordinary members while
-    // treating the named Custom permission as the organization-wide import shortcut it represents.
-    //
-    // A confirmed membership is required though: both checks below are confirmed-gated, so an
-    // invited/accepted member could otherwise only import ciphers without any collection — which lands
-    // unreachable, unmanaged ciphers in the organization.
+    // Organization imports are authorized per target collection. `accessImportExport` gates export,
+    // but does not replace Write authority on an existing collection or Create authority for a new
+    // one. Require confirmation independently so a membership with no target collection cannot create
+    // an unreachable organization cipher.
     if !headers.membership.has_status(MembershipStatus::Confirmed) {
         err!("You need to be a confirmed member of this organization to import into it")
     }
-    let has_org_wide_import_access = may_import_without_collection_access(&headers.membership);
 
     let data: ImportData = data.into_inner();
 
@@ -2434,13 +2433,16 @@ async fn post_org_import(
     // the write loop left the former behind even though the request failed.
     for col in &data.collections {
         if let Some(collection) = col.id.as_ref().and_then(|col_id| existing_collections.get(col_id)) {
-            if !has_org_wide_import_access
-                && headers.membership.atype < MembershipType::Admin
-                && !collection.is_writable_by_user(&headers.membership.user_uuid, &conn).await
-            {
+            let writable = collection.is_writable_by_user(&headers.membership.user_uuid, &conn).await;
+            if !may_import_to_collection(
+                &headers.membership,
+                OrganizationImportTarget::Existing {
+                    writable,
+                },
+            ) {
                 err!(Compact, "The current user isn't allowed to manage this collection")
             }
-        } else if !has_org_wide_import_access && !headers.membership.can_create_new_collections() {
+        } else if !may_import_to_collection(&headers.membership, OrganizationImportTarget::New) {
             err!(Compact, "The current user isn't allowed to create new collections")
         }
     }
@@ -2485,12 +2487,11 @@ async fn post_org_import(
         // Replace the client-provided, unvalidated organizationId with the real target org
         cipher_data.organization_id = Some(org_id.clone());
         let mut cipher = Cipher::new(cipher_data.r#type, cipher_data.name.clone());
-        update_cipher_from_data_with_authority(
+        update_cipher_from_data(
             &mut cipher,
             cipher_data,
             &headers,
             Some(collections.clone()),
-            has_org_wide_import_access,
             &conn,
             &nt,
             UpdateType::None,
@@ -3496,18 +3497,47 @@ async fn caller_may_grant_collection_manage(caller: &Membership, col_id: &Collec
     }
 }
 
-/// Whether a caller may import throughout the organization without proving Create/Write authority
-/// for every target collection. This is the server-side meaning of Bitwarden's
-/// `accessImportExport` Custom permission; Admins and Owners already have equivalent authority.
-fn may_import_without_collection_access(caller: &Membership) -> bool {
-    caller.has_status(MembershipStatus::Confirmed)
-        && (caller.atype >= MembershipType::Admin || caller.has_access_import_export())
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum OrganizationImportTarget {
+    Existing {
+        writable: bool,
+    },
+    New,
+}
+
+/// Organization imports retain the pre-existing per-target authorization model. The
+/// `accessImportExport` permission opens export, but it is deliberately not an organization-wide
+/// Create/Write shortcut for imports.
+fn may_import_to_collection(caller: &Membership, target: OrganizationImportTarget) -> bool {
+    if !caller.has_status(MembershipStatus::Confirmed) {
+        return false;
+    }
+
+    match target {
+        OrganizationImportTarget::Existing {
+            writable,
+        } => caller.atype >= MembershipType::Admin || writable,
+        OrganizationImportTarget::New => caller.can_create_new_collections(),
+    }
 }
 
-/// Organization reports are computed client-side and require every organization cipher. Match
-/// Bitwarden's `AccessReports` semantics instead of silently producing assignment-scoped reports.
-fn may_read_all_organization_ciphers(caller: &Membership) -> bool {
-    caller.has_full_access() || (caller.has_status(MembershipStatus::Confirmed) && caller.has_access_reports())
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum OrganizationReportScope {
+    Complete,
+    Assigned,
+    Denied,
+}
+
+/// Full-access members receive the complete organization view. `accessReports` alone receives the
+/// same assignment-scoped, restriction-bearing cipher representation as the caller's normal sync.
+fn organization_report_scope(caller: &Membership) -> OrganizationReportScope {
+    if caller.has_full_access() {
+        OrganizationReportScope::Complete
+    } else if caller.has_status(MembershipStatus::Confirmed) && caller.has_access_reports() {
+        OrganizationReportScope::Assigned
+    } else {
+        OrganizationReportScope::Denied
+    }
 }
 
 /// Whether `caller` may export the *entire* organization instead of only their own assignments.
@@ -3687,6 +3717,18 @@ async fn delete_group_impl(
         err!("Group support is disabled");
     }
 
+    let caller_can_manage_collections =
+        headers.membership_type >= MembershipType::Admin || headers.membership.has_full_access();
+    let group = authorize_group_deletion(group_id, org_id, caller_can_manage_collections, conn).await?;
+    delete_authorized_group(&group, org_id, headers, conn).await
+}
+
+async fn authorize_group_deletion(
+    group_id: &GroupId,
+    org_id: &OrganizationId,
+    caller_can_manage_collections: bool,
+    conn: &DbConn,
+) -> Result<Group, crate::Error> {
     let Some(group) = Group::find_by_uuid_and_org(group_id, org_id, conn).await else {
         err!("Group not found", "Group uuid is invalid or does not belong to the organization")
     };
@@ -3695,19 +3737,26 @@ async fn delete_group_impl(
     // collections) revokes that access for all its members. A custom user with only manage_groups
     // must not be able to affect collection access, so only callers who can actually manage
     // collections (Admins/Owners or users with full access) may delete such a group. Mirrors the
-    // restriction in put_group_members / post_delete_group_member. Also covers bulk_delete_groups,
-    // which funnels through this function.
-    let caller_can_manage_collections = headers.membership_type >= MembershipType::Admin
-        || match Membership::find_by_user_and_org(&headers.user.uuid, org_id, conn).await {
-            Some(m) => m.has_full_access(),
-            None => false,
-        };
-    if !caller_can_manage_collections
-        && (group.access_all || !CollectionGroup::find_by_group(group_id, org_id, conn).await.is_empty())
-    {
+    // restriction in put_group_members / post_delete_group_member.
+    let group_confers_collection_access =
+        group.access_all || !CollectionGroup::find_by_group(group_id, org_id, conn).await.is_empty();
+    if !may_delete_group(caller_can_manage_collections, group_confers_collection_access) {
         err!("You don't have permission to delete a group that grants collection access")
     }
 
+    Ok(group)
+}
+
+fn may_delete_group(caller_can_manage_collections: bool, group_confers_collection_access: bool) -> bool {
+    caller_can_manage_collections || !group_confers_collection_access
+}
+
+async fn delete_authorized_group(
+    group: &Group,
+    org_id: &OrganizationId,
+    headers: &ManageGroupsHeaders,
+    conn: &DbConn,
+) -> EmptyResult {
     log_event(
         EventType::GroupDeleted as i32,
         &group.uuid,
@@ -3738,8 +3787,22 @@ async fn bulk_delete_groups(
 
     let data: BulkGroupIds = data.into_inner();
 
+    // Authorize the complete request before the first event or deletion. In particular, a
+    // manageGroups-only caller may delete ordinary groups but not collection-bearing groups; a mixed
+    // batch must not delete an authorized prefix and then fail on a later item.
+    let caller_can_manage_collections =
+        headers.membership_type >= MembershipType::Admin || headers.membership.has_full_access();
+    let mut groups = Vec::with_capacity(data.ids.len());
+    let mut seen_group_ids = HashSet::with_capacity(data.ids.len());
     for group_id in data.ids {
-        delete_group_impl(&org_id, &group_id, &headers, &conn).await?;
+        if !seen_group_ids.insert(group_id.clone()) {
+            err!("Duplicate group id in bulk delete request")
+        }
+        groups.push(authorize_group_deletion(&group_id, &org_id, caller_can_manage_collections, &conn).await?);
+    }
+
+    for group in &groups {
+        delete_authorized_group(group, &org_id, &headers, &conn).await?;
     }
     Ok(())
 }
@@ -4265,12 +4328,12 @@ mod tests {
     use serde_json::{Value, json};
 
     use super::{
-        CollectionDetailsResponseScope, CustomRolePermissions, caller_manage_grant_role_check,
-        collection_bearing_membership_unchanged, collection_details_response_scope, filter_ciphers_for_organization,
-        may_change_group_membership, may_change_member_type, may_export_entire_organization,
-        may_import_without_collection_access, may_manage_member_type, may_manage_stored_member_type,
-        may_provision_member_type, may_provision_stored_member_type, may_read_all_organization_ciphers,
-        may_read_complete_collection_list,
+        CollectionDetailsResponseScope, CustomRolePermissions, OrganizationImportTarget, OrganizationReportScope,
+        caller_manage_grant_role_check, collection_bearing_membership_unchanged, collection_details_response_scope,
+        filter_ciphers_for_organization, may_change_group_membership, may_change_member_type, may_delete_group,
+        may_export_entire_organization, may_import_to_collection, may_manage_member_type,
+        may_manage_stored_member_type, may_provision_member_type, may_provision_stored_member_type,
+        may_read_complete_collection_list, organization_report_scope,
     };
     use crate::db::models::{Cipher, GroupId, Membership, MembershipStatus, MembershipType, OrganizationId};
 
@@ -4376,36 +4439,93 @@ mod tests {
     }
 
     #[test]
-    fn access_import_export_opens_the_organization_import() {
+    fn access_import_export_does_not_replace_import_collection_authority() {
         let mut import_export = confirmed_member(MembershipType::Custom);
         import_export.access_import_export = true;
-        assert!(may_import_without_collection_access(&import_export));
+        assert!(!may_import_to_collection(
+            &import_export,
+            OrganizationImportTarget::Existing {
+                writable: false
+            }
+        ));
+        assert!(!may_import_to_collection(&import_export, OrganizationImportTarget::New));
+
+        assert!(may_import_to_collection(
+            &import_export,
+            OrganizationImportTarget::Existing {
+                writable: true
+            }
+        ));
+
+        let mut create = confirmed_member(MembershipType::Custom);
+        create.create_new_collections = true;
+        assert!(may_import_to_collection(&create, OrganizationImportTarget::New));
+
+        let mut edit_any = confirmed_member(MembershipType::Custom);
+        edit_any.edit_any_collection = true;
+        assert!(!may_import_to_collection(&edit_any, OrganizationImportTarget::New));
+
+        assert!(may_import_to_collection(
+            &confirmed_member(MembershipType::User),
+            OrganizationImportTarget::Existing {
+                writable: true
+            }
+        ));
 
-        assert!(!may_import_without_collection_access(&confirmed_member(MembershipType::Custom)));
-        assert!(!may_import_without_collection_access(&confirmed_member(MembershipType::User)));
-        assert!(may_import_without_collection_access(&confirmed_member(MembershipType::Admin)));
-        assert!(may_import_without_collection_access(&confirmed_member(MembershipType::Owner)));
+        assert!(may_import_to_collection(
+            &confirmed_member(MembershipType::Admin),
+            OrganizationImportTarget::Existing {
+                writable: false
+            }
+        ));
+        assert!(may_import_to_collection(&confirmed_member(MembershipType::Owner), OrganizationImportTarget::New));
 
         import_export.status = MembershipStatus::Accepted as i32;
-        assert!(!may_import_without_collection_access(&import_export));
+        assert!(!may_import_to_collection(
+            &import_export,
+            OrganizationImportTarget::Existing {
+                writable: true
+            }
+        ));
     }
 
     #[test]
-    fn access_reports_grants_the_complete_report_input() {
+    fn access_reports_is_assignment_scoped_without_full_access() {
         let mut reports = confirmed_member(MembershipType::Custom);
         reports.access_reports = true;
-        assert!(may_read_all_organization_ciphers(&reports));
+        assert_eq!(organization_report_scope(&reports), OrganizationReportScope::Assigned);
 
-        assert!(!may_read_all_organization_ciphers(&confirmed_member(MembershipType::Custom)));
-        assert!(may_read_all_organization_ciphers(&confirmed_member(MembershipType::Admin)));
-        assert!(may_read_all_organization_ciphers(&confirmed_member(MembershipType::Owner)));
+        assert_eq!(
+            organization_report_scope(&confirmed_member(MembershipType::Custom)),
+            OrganizationReportScope::Denied
+        );
+        assert_eq!(
+            organization_report_scope(&confirmed_member(MembershipType::Admin)),
+            OrganizationReportScope::Complete
+        );
+        assert_eq!(
+            organization_report_scope(&confirmed_member(MembershipType::Owner)),
+            OrganizationReportScope::Complete
+        );
+
+        reports.edit_any_collection = true;
+        assert_eq!(organization_report_scope(&reports), OrganizationReportScope::Complete);
+        reports.edit_any_collection = false;
 
         reports.status = MembershipStatus::Accepted as i32;
-        assert!(!may_read_all_organization_ciphers(&reports));
+        assert_eq!(organization_report_scope(&reports), OrganizationReportScope::Denied);
 
         let mut stale_user = confirmed_member(MembershipType::User);
         stale_user.access_reports = true;
-        assert!(!may_read_all_organization_ciphers(&stale_user));
+        assert_eq!(organization_report_scope(&stale_user), OrganizationReportScope::Denied);
+    }
+
+    #[test]
+    fn collection_bearing_group_deletion_requires_collection_authority() {
+        assert!(may_delete_group(false, false));
+        assert!(!may_delete_group(false, true));
+        assert!(may_delete_group(true, false));
+        assert!(may_delete_group(true, true));
     }
 
     #[test]

src/db/mod.rs

@@ -2960,6 +2960,46 @@ mod custom_role_rollback_sql_tests {
     }
 }
 
+/// MySQL/MariaDB use numeric comparison when one side of an equality is numeric. A malformed
+/// rollback allowlist with an INT column can consequently select UUIDs that were never placed on
+/// the list. These source-level contract tests complement the backend rollback tests: both entry
+/// points must validate the documented CHAR(36) shape before their first role-mapping query.
+#[cfg(test)]
+mod mysql_custom_role_rollback_sql_tests {
+    const STANDALONE_ROLLBACK: &str = include_str!("../../tools/custom_role_rollback/mysql.sql");
+    const DIESEL_DOWN_MIGRATION: &str =
+        include_str!("../../migrations/mysql/2026-06-30-120000_add_custom_role_permissions/down.sql");
+    const ROLE_MAPPING: &str = "uuid IN (SELECT users_organizations_uuid FROM __vw_rollback_manager_allowlist)";
+    const STANDALONE_FIRST_MUTATION: &str = "ALTER TABLE users_organizations ADD COLUMN access_all";
+    const DIESEL_FIRST_AUTHORIZATION_MUTATION: &str = "UPDATE users_organizations SET atype = 3";
+
+    fn assert_char_36_guard_precedes_mutation(sql: &str, mutation: &str, expected_guard_copies: usize) {
+        let role_mapping = sql.find(ROLE_MAPPING).expect("rollback must contain the allowlist role mapping");
+        let mutation = sql.find(mutation).expect("rollback must contain the guarded mutation");
+        assert!(mutation <= role_mapping, "the selected boundary must precede the role mapping");
+        let preconditions = &sql[..mutation];
+
+        assert_eq!(
+            preconditions.matches("data_type = 'char'").count(),
+            expected_guard_copies,
+            "every allowlist shape check must require a character column"
+        );
+        assert_eq!(
+            preconditions.matches("character_maximum_length = 36").count(),
+            expected_guard_copies,
+            "every allowlist shape check must require the complete UUID length"
+        );
+    }
+
+    #[test]
+    fn non_char_36_allowlists_are_rejected_before_mysql_role_mapping() {
+        // The standalone script duplicates each predicate: once for its readable diagnostic and
+        // once for the duplicate-key guard that actually stops execution.
+        assert_char_36_guard_precedes_mutation(STANDALONE_ROLLBACK, STANDALONE_FIRST_MUTATION, 2);
+        assert_char_36_guard_precedes_mutation(DIESEL_DOWN_MIGRATION, DIESEL_FIRST_AUTHORIZATION_MUTATION, 1);
+    }
+}
+
 #[cfg(test)]
 mod custom_role_migration_preflight_tests {
     use std::error::Error as _;

tools/custom_role_rollback/mysql.sql

@@ -194,11 +194,13 @@ WHERE c.n <> 1;
 --    hand-written or colliding table without a usable `users_organizations_uuid` column would pass
 --    every check above and then fail on the first SELECT against it -- which happens *after* the
 --    `ADD COLUMN` below has already committed implicitly, leaving a half-converted database.
---    Require exactly one non-nullable, uniquely indexed column of that name.
+--    Require exactly one non-nullable, uniquely indexed CHAR(36) column of that name. Checking the
+--    type is part of the authorization boundary: MySQL/MariaDB compare a character UUID with a
+--    numeric allowlist as numbers, so an INT value such as 0 could match unrelated UUIDs.
 SELECT CONCAT(
     'REFUSED, nothing was changed: __vw_rollback_manager_allowlist must have exactly one column ',
-    'named users_organizations_uuid, NOT NULL and uniquely indexed. Create it as documented in ',
-    'README.md.'
+    'named users_organizations_uuid, typed CHAR(36), NOT NULL and uniquely indexed. Create it as ',
+    'documented in README.md.'
 ) AS rollback_precondition_failure
 FROM (
     SELECT
@@ -206,7 +208,9 @@ FROM (
           WHERE table_schema = DATABASE() AND table_name = '__vw_rollback_manager_allowlist') AS cols,
         (SELECT COUNT(*) FROM information_schema.columns
           WHERE table_schema = DATABASE() AND table_name = '__vw_rollback_manager_allowlist'
-            AND column_name = 'users_organizations_uuid' AND is_nullable = 'NO') AS usable,
+            AND column_name = 'users_organizations_uuid'
+            AND data_type = 'char' AND character_maximum_length = 36
+            AND is_nullable = 'NO') AS usable,
         (SELECT COUNT(*) FROM information_schema.statistics
           WHERE table_schema = DATABASE() AND table_name = '__vw_rollback_manager_allowlist'
             AND column_name = 'users_organizations_uuid' AND non_unique = 0) AS uniq
@@ -220,7 +224,9 @@ FROM (
           WHERE table_schema = DATABASE() AND table_name = '__vw_rollback_manager_allowlist') AS cols,
         (SELECT COUNT(*) FROM information_schema.columns
           WHERE table_schema = DATABASE() AND table_name = '__vw_rollback_manager_allowlist'
-            AND column_name = 'users_organizations_uuid' AND is_nullable = 'NO') AS usable,
+            AND column_name = 'users_organizations_uuid'
+            AND data_type = 'char' AND character_maximum_length = 36
+            AND is_nullable = 'NO') AS usable,
         (SELECT COUNT(*) FROM information_schema.statistics
           WHERE table_schema = DATABASE() AND table_name = '__vw_rollback_manager_allowlist'
             AND column_name = 'users_organizations_uuid' AND non_unique = 0) AS uniq