"""PR2 — governance gating at request / owner-add (service layer). The four-tier classifier is wired into ``submit_access_request`` and ``owner_add_member`` so the rule holds for HTTP *and* MCP (both go through the service). These tests pin the gate behaviour on a real GCP scope (scope_type=gcp_project, provider=gcp) where the tiers are known: * ``roles/billing.viewer`` → ungrantable (billing-account-level) * ``roles/owner`` / ``roles/editor`` → restricted * ``roles/iam.serviceAccountUser`` → elevated * ``roles/viewer`` → normal """ from __future__ import annotations import uuid import pytest from sqlalchemy.ext.asyncio import AsyncSession # ensure the gcp driver registers in the registry import app.integrations.drivers.gcp_cloud # noqa: F401 from app.core.exceptions import AppException from app.domains.cmdb.lifecycle_service import CloudLifecycleService from app.domains.cmdb.models import ( SCOPE_STATUS_ACTIVE, CmdbCloudRole, CmdbCloudScope, ) from app.identity.person_models import IdentityPerson from app.integrations.models import Integration async def _person(db: AsyncSession, email: str) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), work_email=email, display_name=email.split("@")[0], employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", ) db.add(p) await db.commit() return p async def _gcp_scope(db: AsyncSession, owner: IdentityPerson) -> CmdbCloudScope: integ = Integration( id=uuid.uuid4(), kind="cloud", provider="gcp", name="gcp-gov-test", slug=f"gcp-gov-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False, capabilities=["servers"], ) db.add(integ) await db.commit() scope = CmdbCloudScope( integration_id=str(integ.id), scope_type="gcp_project", external_id=f"proj-{uuid.uuid4().hex[:6]}", display_name="gov-test", owner_person_id=str(owner.id), status=SCOPE_STATUS_ACTIVE, ) db.add(scope) await db.commit() return scope async def _submit(svc, scope, person, role_key, roles): return await svc.submit_access_request( scope_id=str(scope.id), person_id=str(person.id), role_key=role_key, role_display_name=role_key, reason="t", ttl_days=30, requester_id=person.id, trace_id="t", actor_role_names=roles, ) @pytest.mark.asyncio async def test_submit_ungrantable_rejected_422_for_everyone(db_session): owner = await _person(db_session, "owner1@pingcap.com") emp = await _person(db_session, "emp1@pingcap.com") scope = await _gcp_scope(db_session, owner) svc = CloudLifecycleService(db_session) # employee with pytest.raises(AppException) as ei: await _submit(svc, scope, emp, "roles/billing.viewer", []) assert ei.value.status_code == 422 assert ei.value.code == "CLOUD_ROLE_UNGRANTABLE" # even a cloud admin can't bind it at this scope with pytest.raises(AppException) as ei2: await _submit(svc, scope, emp, "roles/billing.viewer", ["cloud_admin"]) assert ei2.value.status_code == 422 @pytest.mark.asyncio async def test_submit_restricted_rejected_403_for_employee(db_session): owner = await _person(db_session, "owner2@pingcap.com") emp = await _person(db_session, "emp2@pingcap.com") scope = await _gcp_scope(db_session, owner) svc = CloudLifecycleService(db_session) with pytest.raises(AppException) as ei: await _submit(svc, scope, emp, "roles/owner", []) assert ei.value.status_code == 403 assert ei.value.code == "CLOUD_ROLE_RESTRICTED" @pytest.mark.asyncio async def test_submit_restricted_allowed_for_cloud_admin(db_session): owner = await _person(db_session, "owner3@pingcap.com") admin = await _person(db_session, "admin3@pingcap.com") scope = await _gcp_scope(db_session, owner) svc = CloudLifecycleService(db_session) grant = await _submit(svc, scope, admin, "roles/editor", ["cloud_admin"]) assert grant is not None assert grant.role_key == "roles/editor" @pytest.mark.asyncio async def test_submit_elevated_allowed_for_employee_pending(db_session): owner = await _person(db_session, "owner4@pingcap.com") emp = await _person(db_session, "emp4@pingcap.com") scope = await _gcp_scope(db_session, owner) svc = CloudLifecycleService(db_session) # elevated is self-requestable (approval gating happens later) grant = await _submit(svc, scope, emp, "roles/iam.serviceAccountUser", []) assert grant is not None @pytest.mark.asyncio async def test_submit_normal_allowed_for_employee(db_session): owner = await _person(db_session, "owner5@pingcap.com") emp = await _person(db_session, "emp5@pingcap.com") scope = await _gcp_scope(db_session, owner) svc = CloudLifecycleService(db_session) grant = await _submit(svc, scope, emp, "roles/viewer", []) assert grant is not None async def _override_tier(db, scope, role_key, tier): db.add(CmdbCloudRole( id=uuid.uuid4(), integration_id=str(scope.integration_id), scope_id=str(scope.id), role_key=role_key, display_name=role_key, is_custom=False, is_promoted=True, governance_tier=tier, governance_overridden_by_admin=True, )) await db.commit() @pytest.mark.asyncio async def test_admin_override_to_restricted_is_enforced_by_gate(db_session): # An admin locks down roles/viewer (normally NORMAL) to restricted via # the catalog. The submit gate must now honor it — the override is not # cosmetic. owner = await _person(db_session, "ovr1@pingcap.com") emp = await _person(db_session, "ove1@pingcap.com") scope = await _gcp_scope(db_session, owner) await _override_tier(db_session, scope, "roles/viewer", "restricted") svc = CloudLifecycleService(db_session) with pytest.raises(AppException) as ei: await _submit(svc, scope, emp, "roles/viewer", []) assert ei.value.status_code == 403 assert ei.value.code == "CLOUD_ROLE_RESTRICTED" @pytest.mark.asyncio async def test_admin_override_cannot_loosen_ungrantable(db_session): # An admin tries to override billing.viewer (ungrantable on a project) # to normal. The ungrantable floor wins — submit still 422. owner = await _person(db_session, "ovr2@pingcap.com") emp = await _person(db_session, "ove2@pingcap.com") scope = await _gcp_scope(db_session, owner) await _override_tier(db_session, scope, "roles/billing.viewer", "normal") svc = CloudLifecycleService(db_session) with pytest.raises(AppException) as ei: await _submit(svc, scope, emp, "roles/billing.viewer", []) assert ei.value.status_code == 422 assert ei.value.code == "CLOUD_ROLE_UNGRANTABLE" @pytest.mark.asyncio async def test_owner_add_elevated_blocked_for_non_admin_owner(db_session): owner = await _person(db_session, "owner6@pingcap.com") member = await _person(db_session, "member6@pingcap.com") scope = await _gcp_scope(db_session, owner) svc = CloudLifecycleService(db_session) with pytest.raises(AppException) as ei: await svc.owner_add_member( scope_id=str(scope.id), person_id=str(member.id), role_key="roles/editor", role_display_name="Editor", ttl_days=30, actor_user_id=owner.id, trace_id="t", actor_role_names=[], # plain employee owner ) assert ei.value.status_code == 403 assert ei.value.code == "CLOUD_ROLE_NEEDS_ADMIN" @pytest.mark.asyncio async def test_billing_viewer_requestable_on_billing_account_scope(db_session): # The whole point of PR5: roles/billing.viewer is ungrantable on a # gcp_project (→422) but a NORMAL, self-requestable role on a # gcp_billing_account scope. An employee can request it and it pends. owner = await _person(db_session, "bo1@pingcap.com") emp = await _person(db_session, "be1@pingcap.com") integ = Integration( id=uuid.uuid4(), kind="cloud", provider="gcp", name="gcp-ba-test", slug=f"gcp-ba-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False, capabilities=["servers"], ) db_session.add(integ) await db_session.commit() scope = CmdbCloudScope( integration_id=str(integ.id), scope_type="gcp_billing_account", external_id="billingAccounts/AAAA-BBBB-CCCC", display_name="Corp Billing", owner_person_id=str(owner.id), status=SCOPE_STATUS_ACTIVE, ) db_session.add(scope) await db_session.commit() svc = CloudLifecycleService(db_session) grant = await _submit(svc, scope, emp, "roles/billing.viewer", []) assert grant is not None assert grant.role_key == "roles/billing.viewer" @pytest.mark.asyncio async def test_owner_add_ungrantable_blocked_even_for_admin(db_session): owner = await _person(db_session, "owner7@pingcap.com") member = await _person(db_session, "member7@pingcap.com") scope = await _gcp_scope(db_session, owner) svc = CloudLifecycleService(db_session) with pytest.raises(AppException) as ei: await svc.owner_add_member( scope_id=str(scope.id), person_id=str(member.id), role_key="roles/billing.viewer", role_display_name="Billing Viewer", ttl_days=30, actor_user_id=owner.id, trace_id="t", actor_role_names=["cloud_admin"], ) assert ei.value.status_code == 422 assert ei.value.code == "CLOUD_ROLE_UNGRANTABLE" @pytest.mark.asyncio async def test_owner_add_member_blocks_leaver(db_session): """#1647: a scope owner (or the MCP dual) must not grant cloud IAM to a leaving/left colleague. deleted_at is still NULL for a leaver_pending/ left person so the None/deleted check misses them, and the UI's leaver filter is display-only — the service is the real boundary.""" owner = await _person(db_session, "lg-owner@pingcap.com") scope = await _gcp_scope(db_session, owner) svc = CloudLifecycleService(db_session) for status in ("left", "leaver_pending"): leaver = IdentityPerson( id=uuid.uuid4(), work_email=f"{status}-lg@pingcap.com", display_name=f"{status} member", employment_status=status, source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", ) db_session.add(leaver) await db_session.commit() with pytest.raises(AppException) as ei: await svc.owner_add_member( scope_id=str(scope.id), person_id=str(leaver.id), role_key="roles/viewer", role_display_name="Viewer", ttl_days=30, actor_user_id=owner.id, trace_id="t", actor_role_names=["global_admin"], ) assert ei.value.status_code == 409 assert ei.value.code == "PERSON_NOT_ELIGIBLE_LEAVER"