"""AWS driver: Owner tag fallback + SSO→IAM auto-fallback. Two regressions caught in production (2026-04-25 audit): 1. ``_iam_user_email`` only matched the ``email`` tag (lowercased key). Production AWS SG uses ``Owner`` tag almost universally — so dozens of bare-username IAM users (``li.da``, ``breezewish``, ``mem9``) had no email resolution, dropping their grants. 2. ``list_scope_grants`` defaulted to SSO mode when ``account_mode`` was unset. Production AWS SG has no Identity Center configured so ``list_permission_sets_provisioned_to_account`` returned 0 → 0 grants. The driver should fall back to IAM enumeration when SSO config is incomplete. Both fixes are covered here. """ from __future__ import annotations from unittest.mock import MagicMock, patch import pytest from app.integrations.driver_base import CloudScopeRef from app.integrations.drivers.aws_cloud import AwsCloudDriver _CFG = { "access_key_id": "AKIA-test", "secret_access_key": "test-sk", "default_region": "ap-southeast-1", } def _driver(extra: dict | None = None) -> AwsCloudDriver: cfg = dict(_CFG) if extra: cfg.update(extra) return AwsCloudDriver(integration_id="aws-test", config=cfg) # ── _iam_user_email: tag-resolution priority ───────────────────── def test_email_resolved_from_username_when_at_present() -> None: """The most common case: ``shuai.feng@pingcap.com`` is itself the UserName. We don't need to call AWS for tags.""" d = _driver() iam = MagicMock() assert d._iam_user_email(iam, "Shuai.Feng@pingcap.com") == "shuai.feng@pingcap.com" iam.list_user_tags.assert_not_called() def test_email_resolved_from_owner_tag_capital_o() -> None: """Production AWS SG truth: bare username + ``Owner`` tag with email value. Pre-fix, ``_iam_user_email`` only matched lowercase ``email`` and returned None — silently dropping the grant.""" d = _driver() iam = MagicMock() iam.list_user_tags.return_value = { "Tags": [{"Key": "Owner", "Value": "li.da@pingcap.com"}], } assert d._iam_user_email(iam, "li.da") == "li.da@pingcap.com" def test_email_tag_takes_precedence_over_owner() -> None: """When both ``email`` and ``Owner`` are set, ``email`` wins — it's the explicit/canonical key, ``Owner`` is the legacy fallback.""" d = _driver() iam = MagicMock() iam.list_user_tags.return_value = { "Tags": [ {"Key": "Owner", "Value": "fallback@x.com"}, {"Key": "email", "Value": "primary@x.com"}, ], } assert d._iam_user_email(iam, "u") == "primary@x.com" def test_owner_tag_with_non_email_value_is_skipped() -> None: """``Owner: BedrockAPIKey-d6l8`` — the AWS Bedrock automation pattern. Tag is present, but the value isn't an email. We must NOT synthesize a fake email from it (the resulting grant would resolve to nobody and be dropped anyway, but worse: it creates confusing 'unresolved' counts).""" d = _driver() iam = MagicMock() iam.list_user_tags.return_value = { "Tags": [{"Key": "Owner", "Value": "BedrockAPIKey-d6l8"}], } assert d._iam_user_email(iam, "svc-key") is None def test_email_resolution_lowercases_value() -> None: """Helix work_email is stored lowercase; the email-to-person resolution is done case-insensitively. Lowercase early so everything downstream sees a consistent key.""" d = _driver() iam = MagicMock() iam.list_user_tags.return_value = { "Tags": [{"Key": "Owner", "Value": "Hao.Chen@PingCAP.com"}], } assert d._iam_user_email(iam, "hao.chen") == "hao.chen@pingcap.com" def test_no_tags_returns_none() -> None: d = _driver() iam = MagicMock() iam.list_user_tags.return_value = {"Tags": []} assert d._iam_user_email(iam, "no-email") is None def test_list_user_tags_error_returns_none() -> None: """Per-user tag fetch can fail with throttling or transient 5xx; we treat that as 'no email' (skipping that user) rather than propagating the error and breaking the whole IAM enumeration.""" from botocore.exceptions import ClientError d = _driver() iam = MagicMock() iam.list_user_tags.side_effect = ClientError( {"Error": {"Code": "Throttling", "Message": "rate limit"}}, "ListUserTags", ) assert d._iam_user_email(iam, "username") is None # ── list_scope_grants: SSO→IAM auto-fallback ──────────────────── @pytest.mark.asyncio async def test_list_scope_grants_uses_iam_when_account_mode_iam() -> None: """When admin explicitly sets ``account_mode='iam'``, we go straight to IAM enumeration without checking SSO config. Locks in that the explicit choice is honored.""" d = _driver({"account_mode": "iam"}) scope = CloudScopeRef( scope_type="aws_account", external_id="123456789012", display_name="test", ) with patch.object( d, "_list_scope_grants_iam_sync", return_value=[], ) as iam_mock, \ patch.object( d, "_list_scope_grants_sso_sync", return_value=[], ) as sso_mock: await d.list_scope_grants(scope) iam_mock.assert_called_once() sso_mock.assert_not_called() @pytest.mark.asyncio async def test_list_scope_grants_falls_back_to_iam_when_sso_unconfigured() -> None: """The actual prod AWS SG case: ``account_mode`` defaults to ``"sso"`` but the integration row has no ``sso_instance_arn`` / ``sso_identity_store_id``. The OLD behavior was to call SSO, which immediately returned ``[]`` (silently). The NEW behavior is to detect the missing config and route to IAM enumeration so the 48 IAM users in prod can produce grants.""" d = _driver() # defaults: no account_mode override, no SSO fields scope = CloudScopeRef( scope_type="aws_account", external_id="123456789012", display_name="test", ) with patch.object( d, "_list_scope_grants_iam_sync", return_value=[], ) as iam_mock, \ patch.object( d, "_list_scope_grants_sso_sync", return_value=[], ) as sso_mock: await d.list_scope_grants(scope) # IAM was called, SSO was NOT — even though account_mode='sso'. iam_mock.assert_called_once() sso_mock.assert_not_called() @pytest.mark.asyncio async def test_list_scope_grants_uses_sso_when_fully_configured() -> None: """When SSO is properly configured, we go through the SSO path — that's where we get the per-user permission-set assignments. The fallback only kicks in when config is missing.""" d = _driver({ "account_mode": "sso", "sso_instance_arn": "arn:aws:sso:::instance/ssoins-12345", "sso_identity_store_id": "d-1234567890", "sso_region": "us-east-1", }) scope = CloudScopeRef( scope_type="aws_account", external_id="123456789012", display_name="test", ) with patch.object( d, "_list_scope_grants_iam_sync", return_value=[], ) as iam_mock, \ patch.object( d, "_list_scope_grants_sso_sync", return_value=[], ) as sso_mock: await d.list_scope_grants(scope) sso_mock.assert_called_once() iam_mock.assert_not_called() @pytest.mark.asyncio async def test_list_scope_grants_falls_back_when_partial_sso_config() -> None: """Half-configured SSO (instance ARN set, identity store missing) is the same UX as no SSO — the SSO call would fail with ``ResourceNotFound`` mid-flight. Pre-empt by routing to IAM.""" d = _driver({ "account_mode": "sso", "sso_instance_arn": "arn:aws:sso:::instance/ssoins-12345", # ``sso_identity_store_id`` deliberately missing. }) scope = CloudScopeRef( scope_type="aws_account", external_id="123456789012", display_name="test", ) with patch.object( d, "_list_scope_grants_iam_sync", return_value=[], ) as iam_mock, \ patch.object( d, "_list_scope_grants_sso_sync", return_value=[], ) as sso_mock: await d.list_scope_grants(scope) iam_mock.assert_called_once() sso_mock.assert_not_called()