diff --git a/CHANGELOG.md b/CHANGELOG.md index 23b049f5c3..34f2e5066f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,10 +8,25 @@ ## [Unreleased] +### Added + +- New REST API endpoint `POST /v1/tools/generate-schemas-from-openapi` for generating MCP tool schemas from OpenAPI specifications without admin UI dependencies (#5142) + ### Fixed +- Fixed RBAC seeder race condition that produced HTTP 500 under concurrent bootstrap: added partial unique indexes on `roles(name, scope) WHERE is_active` and `user_roles` equivalent columns, plus savepoint/retry in `RoleService.create_role()` and `assign_role_to_user()`. The migration (`d21698ae4a19`) now also remaps `user_roles.role_id` from duplicate roles to the kept role before deactivating the duplicates (so `list_user_roles()` joins remain intact), and prefers unexpired / most-recently-granted assignments when deduplicating user-role rows (#4636) - **Custom Auth Headers on Tools** ([#5314](https://github.com/IBM/mcp-context-forge/pull/5314), [#5201](https://github.com/IBM/mcp-context-forge/issues/5201)) - `POST /tools` and `PUT /tools/{tool_id}` now persist the `auth_headers` array instead of silently storing `auth_value: null`. Invalid header keys/values are rejected with a 422 rather than an unhandled 500. +### Security + +- Fixed cross-environment JWT acceptance (GHSA-vgf8-3685-66j9, CVE pending). Gateway-issued tokens + now carry an `env` claim and reject environment mismatches by default (`EMBED_ENVIRONMENT_IN_TOKENS=true`, + `VALIDATE_TOKEN_ENVIRONMENT=true`). Added optional `DERIVE_KEY_PER_ENVIRONMENT` to bind the HS* + signing key (including explicit-secret mints) to the deployment environment, which also closes legacy + tokens lacking an `env` claim. **Upgrade:** use a distinct `JWT_SECRET_KEY` per environment and + rotate long-lived tokens; enabling `DERIVE_KEY_PER_ENVIRONMENT` invalidates tokens issued before it + was turned on. RS*/ES* deployments must use distinct key pairs per environment. + ### Changed - **Stricter `authheaders` Key Validation (Gateways, Tools, A2A Agents)** ([#5314](https://github.com/IBM/mcp-context-forge/pull/5314)) - Header-key validation is now shared across all create/update schemas and the admin form. Keys with embedded whitespace (e.g. `X Api Key`) were previously accepted and stored as invalid HTTP header names that failed at invocation time; they are now rejected with a 422 at config time, and surrounding whitespace is trimmed before storage. Gateway or A2A configs relying on the old behavior will need their header keys corrected on the next update. diff --git a/mcpgateway/alembic/versions/d21698ae4a19_add_rbac_unique_constraints_race_fix.py b/mcpgateway/alembic/versions/d21698ae4a19_add_rbac_unique_constraints_race_fix.py new file mode 100644 index 0000000000..7c45b92921 --- /dev/null +++ b/mcpgateway/alembic/versions/d21698ae4a19_add_rbac_unique_constraints_race_fix.py @@ -0,0 +1,382 @@ +# -*- coding: utf-8 -*- +"""Location: ./mcpgateway/alembic/versions/d21698ae4a19_add_rbac_unique_constraints_race_fix.py +Copyright contributors to the MCP-CONTEXT-FORGE project +SPDX-License-Identifier: Apache-2.0 + +add_rbac_unique_constraints_race_fix + +Fixes issue #4482 - RBAC role/user_role seeder race when fast-path skips advisory lock. + +Adds database-level unique constraints to prevent duplicate active roles and user role +assignments when multiple replicas/workers bootstrap concurrently. This is defense-in-depth: +even with the advisory lock in place today, the DB should be the ultimate authority on +uniqueness. + +Changes: +1. Deduplicates any existing duplicate active rows: + - roles: keeps oldest by created_at (lowest id on ties) + - user_roles: keeps unexpired rows first, then newest granted_at, then lowest id + - remaps user_roles.role_id to the kept role before deactivating duplicates +2. Adds partial unique index on roles(name, scope) WHERE is_active = true +3. Adds partial unique indexes on user_roles for both nullable and non-nullable scope_id cases + +Supports both PostgreSQL and SQLite databases. + +NOTE: These indexes are ALSO defined in mcpgateway/db.py (__table_args__) for fresh databases. +This migration only runs on existing databases (if tables already exist). The migration is +idempotent and checks if indexes exist before creating them, ensuring no conflicts between +the two definition sources. + +Revision ID: d21698ae4a19 +Revises: b6c7d8e9f0a1 +Create Date: 2026-05-06 12:35:58.142694 + +""" +from typing import Sequence, Union + +from alembic import op +from sqlalchemy import text, inspect + + +# revision identifiers, used by Alembic. +revision: str = 'd21698ae4a19' # pragma: allowlist secret +down_revision: Union[str, Sequence[str], None] = 'b6c7d8e9f0a1' # pragma: allowlist secret +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema - add unique constraints for RBAC tables.""" + bind = op.get_bind() + inspector = inspect(bind) + dialect = bind.dialect.name + + # Skip if tables don't exist (fresh DB uses db.py models directly) + existing_tables = inspector.get_table_names() + if "roles" not in existing_tables or "user_roles" not in existing_tables: + return + + # ============================================================================= + # STEP 1: Deduplicate roles table - keep oldest active role by (name, scope) + # ============================================================================= + + # Find duplicate active roles (same name+scope combination) + # Strategy: keep oldest by created_at (lowest id on ties), soft-delete the rest + # + # IMPORTANT: Before deactivating duplicate roles, remap any user_roles that reference + # a duplicate role to the kept (winner) role. list_user_roles() joins on roles.is_active, + # so assignments pointing at a deactivated role would silently disappear from results. + + # Use row_number() for deterministic deduplication (handles timestamp ties via id ordering) + # Keep the row with row_number = 1 (oldest created_at, lowest id on ties) + if dialect == 'postgresql': + # Step 1a: Remap user_roles.role_id from duplicate roles to the kept role + remap_user_roles_sql = text(""" + UPDATE user_roles + SET role_id = keeper.id + FROM ( + SELECT id, + name, + scope, + ROW_NUMBER() OVER (PARTITION BY name, scope ORDER BY created_at, id) as rn + FROM roles + WHERE is_active = true + ) keeper + JOIN ( + SELECT id AS dup_id, + name, + scope, + ROW_NUMBER() OVER (PARTITION BY name, scope ORDER BY created_at, id) as rn + FROM roles + WHERE is_active = true + ) dup ON dup.name = keeper.name AND dup.scope = keeper.scope + WHERE keeper.rn = 1 + AND dup.rn > 1 + AND user_roles.role_id = dup.dup_id + """) + dedupe_roles_sql = text(""" + UPDATE roles + SET is_active = false + WHERE id IN ( + SELECT id + FROM ( + SELECT id, + ROW_NUMBER() OVER (PARTITION BY name, scope ORDER BY created_at, id) as rn + FROM roles + WHERE is_active = true + ) ranked + WHERE rn > 1 + ) + """) + else: # SQLite + # Step 1a: Remap user_roles.role_id from duplicate roles to the kept role + remap_user_roles_sql = text(""" + UPDATE user_roles + SET role_id = ( + SELECT id + FROM roles r2 + WHERE r2.name = (SELECT name FROM roles r3 WHERE r3.id = user_roles.role_id) + AND r2.scope = (SELECT scope FROM roles r3 WHERE r3.id = user_roles.role_id) + AND r2.is_active = 1 + ORDER BY r2.created_at, r2.id + LIMIT 1 + ) + WHERE role_id IN ( + SELECT id + FROM ( + SELECT id, + ROW_NUMBER() OVER (PARTITION BY name, scope ORDER BY created_at, id) as rn + FROM roles + WHERE is_active = 1 + ) ranked + WHERE rn > 1 + ) + """) + dedupe_roles_sql = text(""" + UPDATE roles + SET is_active = 0 + WHERE id IN ( + SELECT id + FROM ( + SELECT id, + ROW_NUMBER() OVER (PARTITION BY name, scope ORDER BY created_at, id) as rn + FROM roles + WHERE is_active = 1 + ) ranked + WHERE rn > 1 + ) + """) + + # First remap assignments, then deactivate duplicate roles + remap_result = bind.execute(remap_user_roles_sql) + remapped_user_roles_count = remap_result.rowcount + if remapped_user_roles_count > 0: + print(f"Remapped {remapped_user_roles_count} user_role assignment(s) from duplicate roles to kept roles") + + result = bind.execute(dedupe_roles_sql) + deduped_roles_count = result.rowcount + if deduped_roles_count > 0: + print(f"Deduped {deduped_roles_count} duplicate active role(s) - set is_active=false for newer duplicates") + + # ============================================================================= + # STEP 2: Deduplicate user_roles table + # ============================================================================= + + # For user_roles, we need to handle scope_id IS NULL and IS NOT NULL separately + # Strategy: soft-delete duplicates. + # + # Preference order for the kept row (rn = 1): + # 1. Unexpired assignments before expired ones (NULL expires_at counts as never-expiring) + # 2. Among ties, newest granted_at (most recently granted wins) + # 3. Among ties, lowest id for determinism + # + # This avoids the original bug where keeping the oldest-granted row could preserve + # an already-expired assignment while discarding a still-valid duplicate. + + # Handle scope_id IS NULL case + if dialect == 'postgresql': + dedupe_user_roles_null_scope_sql = text(""" + UPDATE user_roles + SET is_active = false + WHERE id IN ( + SELECT id + FROM ( + SELECT id, + ROW_NUMBER() OVER ( + PARTITION BY user_email, role_id, scope + ORDER BY + CASE WHEN expires_at IS NULL OR expires_at > NOW() THEN 0 ELSE 1 END, + granted_at DESC, + id + ) as rn + FROM user_roles + WHERE is_active = true AND scope_id IS NULL + ) ranked + WHERE rn > 1 + ) + """) + else: # SQLite + dedupe_user_roles_null_scope_sql = text(""" + UPDATE user_roles + SET is_active = 0 + WHERE id IN ( + SELECT id + FROM ( + SELECT id, + ROW_NUMBER() OVER ( + PARTITION BY user_email, role_id, scope + ORDER BY + CASE WHEN expires_at IS NULL OR expires_at > datetime('now') THEN 0 ELSE 1 END, + granted_at DESC, + id + ) as rn + FROM user_roles + WHERE is_active = 1 AND scope_id IS NULL + ) ranked + WHERE rn > 1 + ) + """) + + result = bind.execute(dedupe_user_roles_null_scope_sql) + deduped_user_roles_null = result.rowcount + + # Handle scope_id IS NOT NULL case + if dialect == 'postgresql': + dedupe_user_roles_with_scope_sql = text(""" + UPDATE user_roles + SET is_active = false + WHERE id IN ( + SELECT id + FROM ( + SELECT id, + ROW_NUMBER() OVER ( + PARTITION BY user_email, role_id, scope, scope_id + ORDER BY + CASE WHEN expires_at IS NULL OR expires_at > NOW() THEN 0 ELSE 1 END, + granted_at DESC, + id + ) as rn + FROM user_roles + WHERE is_active = true AND scope_id IS NOT NULL + ) ranked + WHERE rn > 1 + ) + """) + else: # SQLite + dedupe_user_roles_with_scope_sql = text(""" + UPDATE user_roles + SET is_active = 0 + WHERE id IN ( + SELECT id + FROM ( + SELECT id, + ROW_NUMBER() OVER ( + PARTITION BY user_email, role_id, scope, scope_id + ORDER BY + CASE WHEN expires_at IS NULL OR expires_at > datetime('now') THEN 0 ELSE 1 END, + granted_at DESC, + id + ) as rn + FROM user_roles + WHERE is_active = 1 AND scope_id IS NOT NULL + ) ranked + WHERE rn > 1 + ) + """) + + result = bind.execute(dedupe_user_roles_with_scope_sql) + deduped_user_roles_with_scope = result.rowcount + + total_deduped_user_roles = deduped_user_roles_null + deduped_user_roles_with_scope + if total_deduped_user_roles > 0: + print(f"Deduped {total_deduped_user_roles} duplicate active user_role(s) - kept unexpired/newest-granted assignment, set is_active=false for others") + + # ============================================================================= + # STEP 3: Add partial unique indexes + # ============================================================================= + + # Check if indexes already exist (idempotency) + existing_indexes = [idx['name'] for idx in inspector.get_indexes('roles')] + + # Partial unique index on roles(name, scope) WHERE is_active = true + # This prevents duplicate active roles with same name+scope + if 'uq_roles_name_scope_active' not in existing_indexes: + if dialect == 'postgresql': + bind.execute(text( + "CREATE UNIQUE INDEX uq_roles_name_scope_active " + "ON roles (name, scope) " + "WHERE is_active = true" + )) + elif dialect == 'sqlite': + bind.execute(text( + "CREATE UNIQUE INDEX uq_roles_name_scope_active " + "ON roles (name, scope) " + "WHERE is_active = 1" + )) + else: + # For other databases, create without WHERE clause (less optimal but works) + # Note: This will only allow one row per (name, scope) total, not just active ones + print(f"WARNING: Dialect '{dialect}' may not support partial indexes. Creating full unique index.") + bind.execute(text( + "CREATE UNIQUE INDEX uq_roles_name_scope_active " + "ON roles (name, scope)" + )) + print("Created unique index: uq_roles_name_scope_active") + + # Check user_roles indexes + existing_user_roles_indexes = [idx['name'] for idx in inspector.get_indexes('user_roles')] + + # Partial unique index on user_roles(user_email, role_id, scope) WHERE scope_id IS NULL AND is_active = true + if 'uq_user_roles_email_role_scope_null_active' not in existing_user_roles_indexes: + if dialect == 'postgresql': + bind.execute(text( + "CREATE UNIQUE INDEX uq_user_roles_email_role_scope_null_active " + "ON user_roles (user_email, role_id, scope) " + "WHERE scope_id IS NULL AND is_active = true" + )) + elif dialect == 'sqlite': + bind.execute(text( + "CREATE UNIQUE INDEX uq_user_roles_email_role_scope_null_active " + "ON user_roles (user_email, role_id, scope) " + "WHERE scope_id IS NULL AND is_active = 1" + )) + else: + # Fallback: unique on (user_email, role_id, scope) without WHERE + print(f"WARNING: Dialect '{dialect}' may not support partial indexes. Creating full unique index.") + bind.execute(text( + "CREATE UNIQUE INDEX uq_user_roles_email_role_scope_null_active " + "ON user_roles (user_email, role_id, scope)" + )) + print("Created unique index: uq_user_roles_email_role_scope_null_active") + + # Partial unique index on user_roles(user_email, role_id, scope, scope_id) WHERE scope_id IS NOT NULL AND is_active = true + if 'uq_user_roles_email_role_scope_id_active' not in existing_user_roles_indexes: + if dialect == 'postgresql': + bind.execute(text( + "CREATE UNIQUE INDEX uq_user_roles_email_role_scope_id_active " + "ON user_roles (user_email, role_id, scope, scope_id) " + "WHERE scope_id IS NOT NULL AND is_active = true" + )) + elif dialect == 'sqlite': + bind.execute(text( + "CREATE UNIQUE INDEX uq_user_roles_email_role_scope_id_active " + "ON user_roles (user_email, role_id, scope, scope_id) " + "WHERE scope_id IS NOT NULL AND is_active = 1" + )) + else: + # Fallback: unique on all four columns without WHERE + print(f"WARNING: Dialect '{dialect}' may not support partial indexes. Creating full unique index.") + bind.execute(text( + "CREATE UNIQUE INDEX uq_user_roles_email_role_scope_id_active " + "ON user_roles (user_email, role_id, scope, scope_id)" + )) + print("Created unique index: uq_user_roles_email_role_scope_id_active") + + +def downgrade() -> None: + """Downgrade schema - remove unique constraints.""" + bind = op.get_bind() + inspector = inspect(bind) + + # Skip if tables don't exist + existing_tables = inspector.get_table_names() + if "roles" not in existing_tables or "user_roles" not in existing_tables: + return + + # Drop the unique indexes if they exist + existing_indexes = [idx['name'] for idx in inspector.get_indexes('roles')] + if 'uq_roles_name_scope_active' in existing_indexes: + op.drop_index('uq_roles_name_scope_active', table_name='roles') + print("Dropped unique index: uq_roles_name_scope_active") + + existing_user_roles_indexes = [idx['name'] for idx in inspector.get_indexes('user_roles')] + if 'uq_user_roles_email_role_scope_null_active' in existing_user_roles_indexes: + op.drop_index('uq_user_roles_email_role_scope_null_active', table_name='user_roles') + print("Dropped unique index: uq_user_roles_email_role_scope_null_active") + + if 'uq_user_roles_email_role_scope_id_active' in existing_user_roles_indexes: + op.drop_index('uq_user_roles_email_role_scope_id_active', table_name='user_roles') + print("Dropped unique index: uq_user_roles_email_role_scope_id_active") + + # Note: We do NOT reactivate the deduped rows on downgrade + # The soft-deleted duplicates remain inactive for audit purposes diff --git a/mcpgateway/db.py b/mcpgateway/db.py index b5d22b343a..3f4eec2192 100644 --- a/mcpgateway/db.py +++ b/mcpgateway/db.py @@ -1155,6 +1155,17 @@ class Role(Base): """Role model for RBAC system.""" __tablename__ = "roles" + __table_args__ = ( + # Partial unique index: only one active role per (name, scope) combination + # This prevents race conditions when multiple processes try to create the same role + # + # NOTE: This constraint is defined in BOTH db.py and alembic migration d21698ae4a19: + # - db.py (__table_args__): Creates indexes for FRESH databases (CREATE TABLE) + # - Alembic migration: Creates indexes for EXISTING databases (ALTER TABLE) + # The migration is idempotent and checks if indexes exist before creating them. + # This dual-definition ensures indexes exist regardless of database initialization path. + Index("uq_roles_name_scope_active", "name", "scope", unique=True, postgresql_where=text("is_active = true"), sqlite_where=text("is_active = 1")), + ) # Primary key id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4())) @@ -1197,6 +1208,40 @@ class UserRole(Base): """User role assignment model.""" __tablename__ = "user_roles" + __table_args__ = ( + # Partial unique indexes: only one active assignment per (user, role, scope, scope_id) combination + # Need two separate indexes to handle NULL vs non-NULL scope_id cases (SQL NULL != NULL semantics) + # + # Note: We only check is_active here (not expires_at) because SQLite forbids non-deterministic + # functions like datetime('now') in partial index WHERE clauses. Expiration filtering happens + # in the application layer: assign_role_to_user() soft-deletes expired assignments before creating + # new ones, and the IntegrityError refetch path checks expiration. + # + # NOTE: These constraints are defined in BOTH db.py and alembic migration d21698ae4a19: + # - db.py (__table_args__): Creates indexes for FRESH databases (CREATE TABLE) + # - Alembic migration: Creates indexes for EXISTING databases (ALTER TABLE) + # The migration is idempotent and checks if indexes exist before creating them. + # This dual-definition ensures indexes exist regardless of database initialization path. + Index( + "uq_user_roles_email_role_scope_null_active", + "user_email", + "role_id", + "scope", + unique=True, + postgresql_where=text("scope_id IS NULL AND is_active = true"), + sqlite_where=text("scope_id IS NULL AND is_active = 1"), + ), + Index( + "uq_user_roles_email_role_scope_id_active", + "user_email", + "role_id", + "scope", + "scope_id", + unique=True, + postgresql_where=text("scope_id IS NOT NULL AND is_active = true"), + sqlite_where=text("scope_id IS NOT NULL AND is_active = 1"), + ), + ) # Primary key id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4())) diff --git a/mcpgateway/services/role_service.py b/mcpgateway/services/role_service.py index dbc663cee9..b77eb134da 100644 --- a/mcpgateway/services/role_service.py +++ b/mcpgateway/services/role_service.py @@ -16,6 +16,7 @@ # Third-Party from sqlalchemy import and_, delete, select, update +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session # First-Party @@ -217,15 +218,40 @@ async def create_role(self, name: str, description: str, scope: str, permissions if await self._would_create_cycle(inherits_from, None): raise ValueError("Role inheritance would create a cycle") - # Create the role + # Create the role with savepoint to handle race conditions + # If another process created the same role concurrently, we'll catch IntegrityError + # and return the existing role instead of failing role = Role(name=name, description=description, scope=scope, permissions=permissions, created_by=created_by, inherits_from=inherits_from, is_system_role=is_system_role) - self.db.add(role) - self.db.commit() - self.db.refresh(role) + try: + # Use nested transaction (savepoint) to allow rollback on conflict + # Try to use begin_nested if available (won't be in Mock objects for tests) + try: + with self.db.begin_nested(): + self.db.add(role) + except (AttributeError, TypeError): + # Mock object or DB doesn't support savepoints - use regular add + self.db.add(role) - logger.info("Created role: %s (scope: %s, id: %s)", role.name, role.scope, role.id) - return role + self.db.commit() + self.db.refresh(role) + + logger.info("Created role: %s (scope: %s, id: %s)", role.name, role.scope, role.id) + return role + + except IntegrityError as e: + # Another process created this role concurrently - rollback savepoint and refetch + self.db.rollback() + logger.info("Role '%s' (scope: %s) was created concurrently by another process - refetching existing role", name, scope) + + # Refetch the winner's row + existing = await self.get_role_by_name(name, scope) + if existing: + return existing + + # If we still can't find it, something else went wrong - re-raise + logger.error("IntegrityError but role not found after refetch: %s", e) + raise ValueError(f"Failed to create or fetch role '{name}' in scope '{scope}': {e}") from e async def get_role_by_id(self, role_id: str) -> Optional[Role]: """Get role by ID. @@ -618,24 +644,80 @@ async def assign_role_to_user( # Check for existing active assignment existing = await self.get_user_role_assignment(user_email, role_id, scope, scope_id) - if existing and existing.is_active and not existing.is_expired(): - raise ValueError("User already has this role assignment") - - # Create the assignment + if existing and existing.is_active: + if not existing.is_expired(): + # Active and not expired - reject the new assignment + raise ValueError("User already has this role assignment") + + # Active but expired - soft-delete it to allow the new assignment + # This handles the SQLite limitation where we can't use datetime('now') in partial indexes + # Wrap in try/except to handle race where another process soft-deletes concurrently + try: + existing.is_active = False + self.db.commit() + logger.info("Soft-deleted expired assignment for %s to role %s (scope: %s, scope_id: %s) to allow re-grant", user_email, role_id, scope, scope_id) + except IntegrityError: + # Another process modified this row concurrently - rollback and refetch + self.db.rollback() + logger.info("Concurrent modification detected during expired assignment soft-delete - refetching") + + # Refetch to see the current state + existing = await self.get_user_role_assignment(user_email, role_id, scope, scope_id) + if existing and existing.is_active and not existing.is_expired(): + # Another process already created a fresh assignment + return existing + # If no active assignment exists, continue to create new one below + logger.info("No active assignment found after concurrent soft-delete race - proceeding with creation") + + # Create the assignment with savepoint to handle race conditions + # If another process created the same assignment concurrently, we'll catch IntegrityError + # and return the existing assignment instead of failing user_role = UserRole(user_email=user_email, role_id=role_id, scope=scope, scope_id=scope_id, granted_by=granted_by, expires_at=expires_at, grant_source=grant_source) - self.db.add(user_role) - self.db.commit() - self.db.refresh(user_role) - - logger.info( - "Assigned role %s to %s (scope: %s, scope_id: %s)", - SecurityValidator.sanitize_log_message(role.name), - SecurityValidator.sanitize_log_message(user_email), - SecurityValidator.sanitize_log_message(scope), - SecurityValidator.sanitize_log_message(scope_id), - ) - return user_role + try: + # Use nested transaction (savepoint) to allow rollback on conflict + # Try to use begin_nested if available (won't be in Mock objects for tests) + try: + with self.db.begin_nested(): + self.db.add(user_role) + except (AttributeError, TypeError): + # Mock object or DB doesn't support savepoints - use regular add + self.db.add(user_role) + + self.db.commit() + self.db.refresh(user_role) + + logger.info( + "Assigned role %s to %s (scope: %s, scope_id: %s)", + SecurityValidator.sanitize_log_message(role.name), + SecurityValidator.sanitize_log_message(user_email), + SecurityValidator.sanitize_log_message(scope), + SecurityValidator.sanitize_log_message(scope_id), + ) + return user_role + + except IntegrityError as e: + # Another process created this assignment concurrently - rollback savepoint and refetch + self.db.rollback() + logger.info("Role assignment for %s to role %s (scope: %s, scope_id: %s) was created concurrently - refetching existing assignment", user_email, role_id, scope, scope_id) + + # Refetch the winner's row + existing = await self.get_user_role_assignment(user_email, role_id, scope, scope_id) + if existing: + # CRITICAL: Check if the refetched assignment is expired + # This handles the race where another process created an expired assignment + # or the assignment expired between creation and refetch + if existing.is_expired(): + logger.warning("Refetched assignment for %s to role %s is expired - soft-deleting and retrying", user_email, role_id) + existing.is_active = False + self.db.commit() + # Raise ValueError to signal that the refetched assignment was expired + raise ValueError(f"Refetched assignment for {user_email} to role {role_id} was expired") + return existing + + # If we still can't find it, something else went wrong - re-raise + logger.error("IntegrityError but user_role assignment not found after refetch: %s", e) + raise ValueError(f"Failed to create or fetch role assignment for {user_email} to role {role_id}: {e}") from e async def revoke_role_from_user(self, user_email: str, role_id: str, scope: str, scope_id: Optional[str]) -> bool: """Revoke a role from a user. diff --git a/tests/integration/test_role_service_duplicate_handling.py b/tests/integration/test_role_service_duplicate_handling.py index 5bff83908d..add8e73393 100644 --- a/tests/integration/test_role_service_duplicate_handling.py +++ b/tests/integration/test_role_service_duplicate_handling.py @@ -156,3 +156,623 @@ async def test_migration_cleanup_removes_inactive_duplicates(test_db: Session, t assert len(all_after) == 1 assert all_after[0].is_active is True assert all_after[0].id == active.id + + +@pytest.mark.asyncio +async def test_concurrent_role_creation_handles_integrity_error(test_db: Session): + """Test that concurrent role creation handles IntegrityError gracefully. + + This tests the race condition handling in create_role() lines 242-254. + We force an IntegrityError by directly inserting a duplicate in the database + after the duplicate check but before commit. + """ + from unittest.mock import patch + from sqlalchemy.exc import IntegrityError + + role_name = f"concurrent-role-{uuid.uuid4().hex[:8]}" + role_service = RoleService(test_db) + + # First, create a role that will be the "winner" + winner_role = Role( + id=str(uuid.uuid4()), + name=role_name, + description="Winner role", + scope="global", + permissions=["tools.read"], + created_by="admin@example.com", + is_system_role=False, + is_active=True + ) + test_db.add(winner_role) + test_db.commit() + test_db.refresh(winner_role) + + # Track calls to get_role_by_name and simulate race condition + call_count = [0] + original_get = role_service.get_role_by_name + + async def mock_get_role(name: str, scope: str): + call_count[0] += 1 + if call_count[0] == 1: + # First call during duplicate check - return None (race window) + # Winner role exists but we're simulating the timing window + return None + # Second call after IntegrityError - return the actual winner role + return await original_get(name, scope) + + with patch.object(role_service, 'get_role_by_name', side_effect=mock_get_role): + # This will naturally trigger a REAL IntegrityError because the role already exists + # No need to mock commit - let SQLAlchemy naturally raise the error + result = await role_service.create_role( + name=role_name, + description="Loser role", + scope="global", + permissions=["tools.read"], + created_by="other@example.com" + ) + + # Should have refetched and returned the winner's role + assert result is not None + assert result.id == winner_role.id + assert result.description == "Winner role" # Winner's description, not loser's + assert call_count[0] == 2 # Called twice: initial check + refetch after IntegrityError + + +@pytest.mark.asyncio +async def test_concurrent_role_assignment_handles_integrity_error(test_db: Session, test_role: Role, test_user: EmailUser): + """Test that concurrent role assignment handles IntegrityError gracefully. + + This tests the race condition handling in assign_role_to_user() lines 671-683. + We force an IntegrityError by directly inserting a duplicate in the database + after the duplicate check but before commit. + """ + from unittest.mock import patch + + role_service = RoleService(test_db) + + # First, create the "winner" assignment directly + winner_assignment = UserRole( + id=str(uuid.uuid4()), + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-race", + granted_by="admin@example.com", + is_active=True, + granted_at=datetime.now(timezone.utc) + ) + test_db.add(winner_assignment) + test_db.commit() + test_db.refresh(winner_assignment) + + # Track calls to get_user_role_assignment and simulate race condition + call_count = [0] + original_get = role_service.get_user_role_assignment + + async def mock_get_assignment(user_email: str, role_id: str, scope: str, scope_id: str | None): + call_count[0] += 1 + if call_count[0] == 1: + # First call during duplicate check - return None (race window) + # Winner assignment exists but we're simulating the timing window + return None + # Second call after IntegrityError - return the actual winner assignment + return await original_get(user_email, role_id, scope, scope_id) + + with patch.object(role_service, 'get_user_role_assignment', side_effect=mock_get_assignment): + # This will naturally trigger a REAL IntegrityError because the assignment already exists + # No need to mock commit - let SQLAlchemy naturally raise the error + result = await role_service.assign_role_to_user( + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-race", + granted_by="other@example.com" + ) + + # Should have refetched and returned the winner's assignment + assert result is not None + assert result.id == winner_assignment.id + assert result.granted_by == "admin@example.com" # Winner's granted_by, not loser's + assert call_count[0] == 2 # Called twice: initial check + refetch after IntegrityError + + +@pytest.mark.asyncio +async def test_integrity_error_with_no_existing_role_raises_error(test_db: Session): + """Test that IntegrityError without finding existing role raises ValueError. + + This tests the error path in create_role() lines 253-254. + We simulate a scenario where: + 1. Initial check returns None (race window) + 2. Commit succeeds initially but then we manually create the role again + 3. Second attempt triggers IntegrityError + 4. Refetch returns None (role mysteriously disappeared) + 5. Should raise ValueError + """ + from unittest.mock import patch, AsyncMock + from sqlalchemy.exc import IntegrityError + + role_name = f"mystery-role-{uuid.uuid4().hex[:8]}" + role_service = RoleService(test_db) + + # First, create a role normally + initial_role = await role_service.create_role( + name=role_name, + description="Initial role", + scope="global", + permissions=["tools.read"], + created_by="admin@example.com" + ) + + # Now delete it to simulate the "mystery" scenario + test_db.delete(initial_role) + test_db.commit() + + # Create second service instance for the race scenario + role_service2 = RoleService(test_db) + + # Mock the initial check to return None + call_count = [0] + async def mock_get_role(name: str, scope: str): + call_count[0] += 1 + # Always return None to simulate the role being gone + return None + + # Mock commit to raise IntegrityError on the first call, then work normally + commit_calls = [0] + original_commit = test_db.commit + def mock_commit(): + commit_calls[0] += 1 + if commit_calls[0] == 1: + raise IntegrityError("UNIQUE constraint violated", {}, None) + return original_commit() + + with patch.object(role_service2, 'get_role_by_name', side_effect=mock_get_role): + with patch.object(test_db, 'commit', side_effect=mock_commit): + # This should trigger IntegrityError, then try to refetch but find nothing + # Should raise ValueError with "Failed to create or fetch role" message + with pytest.raises(ValueError, match="Failed to create or fetch role"): + await role_service2.create_role( + name=role_name, + description="Mystery role", + scope="global", + permissions=["tools.read"], + created_by="admin@example.com" + ) + + # Verify the error path was taken (lines 253-254) + assert call_count[0] >= 2 # Initial check + refetch attempt + + +@pytest.mark.asyncio +async def test_integrity_error_with_no_existing_assignment_raises_error(test_db: Session, test_role: Role, test_user: EmailUser): + """Test that IntegrityError without finding existing assignment raises ValueError. + + This tests the error path in assign_role_to_user() lines 682-683. + We simulate a scenario where: + 1. Initial check returns None (race window) + 2. Commit succeeds initially but then we manually create the assignment again + 3. Second attempt triggers IntegrityError + 4. Refetch returns None (assignment mysteriously disappeared) + 5. Should raise ValueError + """ + from unittest.mock import patch, AsyncMock + from sqlalchemy.exc import IntegrityError + + role_service = RoleService(test_db) + + # First, create an assignment normally + initial_assignment = await role_service.assign_role_to_user( + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-mystery", + granted_by="admin@example.com" + ) + + # Now delete it to simulate the "mystery" scenario + test_db.delete(initial_assignment) + test_db.commit() + + # Create second service instance for the race scenario + role_service2 = RoleService(test_db) + + # Mock the initial check to return None + call_count = [0] + async def mock_get_assignment(user_email: str, role_id: str, scope: str, scope_id: str | None): + call_count[0] += 1 + # Always return None to simulate the assignment being gone + return None + + # Mock commit to raise IntegrityError on the first call, then work normally + commit_calls = [0] + original_commit = test_db.commit + def mock_commit(): + commit_calls[0] += 1 + if commit_calls[0] == 1: + raise IntegrityError("UNIQUE constraint violated", {}, None) + return original_commit() + + with patch.object(role_service2, 'get_user_role_assignment', side_effect=mock_get_assignment): + with patch.object(test_db, 'commit', side_effect=mock_commit): + # This should trigger IntegrityError, then try to refetch but find nothing + # Should raise ValueError with "Failed to create or fetch role assignment" message + with pytest.raises(ValueError, match="Failed to create or fetch role assignment"): + await role_service2.assign_role_to_user( + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-mystery", + granted_by="admin@example.com" + ) + + # Verify the error path was taken (lines 682-683) + assert call_count[0] >= 2 # Initial check + refetch attempt + + +@pytest.mark.asyncio +async def test_expired_assignment_soft_delete_race(test_db: Session, test_role: Role, test_user: EmailUser): + """Test the expired assignment soft-delete race condition. + + This tests the fix for the critical race where two processes try to re-grant an expired assignment: + 1. Process A checks existing (finds expired), starts soft-delete + 2. Process B checks existing (finds expired), starts soft-delete + 3. One wins, one gets IntegrityError + 4. The loser should refetch and return the winner's fresh assignment (not the expired one) + """ + from datetime import datetime, timezone, timedelta + from unittest.mock import patch + + role_service = RoleService(test_db) + + # Create an expired active assignment + expired_date = datetime.now(timezone.utc) - timedelta(days=1) + expired_assignment = UserRole( + id=str(uuid.uuid4()), + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-expired", + granted_by="admin@example.com", + is_active=True, + granted_at=datetime.now(timezone.utc) - timedelta(days=2), + expires_at=expired_date + ) + test_db.add(expired_assignment) + test_db.commit() + test_db.refresh(expired_assignment) + + # Verify it's expired but active + assert expired_assignment.is_active is True + assert expired_assignment.is_expired() is True + + # Track soft-delete commits + commit_count = [0] + original_commit = test_db.commit + + def mock_commit(): + commit_count[0] += 1 + if commit_count[0] == 1: + # First commit (soft-delete) succeeds + return original_commit() + # Subsequent commits succeed normally + return original_commit() + + with patch.object(test_db, 'commit', side_effect=mock_commit): + # Process A tries to re-grant - should soft-delete expired and create new + result = await role_service.assign_role_to_user( + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-expired", + granted_by="other@example.com", + expires_at=datetime.now(timezone.utc) + timedelta(days=30) + ) + + # Verify the new assignment is not expired + assert result is not None + assert result.is_active is True + assert result.is_expired() is False + assert result.granted_by == "other@example.com" + assert result.id != expired_assignment.id # New assignment, not the old one + + # Verify the old assignment was soft-deleted + test_db.refresh(expired_assignment) + assert expired_assignment.is_active is False + + +@pytest.mark.asyncio +async def test_concurrent_expired_assignment_handling(test_db: Session, test_role: Role, test_user: EmailUser): + """Test that concurrent processes can handle expired assignments safely. + + This tests that when two processes try to re-grant an expired assignment concurrently, + both can proceed without error - one will soft-delete and create new, the other + will get IntegrityError and refetch the new assignment. + """ + from datetime import datetime, timezone, timedelta + + role_service = RoleService(test_db) + + # Create an expired active assignment + expired_date = datetime.now(timezone.utc) - timedelta(days=1) + expired_assignment = UserRole( + id=str(uuid.uuid4()), + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-concurrent", + granted_by="admin@example.com", + is_active=True, + granted_at=datetime.now(timezone.utc) - timedelta(days=2), + expires_at=expired_date + ) + test_db.add(expired_assignment) + test_db.commit() + test_db.refresh(expired_assignment) + + # Verify it's expired but active + assert expired_assignment.is_active is True + assert expired_assignment.is_expired() is True + + # Process tries to re-grant - should soft-delete expired and create new + new_expires = datetime.now(timezone.utc) + timedelta(days=30) + result = await role_service.assign_role_to_user( + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-concurrent", + granted_by="other@example.com", + expires_at=new_expires + ) + + # Verify the new assignment is not expired + assert result is not None + assert result.is_active is True + assert result.is_expired() is False + assert result.granted_by == "other@example.com" + assert result.id != expired_assignment.id # New assignment, not the old one + + # Verify the old assignment was soft-deleted + test_db.refresh(expired_assignment) + assert expired_assignment.is_active is False + + +@pytest.mark.asyncio +async def test_expired_assignment_soft_delete_concurrent_modification(test_db: Session, test_role: Role, test_user: EmailUser): + """Test IntegrityError during expired assignment soft-delete is handled. + + Covers role_service.py lines 659, 661-662: + - IntegrityError during soft-delete commit + - Rollback and refetch + - Return fresh assignment if exists + """ + from datetime import datetime, timezone, timedelta + from unittest.mock import patch + from sqlalchemy.exc import IntegrityError + + role_service = RoleService(test_db) + + # Create an expired active assignment + expired_date = datetime.now(timezone.utc) - timedelta(days=1) + expired_assignment = UserRole( + id=str(uuid.uuid4()), + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-soft-delete-race", + granted_by="admin@example.com", + is_active=True, + granted_at=datetime.now(timezone.utc) - timedelta(days=2), + expires_at=expired_date + ) + test_db.add(expired_assignment) + test_db.commit() + test_db.refresh(expired_assignment) + + # Verify it's expired but active + assert expired_assignment.is_active is True + assert expired_assignment.is_expired() is True + + # Mock commit to raise IntegrityError on first call (soft-delete), then succeed + commit_count = [0] + original_commit = test_db.commit + + def mock_commit(): + commit_count[0] += 1 + if commit_count[0] == 1: + # First commit (soft-delete) raises IntegrityError - simulating another process + # already modified this row + raise IntegrityError("UNIQUE constraint failed", {}, None) + # Subsequent commits succeed + return original_commit() + + # Create a fresh assignment that will be returned after the race + fresh_assignment = UserRole( + id=str(uuid.uuid4()), + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-soft-delete-race", + granted_by="winner@example.com", + is_active=True, + granted_at=datetime.now(timezone.utc), + expires_at=datetime.now(timezone.utc) + timedelta(days=30) + ) + + # Mock the refetch to return the fresh assignment + original_get = role_service.get_user_role_assignment + get_count = [0] + + async def mock_get_assignment(user_email: str, role_id: str, scope: str, scope_id: str | None): + get_count[0] += 1 + if get_count[0] == 1: + # First call: return expired assignment + return expired_assignment + # Second call after IntegrityError: return fresh assignment (winner's result) + return fresh_assignment + + with patch.object(test_db, 'commit', side_effect=mock_commit): + with patch.object(role_service, 'get_user_role_assignment', side_effect=mock_get_assignment): + # This should trigger the IntegrityError during soft-delete, rollback, and refetch + result = await role_service.assign_role_to_user( + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-soft-delete-race", + granted_by="loser@example.com" + ) + + # Verify the fresh assignment (winner's) was returned + assert result is not None + assert result.id == fresh_assignment.id + assert result.granted_by == "winner@example.com" + assert result.is_expired() is False + + +@pytest.mark.asyncio +async def test_expired_assignment_soft_delete_no_active_after_race(test_db: Session, test_role: Role, test_user: EmailUser): + """Test continuing with creation when no active assignment exists after soft-delete race. + + Covers role_service.py lines 665-666, 668, 670: + - Refetch after IntegrityError during soft-delete + - Check if fresh assignment exists + - Continue to creation if no active assignment found + """ + from datetime import datetime, timezone, timedelta + from unittest.mock import patch, MagicMock + from sqlalchemy.exc import IntegrityError + + role_service = RoleService(test_db) + + # Create an expired active assignment + expired_date = datetime.now(timezone.utc) - timedelta(days=1) + expired_assignment = UserRole( + id=str(uuid.uuid4()), + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-no-active", + granted_by="admin@example.com", + is_active=True, + granted_at=datetime.now(timezone.utc) - timedelta(days=2), + expires_at=expired_date + ) + test_db.add(expired_assignment) + test_db.commit() + test_db.refresh(expired_assignment) + + # Mock the refetch to return expired then None (another process cleaned up) + original_get = role_service.get_user_role_assignment + get_count = [0] + + async def mock_get_assignment(user_email: str, role_id: str, scope: str, scope_id: str | None): + get_count[0] += 1 + if get_count[0] == 1: + # First call: return expired assignment + return expired_assignment + elif get_count[0] == 2: + # Second call after IntegrityError: return None (no active assignment) + # But we need to actually soft-delete the expired one now + test_db.query(UserRole).filter_by(id=expired_assignment.id).update({"is_active": False}) + test_db.commit() + return None + # Any subsequent calls use the original + return await original_get(user_email, role_id, scope, scope_id) + + # Mock commit to raise IntegrityError on first call (soft-delete), then succeed + commit_count = [0] + original_commit = test_db.commit + + def mock_commit(): + commit_count[0] += 1 + if commit_count[0] == 1: + # First commit (soft-delete) raises IntegrityError + raise IntegrityError("UNIQUE constraint failed", {}, None) + # Subsequent commits succeed + return original_commit() + + with patch.object(test_db, 'commit', side_effect=mock_commit): + with patch.object(role_service, 'get_user_role_assignment', side_effect=mock_get_assignment): + # This should proceed with creation after finding no active assignment + result = await role_service.assign_role_to_user( + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-no-active", + granted_by="creator@example.com", + expires_at=datetime.now(timezone.utc) + timedelta(days=30) + ) + + # Verify a new assignment was created + assert result is not None + assert result.is_active is True + assert result.is_expired() is False + assert result.granted_by == "creator@example.com" + + +@pytest.mark.asyncio +async def test_refetched_expired_assignment_soft_delete(test_db: Session, test_role: Role, test_user: EmailUser): + """Test soft-deleting refetched expired assignments after IntegrityError. + + Covers role_service.py lines 705-707, 709: + - Check if refetched assignment is expired after IntegrityError + - Soft-delete expired assignment + - Raise ValueError to signal retry needed + """ + from datetime import datetime, timezone, timedelta + from unittest.mock import patch + from sqlalchemy.exc import IntegrityError + + role_service = RoleService(test_db) + + # Create an expired assignment that will be "created" by the winner + expired_date = datetime.now(timezone.utc) - timedelta(days=1) + expired_assignment = UserRole( + id=str(uuid.uuid4()), + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-refetch-expired", + granted_by="winner@example.com", + is_active=True, + granted_at=datetime.now(timezone.utc) - timedelta(days=2), + expires_at=expired_date + ) + + # Mock the assignment query + original_get = role_service.get_user_role_assignment + get_count = [0] + + async def mock_get_assignment(user_email: str, role_id: str, scope: str, scope_id: str | None): + get_count[0] += 1 + if get_count[0] == 1: + # First call: no existing assignment + return None + # Second call after IntegrityError: return expired assignment (winner created it but it's expired) + return expired_assignment + + # Mock commit to raise IntegrityError on first attempt + commit_count = [0] + original_commit = test_db.commit + + def mock_commit(): + commit_count[0] += 1 + if commit_count[0] == 1: + # First commit raises IntegrityError (another process created assignment) + raise IntegrityError("UNIQUE constraint failed", {}, None) + # Second commit (soft-delete) succeeds + return original_commit() + + with patch.object(role_service, 'get_user_role_assignment', side_effect=mock_get_assignment): + with patch.object(test_db, 'commit', side_effect=mock_commit): + # This should raise ValueError when refetched assignment is expired + with pytest.raises(ValueError, match="Refetched assignment.*was expired"): + await role_service.assign_role_to_user( + user_email=test_user.email, + role_id=test_role.id, + scope="team", + scope_id="team-refetch-expired", + granted_by="loser@example.com" + ) + + # Verify the expired assignment was soft-deleted + assert expired_assignment.is_active is False diff --git a/tests/unit/mcpgateway/db/test_rbac_unique_constraints_migration.py b/tests/unit/mcpgateway/db/test_rbac_unique_constraints_migration.py new file mode 100644 index 0000000000..edd2f515c9 --- /dev/null +++ b/tests/unit/mcpgateway/db/test_rbac_unique_constraints_migration.py @@ -0,0 +1,513 @@ +# -*- coding: utf-8 -*- +"""Location: ./tests/unit/mcpgateway/db/test_rbac_unique_constraints_migration.py +Copyright contributors to the MCP-CONTEXT-FORGE project +SPDX-License-Identifier: Apache-2.0 + +Unit tests for migration d21698ae4a19 (add_rbac_unique_constraints_race_fix). + +Tests verify: +- Migration module structure (import, revision chain, function signatures) +- Roles deduplication keeps the oldest role by created_at and deactivates others +- user_roles remapping: assignments pointing at a duplicate role are remapped to the kept role + *before* that duplicate role is deactivated, so list_user_roles() join on active roles still works +- user_roles deduplication prefers unexpired assignments over expired ones, and newest-granted wins + among same-expiry rows +- Partial unique index creation is idempotent (safe to run twice) +- upgrade() and downgrade() are no-ops when the tables do not exist +""" + +# Standard +import importlib +import inspect as pyinspect +from datetime import datetime, timedelta, timezone + +# Third-Party +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy import text +from sqlalchemy.pool import StaticPool + +MODULE_NAME = "mcpgateway.alembic.versions.d21698ae4a19_add_rbac_unique_constraints_race_fix" +REVISION = "d21698ae4a19" # pragma: allowlist secret +DOWN_REVISION = "b6c7d8e9f0a1" # pragma: allowlist secret + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_engine(): + """Return an in-memory SQLite engine.""" + return sa.create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool) + + +def _create_tables(conn): + """Create minimal roles and user_roles tables matching the production schema.""" + conn.execute(text(""" + CREATE TABLE roles ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + scope TEXT NOT NULL DEFAULT 'global', + is_active INTEGER NOT NULL DEFAULT 1, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ) + """)) + conn.execute(text(""" + CREATE TABLE user_roles ( + id TEXT PRIMARY KEY, + user_email TEXT NOT NULL, + role_id TEXT NOT NULL, + scope TEXT NOT NULL DEFAULT 'global', + scope_id TEXT, + is_active INTEGER NOT NULL DEFAULT 1, + granted_at TEXT NOT NULL DEFAULT (datetime('now')), + expires_at TEXT + ) + """)) + conn.commit() + + +def _run_upgrade(conn): + """Run the migration upgrade() with the given connection.""" + ctx = MigrationContext.configure(conn, opts={"as_sql": False}) + with Operations.context(ctx): + module = importlib.import_module(MODULE_NAME) + module.upgrade() + + +def _run_downgrade(conn): + """Run the migration downgrade() with the given connection.""" + ctx = MigrationContext.configure(conn, opts={"as_sql": False}) + with Operations.context(ctx): + module = importlib.import_module(MODULE_NAME) + module.downgrade() + + +def _get_table_names(conn): + inspector = sa.inspect(conn) + return set(inspector.get_table_names()) + + +def _get_index_names(conn, table): + inspector = sa.inspect(conn) + return {idx["name"] for idx in inspector.get_indexes(table)} + + +def _now_str(offset_seconds=0): + """Return an ISO datetime string offset from now by `offset_seconds`.""" + dt = datetime.now(timezone.utc) + timedelta(seconds=offset_seconds) + return dt.strftime("%Y-%m-%d %H:%M:%S") + + +# --------------------------------------------------------------------------- +# Module structure tests +# --------------------------------------------------------------------------- + + +class TestModuleStructure: + """Verify migration module metadata and function signatures.""" + + def test_module_imports(self): + """Migration module can be imported without errors.""" + module = importlib.import_module(MODULE_NAME) + assert module is not None + + def test_revision_id(self): + """Revision ID matches expected value.""" + module = importlib.import_module(MODULE_NAME) + assert module.revision == REVISION + + def test_down_revision(self): + """Down revision points to the correct parent.""" + module = importlib.import_module(MODULE_NAME) + assert module.down_revision == DOWN_REVISION + + def test_has_upgrade_function(self): + """Module has a callable upgrade() function.""" + module = importlib.import_module(MODULE_NAME) + assert callable(module.upgrade) + + def test_has_downgrade_function(self): + """Module has a callable downgrade() function.""" + module = importlib.import_module(MODULE_NAME) + assert callable(module.downgrade) + + def test_upgrade_accepts_no_params(self): + """upgrade() takes no parameters.""" + module = importlib.import_module(MODULE_NAME) + assert len(pyinspect.signature(module.upgrade).parameters) == 0 + + def test_downgrade_accepts_no_params(self): + """downgrade() takes no parameters.""" + module = importlib.import_module(MODULE_NAME) + assert len(pyinspect.signature(module.downgrade).parameters) == 0 + + +# --------------------------------------------------------------------------- +# No-op when tables are absent +# --------------------------------------------------------------------------- + + +class TestNoopWithoutTables: + """Migration is a no-op when the target tables do not exist.""" + + def test_upgrade_skips_when_tables_missing(self): + """upgrade() exits cleanly on an empty database.""" + engine = _make_engine() + try: + with engine.connect() as conn: + _run_upgrade(conn) + # No tables created by migration itself + assert "roles" not in _get_table_names(conn) + finally: + engine.dispose() + + def test_downgrade_skips_when_tables_missing(self): + """downgrade() exits cleanly on an empty database.""" + engine = _make_engine() + try: + with engine.connect() as conn: + _run_downgrade(conn) + assert "roles" not in _get_table_names(conn) + finally: + engine.dispose() + + +# --------------------------------------------------------------------------- +# Roles deduplication +# --------------------------------------------------------------------------- + + +class TestRolesDeduplications: + """Roles table deduplication keeps oldest active role by (name, scope).""" + + def test_no_duplicates_untouched(self): + """No rows are changed when there are no duplicates.""" + engine = _make_engine() + try: + with engine.connect() as conn: + _create_tables(conn) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active, created_at) VALUES ('r1','admin','global',1,'2024-01-01')")) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active, created_at) VALUES ('r2','viewer','global',1,'2024-01-02')")) + conn.commit() + + _run_upgrade(conn) + + rows = conn.execute(text("SELECT id, is_active FROM roles ORDER BY id")).fetchall() + assert len(rows) == 2 + for row in rows: + assert row[1] == 1, f"Role {row[0]} should still be active" + finally: + engine.dispose() + + def test_duplicate_roles_oldest_kept(self): + """When two active roles share (name, scope), the oldest (earliest created_at) is kept.""" + engine = _make_engine() + try: + with engine.connect() as conn: + _create_tables(conn) + # r1 is older — it should be kept + conn.execute(text("INSERT INTO roles (id, name, scope, is_active, created_at) VALUES ('r1','admin','global',1,'2024-01-01')")) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active, created_at) VALUES ('r2','admin','global',1,'2024-06-01')")) + conn.commit() + + _run_upgrade(conn) + + r1 = conn.execute(text("SELECT is_active FROM roles WHERE id='r1'")).scalar() + r2 = conn.execute(text("SELECT is_active FROM roles WHERE id='r2'")).scalar() + assert r1 == 1, "Oldest role should remain active" + assert r2 == 0, "Newer duplicate should be deactivated" + finally: + engine.dispose() + + def test_inactive_roles_not_touched(self): + """Inactive roles are not re-activated or further modified.""" + engine = _make_engine() + try: + with engine.connect() as conn: + _create_tables(conn) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active, created_at) VALUES ('r1','admin','global',1,'2024-01-01')")) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active, created_at) VALUES ('r2','admin','global',0,'2023-01-01')")) + conn.commit() + + _run_upgrade(conn) + + r1 = conn.execute(text("SELECT is_active FROM roles WHERE id='r1'")).scalar() + r2 = conn.execute(text("SELECT is_active FROM roles WHERE id='r2'")).scalar() + assert r1 == 1 + assert r2 == 0 # remains inactive + finally: + engine.dispose() + + +# --------------------------------------------------------------------------- +# user_roles remapping (review comment #1) +# --------------------------------------------------------------------------- + + +class TestUserRolesRemapping: + """Assignments pointing at a duplicate role are remapped to the kept role.""" + + def test_assignment_remapped_before_role_deactivated(self): + """user_roles.role_id is updated to the kept role id before the duplicate is deactivated.""" + engine = _make_engine() + try: + with engine.connect() as conn: + _create_tables(conn) + # Two duplicate active roles — r1 (oldest) will be kept, r2 deactivated + conn.execute(text("INSERT INTO roles (id, name, scope, is_active, created_at) VALUES ('r1','admin','global',1,'2024-01-01')")) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active, created_at) VALUES ('r2','admin','global',1,'2024-06-01')")) + # Assignment points at r2 (the duplicate that will be deactivated) + conn.execute(text( + "INSERT INTO user_roles (id, user_email, role_id, scope, is_active, granted_at) " + "VALUES ('ur1','alice@example.com','r2','global',1,'2024-06-01')" + )) + conn.commit() + + _run_upgrade(conn) + + # Assignment must now point at r1 (the kept role) + role_id = conn.execute(text("SELECT role_id FROM user_roles WHERE id='ur1'")).scalar() + assert role_id == "r1", f"Assignment should be remapped to r1 but got {role_id}" + + # r2 should be deactivated + r2_active = conn.execute(text("SELECT is_active FROM roles WHERE id='r2'")).scalar() + assert r2_active == 0 + finally: + engine.dispose() + + def test_assignment_to_kept_role_unchanged(self): + """Assignments already pointing at the kept role are not altered.""" + engine = _make_engine() + try: + with engine.connect() as conn: + _create_tables(conn) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active, created_at) VALUES ('r1','admin','global',1,'2024-01-01')")) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active, created_at) VALUES ('r2','admin','global',1,'2024-06-01')")) + # Assignment already points at r1 (the winner) + conn.execute(text( + "INSERT INTO user_roles (id, user_email, role_id, scope, is_active, granted_at) " + "VALUES ('ur1','bob@example.com','r1','global',1,'2024-01-15')" + )) + conn.commit() + + _run_upgrade(conn) + + role_id = conn.execute(text("SELECT role_id FROM user_roles WHERE id='ur1'")).scalar() + assert role_id == "r1" + finally: + engine.dispose() + + +# --------------------------------------------------------------------------- +# user_roles deduplication ordering (review comment #2) +# --------------------------------------------------------------------------- + + +class TestUserRolesDeduplicationOrdering: + """Unexpired assignments are preferred over expired ones when deduplicating user_roles.""" + + def test_unexpired_kept_over_expired(self): + """When one assignment is expired and one is not, the unexpired one is kept.""" + engine = _make_engine() + try: + with engine.connect() as conn: + _create_tables(conn) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active) VALUES ('r1','admin','global',1)")) + + past = _now_str(-3600) # expired 1 hour ago + future = _now_str(+86400) # expires in 24 hours + + # ur1: granted earlier but already expired + conn.execute(text( + "INSERT INTO user_roles (id, user_email, role_id, scope, is_active, granted_at, expires_at) " + "VALUES ('ur1','alice@example.com','r1','global',1,'2024-01-01',:past)" + ), {"past": past}) + # ur2: granted later but still valid + conn.execute(text( + "INSERT INTO user_roles (id, user_email, role_id, scope, is_active, granted_at, expires_at) " + "VALUES ('ur2','alice@example.com','r1','global',1,'2024-06-01',:future)" + ), {"future": future}) + conn.commit() + + _run_upgrade(conn) + + ur1_active = conn.execute(text("SELECT is_active FROM user_roles WHERE id='ur1'")).scalar() + ur2_active = conn.execute(text("SELECT is_active FROM user_roles WHERE id='ur2'")).scalar() + assert ur2_active == 1, "Unexpired assignment should be kept active" + assert ur1_active == 0, "Expired assignment should be deactivated as the duplicate" + finally: + engine.dispose() + + def test_null_expires_at_treated_as_unexpired(self): + """An assignment with NULL expires_at (never expires) is treated as unexpired.""" + engine = _make_engine() + try: + with engine.connect() as conn: + _create_tables(conn) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active) VALUES ('r1','admin','global',1)")) + + past = _now_str(-3600) + + # ur1: expired + conn.execute(text( + "INSERT INTO user_roles (id, user_email, role_id, scope, is_active, granted_at, expires_at) " + "VALUES ('ur1','bob@example.com','r1','global',1,'2024-01-01',:past)" + ), {"past": past}) + # ur2: never-expiring (NULL) + conn.execute(text( + "INSERT INTO user_roles (id, user_email, role_id, scope, is_active, granted_at, expires_at) " + "VALUES ('ur2','bob@example.com','r1','global',1,'2024-06-01',NULL)" + )) + conn.commit() + + _run_upgrade(conn) + + ur1_active = conn.execute(text("SELECT is_active FROM user_roles WHERE id='ur1'")).scalar() + ur2_active = conn.execute(text("SELECT is_active FROM user_roles WHERE id='ur2'")).scalar() + assert ur2_active == 1, "Never-expiring assignment should be kept" + assert ur1_active == 0, "Expired assignment should be deactivated" + finally: + engine.dispose() + + def test_among_unexpired_newest_granted_wins(self): + """Among multiple unexpired assignments, the most recently granted one is kept.""" + engine = _make_engine() + try: + with engine.connect() as conn: + _create_tables(conn) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active) VALUES ('r1','admin','global',1)")) + + future = _now_str(+86400) + + # ur1: unexpired but granted earlier + conn.execute(text( + "INSERT INTO user_roles (id, user_email, role_id, scope, is_active, granted_at, expires_at) " + "VALUES ('ur1','carol@example.com','r1','global',1,'2024-01-01',:future)" + ), {"future": future}) + # ur2: unexpired and granted more recently + conn.execute(text( + "INSERT INTO user_roles (id, user_email, role_id, scope, is_active, granted_at, expires_at) " + "VALUES ('ur2','carol@example.com','r1','global',1,'2024-09-01',:future)" + ), {"future": future}) + conn.commit() + + _run_upgrade(conn) + + ur1_active = conn.execute(text("SELECT is_active FROM user_roles WHERE id='ur1'")).scalar() + ur2_active = conn.execute(text("SELECT is_active FROM user_roles WHERE id='ur2'")).scalar() + assert ur2_active == 1, "Most recently granted unexpired assignment should be kept" + assert ur1_active == 0, "Earlier-granted assignment should be deactivated" + finally: + engine.dispose() + + def test_with_scope_id_unexpired_preferred(self): + """Prefer unexpired assignments for team-scoped (scope_id IS NOT NULL) duplicates too.""" + engine = _make_engine() + try: + with engine.connect() as conn: + _create_tables(conn) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active) VALUES ('r1','viewer','team',1)")) + + past = _now_str(-3600) + future = _now_str(+86400) + + # Expired assignment + conn.execute(text( + "INSERT INTO user_roles (id, user_email, role_id, scope, scope_id, is_active, granted_at, expires_at) " + "VALUES ('ur1','dave@example.com','r1','team','team-abc',1,'2024-01-01',:past)" + ), {"past": past}) + # Unexpired assignment + conn.execute(text( + "INSERT INTO user_roles (id, user_email, role_id, scope, scope_id, is_active, granted_at, expires_at) " + "VALUES ('ur2','dave@example.com','r1','team','team-abc',1,'2024-06-01',:future)" + ), {"future": future}) + conn.commit() + + _run_upgrade(conn) + + ur1_active = conn.execute(text("SELECT is_active FROM user_roles WHERE id='ur1'")).scalar() + ur2_active = conn.execute(text("SELECT is_active FROM user_roles WHERE id='ur2'")).scalar() + assert ur2_active == 1 + assert ur1_active == 0 + finally: + engine.dispose() + + +# --------------------------------------------------------------------------- +# Index creation and idempotency +# --------------------------------------------------------------------------- + + +class TestIndexCreation: + """Partial unique indexes are created by upgrade() and dropped by downgrade().""" + + def test_upgrade_creates_indexes(self): + """upgrade() creates the three expected partial unique indexes.""" + engine = _make_engine() + try: + with engine.connect() as conn: + _create_tables(conn) + conn.commit() + _run_upgrade(conn) + + role_indexes = _get_index_names(conn, "roles") + ur_indexes = _get_index_names(conn, "user_roles") + + assert "uq_roles_name_scope_active" in role_indexes + assert "uq_user_roles_email_role_scope_null_active" in ur_indexes + assert "uq_user_roles_email_role_scope_id_active" in ur_indexes + finally: + engine.dispose() + + def test_upgrade_idempotent(self): + """Running upgrade() twice does not raise an error.""" + engine = _make_engine() + try: + with engine.connect() as conn: + _create_tables(conn) + conn.commit() + _run_upgrade(conn) + # Second run must not raise + _run_upgrade(conn) + finally: + engine.dispose() + + def test_downgrade_drops_indexes(self): + """downgrade() removes the indexes created by upgrade().""" + engine = _make_engine() + try: + with engine.connect() as conn: + _create_tables(conn) + conn.commit() + _run_upgrade(conn) + _run_downgrade(conn) + + role_indexes = _get_index_names(conn, "roles") + ur_indexes = _get_index_names(conn, "user_roles") + + assert "uq_roles_name_scope_active" not in role_indexes + assert "uq_user_roles_email_role_scope_null_active" not in ur_indexes + assert "uq_user_roles_email_role_scope_id_active" not in ur_indexes + finally: + engine.dispose() + + def test_downgrade_does_not_reactivate_deduped_rows(self): + """Soft-deleted duplicates remain inactive after downgrade (audit trail preserved).""" + engine = _make_engine() + try: + with engine.connect() as conn: + _create_tables(conn) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active, created_at) VALUES ('r1','admin','global',1,'2024-01-01')")) + conn.execute(text("INSERT INTO roles (id, name, scope, is_active, created_at) VALUES ('r2','admin','global',1,'2024-06-01')")) + conn.commit() + + _run_upgrade(conn) + _run_downgrade(conn) + + r2_active = conn.execute(text("SELECT is_active FROM roles WHERE id='r2'")).scalar() + assert r2_active == 0, "Deduped role must stay inactive after downgrade" + finally: + engine.dispose() diff --git a/tests/unit/mcpgateway/services/test_role_service_race_conditions.py b/tests/unit/mcpgateway/services/test_role_service_race_conditions.py new file mode 100644 index 0000000000..45981837c3 --- /dev/null +++ b/tests/unit/mcpgateway/services/test_role_service_race_conditions.py @@ -0,0 +1,629 @@ +# -*- coding: utf-8 -*- +"""Location: ./tests/unit/mcpgateway/services/test_role_service_race_conditions.py +Copyright contributors to the MCP-CONTEXT-FORGE project +SPDX-License-Identifier: Apache-2.0 + +Unit tests for role service IntegrityError race condition handling. + +These tests use mocks to simulate database constraint violations without +requiring a real database with unique constraints. +""" + +import pytest +from unittest.mock import MagicMock, AsyncMock, patch +from sqlalchemy.exc import IntegrityError +import uuid + +# First-Party +from mcpgateway.services.role_service import RoleService +from mcpgateway.db import Role, UserRole + + +@pytest.mark.asyncio +async def test_create_role_success_path(): + """Test successful role creation (no IntegrityError). + + Covers role_service.py lines 231, 237, 239-240: + - Add role inside savepoint or regular add + - Commit and refresh + - Log and return created role + """ + mock_db = MagicMock() + + # Mock begin_nested to work (use context manager) + mock_savepoint = MagicMock() + mock_savepoint.__enter__ = MagicMock(return_value=mock_savepoint) + mock_savepoint.__exit__ = MagicMock(return_value=None) + mock_db.begin_nested.return_value = mock_savepoint + + role_service = RoleService(mock_db) + + # Mock get_role_by_name to return None (no duplicate) + async def mock_get_role(name: str, scope: str): + return None + role_service.get_role_by_name = mock_get_role + + # Call create_role - should succeed without IntegrityError + result = await role_service.create_role( + name="new-role", + description="New role", + scope="global", + permissions=["tools.read"], + created_by="admin@example.com" + ) + + # Verify success path + assert result is not None + assert result.name == "new-role" + + # Verify add was called inside savepoint (line 231) + mock_db.add.assert_called_once() + + # Verify commit and refresh were called (lines 236-237) + mock_db.commit.assert_called_once() + mock_db.refresh.assert_called_once() + + +@pytest.mark.asyncio +async def test_create_role_handles_integrity_error_and_refetches(): + """Test that create_role handles IntegrityError and refetches existing role. + + Covers role_service.py lines 244-250: + - Rollback after IntegrityError + - Log concurrent creation + - Refetch and return existing role + """ + mock_db = MagicMock() + mock_db.begin_nested.side_effect = AttributeError # Simulate no savepoint support + + # Create mock winner role + winner_role = Role( + id=str(uuid.uuid4()), + name="test-role", + description="Winner role", + scope="global", + permissions=["tools.read"], + created_by="admin@example.com", + is_system_role=False, + is_active=True + ) + + role_service = RoleService(mock_db) + + # Mock get_role_by_name to simulate race window then successful refetch + call_count = [0] + async def mock_get_role(name: str, scope: str): + call_count[0] += 1 + if call_count[0] == 1: + # First call during duplicate check - return None (race window) + return None + # Second call after IntegrityError - return winner role + return winner_role + + role_service.get_role_by_name = mock_get_role + + # Mock commit to raise IntegrityError + mock_db.commit.side_effect = IntegrityError("UNIQUE constraint failed", {}, None) + + # Call create_role - should catch IntegrityError and refetch + result = await role_service.create_role( + name="test-role", + description="Loser role", + scope="global", + permissions=["tools.read"], + created_by="other@example.com" + ) + + # Verify IntegrityError was handled + assert result is not None + assert result.id == winner_role.id + assert result.description == "Winner role" + assert call_count[0] == 2 # Initial check + refetch + + # Verify rollback was called (line 244) + mock_db.rollback.assert_called_once() + + +@pytest.mark.asyncio +async def test_create_role_raises_error_when_refetch_fails(): + """Test that create_role raises ValueError when refetch returns None. + + Covers role_service.py lines 253-254: + - Error logging + - Raise ValueError when role not found after IntegrityError + """ + mock_db = MagicMock() + mock_db.begin_nested.side_effect = AttributeError + + role_service = RoleService(mock_db) + + # Mock get_role_by_name to always return None + async def mock_get_role(name: str, scope: str): + return None + + role_service.get_role_by_name = mock_get_role + + # Mock commit to raise IntegrityError + mock_db.commit.side_effect = IntegrityError("UNIQUE constraint failed", {}, None) + + # Call create_role - should raise ValueError + with pytest.raises(ValueError, match="Failed to create or fetch role"): + await role_service.create_role( + name="mystery-role", + description="Mystery role", + scope="global", + permissions=["tools.read"], + created_by="admin@example.com" + ) + + # Verify rollback was called + mock_db.rollback.assert_called_once() + + +@pytest.mark.asyncio +async def test_assign_role_success_path(): + """Test successful role assignment (no IntegrityError). + + Covers role_service.py lines 660, 666, 668-669: + - Add user_role inside savepoint or regular add + - Commit and refresh + - Log and return assignment + """ + mock_db = MagicMock() + + # Mock begin_nested to work + mock_savepoint = MagicMock() + mock_savepoint.__enter__ = MagicMock(return_value=mock_savepoint) + mock_savepoint.__exit__ = MagicMock(return_value=None) + mock_db.begin_nested.return_value = mock_savepoint + + mock_role = Role( + id=str(uuid.uuid4()), + name="test-role", + description="Test role", + scope="team", + permissions=["tools.read"], + created_by="admin@example.com", + is_system_role=False, + is_active=True + ) + + role_service = RoleService(mock_db) + + # Mock get_role_by_id + async def mock_get_role_by_id(role_id: str): + return mock_role + role_service.get_role_by_id = mock_get_role_by_id + + # Mock get_user_role_assignment to return None (no existing assignment) + async def mock_get_assignment(user_email: str, role_id: str, scope: str, scope_id: str): + return None + role_service.get_user_role_assignment = mock_get_assignment + + # Call assign_role_to_user - should succeed + result = await role_service.assign_role_to_user( + user_email="user@example.com", + role_id=mock_role.id, + scope="team", + scope_id="team-123", + granted_by="admin@example.com" + ) + + # Verify success path + assert result is not None + assert result.user_email == "user@example.com" + + # Verify add was called inside savepoint (line 660) + mock_db.add.assert_called_once() + + # Verify commit and refresh were called (lines 665-666) + mock_db.commit.assert_called_once() + mock_db.refresh.assert_called_once() + + +@pytest.mark.asyncio +async def test_assign_role_handles_integrity_error_and_refetches(): + """Test that assign_role_to_user handles IntegrityError and refetches existing assignment. + + Covers role_service.py lines 673-679: + - Rollback after IntegrityError + - Log concurrent assignment + - Refetch and return existing assignment + """ + mock_db = MagicMock() + mock_db.begin_nested.side_effect = AttributeError + + # Create mock role and assignment + mock_role = Role( + id=str(uuid.uuid4()), + name="test-role", + description="Test role", + scope="team", + permissions=["tools.read"], + created_by="admin@example.com", + is_system_role=False, + is_active=True + ) + + winner_assignment = UserRole( + id=str(uuid.uuid4()), + user_email="user@example.com", + role_id=mock_role.id, + scope="team", + scope_id="team-123", + granted_by="admin@example.com", + is_active=True + ) + + role_service = RoleService(mock_db) + + # Mock get_role_by_id to return the role + async def mock_get_role_by_id(role_id: str): + return mock_role + role_service.get_role_by_id = mock_get_role_by_id + + # Mock get_user_role_assignment to simulate race window then successful refetch + call_count = [0] + async def mock_get_assignment(user_email: str, role_id: str, scope: str, scope_id: str): + call_count[0] += 1 + if call_count[0] == 1: + # First call during duplicate check - return None (race window) + return None + # Second call after IntegrityError - return winner assignment + return winner_assignment + + role_service.get_user_role_assignment = mock_get_assignment + + # Mock commit to raise IntegrityError + mock_db.commit.side_effect = IntegrityError("UNIQUE constraint failed", {}, None) + + # Call assign_role_to_user - should catch IntegrityError and refetch + result = await role_service.assign_role_to_user( + user_email="user@example.com", + role_id=mock_role.id, + scope="team", + scope_id="team-123", + granted_by="other@example.com" + ) + + # Verify IntegrityError was handled + assert result is not None + assert result.id == winner_assignment.id + assert result.granted_by == "admin@example.com" + assert call_count[0] == 2 # Initial check + refetch + + # Verify rollback was called (line 673) + mock_db.rollback.assert_called_once() + + +@pytest.mark.asyncio +async def test_assign_role_raises_error_when_refetch_fails(): + """Test that assign_role_to_user raises ValueError when refetch returns None. + + Covers role_service.py lines 682-683: + - Error logging + - Raise ValueError when assignment not found after IntegrityError + """ + mock_db = MagicMock() + mock_db.begin_nested.side_effect = AttributeError + + mock_role = Role( + id=str(uuid.uuid4()), + name="test-role", + description="Test role", + scope="team", + permissions=["tools.read"], + created_by="admin@example.com", + is_system_role=False, + is_active=True + ) + + role_service = RoleService(mock_db) + + # Mock get_role_by_id + async def mock_get_role_by_id(role_id: str): + return mock_role + role_service.get_role_by_id = mock_get_role_by_id + + # Mock get_user_role_assignment to always return None + async def mock_get_assignment(user_email: str, role_id: str, scope: str, scope_id: str): + return None + + role_service.get_user_role_assignment = mock_get_assignment + + # Mock commit to raise IntegrityError + mock_db.commit.side_effect = IntegrityError("UNIQUE constraint failed", {}, None) + + # Call assign_role_to_user - should raise ValueError + with pytest.raises(ValueError, match="Failed to create or fetch role assignment"): + await role_service.assign_role_to_user( + user_email="user@example.com", + role_id=mock_role.id, + scope="team", + scope_id="team-mystery", + granted_by="admin@example.com" + ) + + # Verify rollback was called + mock_db.rollback.assert_called_once() + + +@pytest.mark.asyncio +async def test_assign_role_handles_expired_assignment_soft_delete(): + """Test that assign_role_to_user soft-deletes expired assignments during IntegrityError handling. + + Covers role_service.py lines 659, 661-662: + - Catch IntegrityError during expired assignment soft-delete + - Rollback and log concurrent modification + - Refetch to see current state + """ + from datetime import datetime, timedelta, timezone + + mock_db = MagicMock() + mock_db.begin_nested.side_effect = AttributeError + + # Create mock role + mock_role = Role( + id=str(uuid.uuid4()), + name="test-role", + description="Test role", + scope="team", + permissions=["tools.read"], + created_by="admin@example.com", + is_system_role=False, + is_active=True + ) + + # Create expired assignment that will trigger soft-delete path + expired_assignment = UserRole( + id=str(uuid.uuid4()), + user_email="user@example.com", + role_id=mock_role.id, + scope="team", + scope_id="team-123", + granted_by="admin@example.com", + is_active=True, + expires_at=datetime.now(timezone.utc) - timedelta(days=1) # Expired + ) + + # Create a fresh assignment that will be returned after race condition + fresh_assignment = UserRole( + id=str(uuid.uuid4()), + user_email="user@example.com", + role_id=mock_role.id, + scope="team", + scope_id="team-123", + granted_by="other@example.com", + is_active=True, + expires_at=None + ) + + role_service = RoleService(mock_db) + + # Mock get_role_by_id + async def mock_get_role_by_id(role_id: str): + return mock_role + role_service.get_role_by_id = mock_get_role_by_id + + # Mock get_user_role_assignment to return expired assignment first, then fresh + call_count = [0] + async def mock_get_assignment(user_email: str, role_id: str, scope: str, scope_id: str): + call_count[0] += 1 + if call_count[0] == 1: + # First call: return expired assignment + return expired_assignment + elif call_count[0] == 2: + # Second call after soft-delete race: return fresh assignment + return fresh_assignment + return None + + role_service.get_user_role_assignment = mock_get_assignment + + # Mock commit to raise IntegrityError on first commit (during soft-delete) + commit_count = [0] + def mock_commit(): + commit_count[0] += 1 + if commit_count[0] == 1: + # First commit (soft-delete) raises IntegrityError + raise IntegrityError("UNIQUE constraint failed", {}, None) + # Second commit succeeds (no-op for testing) + pass + + mock_db.commit = mock_commit + + # Call assign_role_to_user - should handle expired assignment soft-delete race + result = await role_service.assign_role_to_user( + user_email="user@example.com", + role_id=mock_role.id, + scope="team", + scope_id="team-123", + granted_by="admin@example.com" + ) + + # Verify that the fresh assignment was returned + assert result is not None + assert result.id == fresh_assignment.id + assert result.granted_by == "other@example.com" + + # Verify rollback was called after IntegrityError (line 661) + mock_db.rollback.assert_called() + + +@pytest.mark.asyncio +async def test_assign_role_handles_expired_assignment_no_active_after_race(): + """Test that assign_role_to_user continues with creation when no active assignment exists after race. + + Covers role_service.py lines 665-666, 668, 670: + - Refetch after concurrent modification during expired soft-delete + - Check if fresh assignment exists + - Continue to creation if no active assignment + """ + from datetime import datetime, timedelta, timezone + + mock_db = MagicMock() + + # Mock begin_nested to work for final creation + mock_savepoint = MagicMock() + mock_savepoint.__enter__ = MagicMock(return_value=mock_savepoint) + mock_savepoint.__exit__ = MagicMock(return_value=None) + mock_db.begin_nested.return_value = mock_savepoint + + # Create mock role + mock_role = Role( + id=str(uuid.uuid4()), + name="test-role", + description="Test role", + scope="team", + permissions=["tools.read"], + created_by="admin@example.com", + is_system_role=False, + is_active=True + ) + + # Create expired assignment + expired_assignment = UserRole( + id=str(uuid.uuid4()), + user_email="user@example.com", + role_id=mock_role.id, + scope="team", + scope_id="team-123", + granted_by="admin@example.com", + is_active=True, + expires_at=datetime.now(timezone.utc) - timedelta(days=1) + ) + + role_service = RoleService(mock_db) + + # Mock get_role_by_id + async def mock_get_role_by_id(role_id: str): + return mock_role + role_service.get_role_by_id = mock_get_role_by_id + + # Mock get_user_role_assignment to return expired, then None after race + call_count = [0] + async def mock_get_assignment(user_email: str, role_id: str, scope: str, scope_id: str): + call_count[0] += 1 + if call_count[0] == 1: + # First call: return expired assignment + return expired_assignment + # Second call after soft-delete race: return None (no active assignment) + return None + + role_service.get_user_role_assignment = mock_get_assignment + + # Mock commit to raise IntegrityError on first commit, succeed on second + commit_count = [0] + def mock_commit(): + commit_count[0] += 1 + if commit_count[0] == 1: + # First commit (soft-delete) raises IntegrityError + raise IntegrityError("UNIQUE constraint failed", {}, None) + # Second commit succeeds + pass + + mock_db.commit = mock_commit + + # Call assign_role_to_user - should proceed with creation after race + result = await role_service.assign_role_to_user( + user_email="user@example.com", + role_id=mock_role.id, + scope="team", + scope_id="team-123", + granted_by="admin@example.com" + ) + + # Verify that a new assignment was created + assert result is not None + assert result.user_email == "user@example.com" + + # Verify rollback was called after first IntegrityError + mock_db.rollback.assert_called() + + # Verify add was called (line 682) for the new assignment + mock_db.add.assert_called() + + +@pytest.mark.asyncio +async def test_assign_role_handles_refetched_expired_assignment(): + """Test that assign_role_to_user soft-deletes refetched expired assignments after IntegrityError. + + Covers role_service.py lines 705-707, 709: + - Check if refetched assignment is expired after IntegrityError + - Soft-delete expired assignment + - Raise ValueError to signal retry needed + """ + from datetime import datetime, timedelta, timezone + + mock_db = MagicMock() + mock_db.begin_nested.side_effect = AttributeError + + # Create mock role + mock_role = Role( + id=str(uuid.uuid4()), + name="test-role", + description="Test role", + scope="team", + permissions=["tools.read"], + created_by="admin@example.com", + is_system_role=False, + is_active=True + ) + + # Create expired assignment that will be refetched after IntegrityError + expired_assignment = UserRole( + id=str(uuid.uuid4()), + user_email="user@example.com", + role_id=mock_role.id, + scope="team", + scope_id="team-123", + granted_by="admin@example.com", + is_active=True, + expires_at=datetime.now(timezone.utc) - timedelta(days=1) + ) + + role_service = RoleService(mock_db) + + # Mock get_role_by_id + async def mock_get_role_by_id(role_id: str): + return mock_role + role_service.get_role_by_id = mock_get_role_by_id + + # Mock get_user_role_assignment + call_count = [0] + async def mock_get_assignment(user_email: str, role_id: str, scope: str, scope_id: str): + call_count[0] += 1 + if call_count[0] == 1: + # First call: return None (no existing assignment) + return None + # Second call after IntegrityError: return expired assignment + return expired_assignment + + role_service.get_user_role_assignment = mock_get_assignment + + # Mock commit to raise IntegrityError on first attempt + commit_count = [0] + def mock_commit(): + commit_count[0] += 1 + if commit_count[0] == 1: + # First commit raises IntegrityError + raise IntegrityError("UNIQUE constraint failed", {}, None) + # Second commit (soft-delete) succeeds + pass + + mock_db.commit = mock_commit + + # Call assign_role_to_user - should raise ValueError for expired refetched assignment + with pytest.raises(ValueError, match="Refetched assignment.*was expired"): + await role_service.assign_role_to_user( + user_email="user@example.com", + role_id=mock_role.id, + scope="team", + scope_id="team-123", + granted_by="admin@example.com" + ) + + # Verify that expired assignment was soft-deleted + assert expired_assignment.is_active is False + + # Verify commits happened (first attempt + soft-delete) + assert commit_count[0] == 2