iam-ape: resolve principal-based deny conditions (ARN/account/tag) (CIEM-685)

iam-ape/iam_ape/evaluator.py

@@ -24,6 +24,140 @@
 
 logger = logging.getLogger("IAM-APE:evaluator")
 
+_LIKE_OPERATORS = {"ArnLike", "StringLike"}
+_EQUALS_OPERATORS = {"ArnEquals", "StringEquals"}
+_NOTLIKE_OPERATORS = {"ArnNotLike", "StringNotLike"}
+_NOTEQUALS_OPERATORS = {"ArnNotEquals", "StringNotEquals"}
+_MODELED_OPERATORS = (
+    _LIKE_OPERATORS | _EQUALS_OPERATORS | _NOTLIKE_OPERATORS | _NOTEQUALS_OPERATORS
+)
+_PRINCIPAL_ARN_KEY = "aws:principalarn"
+_PRINCIPAL_ACCOUNT_KEY = "aws:principalaccount"
+_PRINCIPAL_TAG_PREFIX = "aws:principaltag/"
+_UNKNOWN = object()
+
+
+def build_principal_context(arn: str, entity: Dict[str, Any]) -> Dict[str, Any]:
+    """Principal facts known statically, used to resolve deny conditions."""
+    parts = arn.split(":")
+    return {
+        _PRINCIPAL_ARN_KEY: arn,
+        _PRINCIPAL_ACCOUNT_KEY: parts[4] if len(parts) > 4 and parts[4] else None,
+        "tags": {t.get("Key"): t.get("Value") for t in entity.get("Tags", [])},
+    }
+
+
+def _principal_condition_value(context: Dict[str, Any], key: str) -> Any:
+    key_lower = key.lower()
+    if key_lower == _PRINCIPAL_ARN_KEY:
+        return context[_PRINCIPAL_ARN_KEY]
+    if key_lower == _PRINCIPAL_ACCOUNT_KEY:
+        return context[_PRINCIPAL_ACCOUNT_KEY] or _UNKNOWN
+    if key_lower.startswith(_PRINCIPAL_TAG_PREFIX):
+        return context["tags"].get(key.split("/", 1)[1], _UNKNOWN)
+    return _UNKNOWN
+
+
+def condition_is_principal_resolvable(condition: Dict[str, Any]) -> bool:
+    """True if every clause is a modeled operator over a principal fact we know."""
+    for operator, key_values in condition.items():
+        if operator not in _MODELED_OPERATORS or not isinstance(key_values, dict):
+            return False
+        for key in key_values:
+            key_lower = key.lower()
+            if key_lower in (_PRINCIPAL_ARN_KEY, _PRINCIPAL_ACCOUNT_KEY):
+                continue
+            if key_lower.startswith(_PRINCIPAL_TAG_PREFIX):
+                continue
+            return False
+    return True
+
+
+def _clause_satisfied(operator: str, value: str, patterns: List[str]) -> bool:
+    if operator in _LIKE_OPERATORS:
+        return any(wildcard_match(value, p) for p in patterns)
+    if operator in _EQUALS_OPERATORS:
+        return value in patterns
+    if operator in _NOTLIKE_OPERATORS:
+        return not any(wildcard_match(value, p) for p in patterns)
+    return value not in patterns
+
+
+def _operator_group(
+    operator: str, key_values: Dict[str, Any], context: Dict[str, Any]
+) -> Optional[bool]:
+    result = True
+    for key, values in key_values.items():
+        value = _principal_condition_value(context, key)
+        if value is _UNKNOWN:
+            return None
+        patterns = list(values) if isinstance(values, (list, tuple)) else [values]
+        result = result and _clause_satisfied(operator, value, patterns)
+    return result
+
+
+def resolve_principal_condition(
+    condition: Optional[Dict[str, Any]], context: Dict[str, Any]
+) -> Optional[bool]:
+    """Resolve a whole Condition against known principal facts (ARN, account, tags).
+    Returns True (applies), False (doesn't), or None if a clause can't be decided
+    statically (e.g. an absent tag) -> caller keeps symbolic handling."""
+    if not condition:
+        return None
+    result = True
+    for operator, key_values in condition.items():
+        if operator not in _MODELED_OPERATORS or not isinstance(key_values, dict):
+            return None
+        group = _operator_group(operator, key_values, context)
+        if group is None:
+            return None
+        result = result and group
+    return result
+
+
+def _split_scp_statements(
+    scp_policies: List[PolicyWithSource],
+) -> Tuple[List[PolicyWithSource], List[Tuple[str, Any]]]:
+    """Partition SCPs into a static baseline (expanded once) and principal-
+    conditional Deny statements (resolved per-principal at evaluation time)."""
+    static: List[PolicyWithSource] = []
+    conditional: List[Tuple[str, Any]] = []
+    for policy in scp_policies:
+        statements = policy.policy.get("Statement", [])
+        if isinstance(statements, dict):
+            statements = [statements]
+        static_statements = []
+        for statement in statements:
+            condition = statement.get("Condition")
+            if (
+                statement.get("Effect") == "Deny"
+                and condition
+                and condition_is_principal_resolvable(condition)
+            ):
+                conditional.append((policy.source, statement))
+            else:
+                static_statements.append(statement)
+        if static_statements:
+            static.append(
+                PolicyWithSource(
+                    policy.source, {**policy.policy, "Statement": static_statements}
+                )
+            )
+    return static, conditional
+
+
+def _merge_containers(
+    base: PermissionsContainer, extra: PermissionsContainer
+) -> PermissionsContainer:
+    allowed: Dict[str, Set[Action]] = defaultdict(set)
+    denied: Dict[str, Set[Action]] = defaultdict(set)
+    for container in (base, extra):
+        for key, values in container.allowed_permissions.items():
+            allowed[key] |= set(values)
+        for key, values in container.denied_permissions.items():
+            denied[key] |= set(values)
+    return PermissionsContainer(allowed_permissions=allowed, denied_permissions=denied)
+
 
 def should_deny(
     iam_action: Action, denied_actions: Dict[str, Set[Action]]
@@ -84,9 +218,11 @@ def should_deny(
                     res.add(
                         Action(
                             action=iam_action.action,
-                            resource=None
-                            if iam_action.resource == PolicyElement.WILDCARD
-                            else iam_action.resource,
+                            resource=(
+                                None
+                                if iam_action.resource == PolicyElement.WILDCARD
+                                else iam_action.resource
+                            ),
                             not_resource=denied_action.resource,
                             condition=merge_condition(
                                 iam_action.condition, denied_action.condition
@@ -99,9 +235,11 @@ def should_deny(
                     res.add(
                         Action(
                             action=iam_action.action,
-                            resource=None
-                            if iam_action.resource == PolicyElement.WILDCARD
-                            else iam_action.resource,
+                            resource=(
+                                None
+                                if iam_action.resource == PolicyElement.WILDCARD
+                                else iam_action.resource
+                            ),
                             not_resource=denied_action.resource,
                             condition=merge_condition(
                                 iam_action.condition, denied_action.condition
@@ -490,12 +628,43 @@ def __init__(
     ) -> None:
         self.auth_details = authorization_details
         self.policy_expander = policy_expander or PolicyExpander()
+        static_scp, self._scp_principal_denies = _split_scp_statements(
+            scp_policies or []
+        )
         self.scp_policy = (
-            self.policy_expander.expand_policies(scp_policies)
-            if scp_policies
+            self.policy_expander.expand_policies(static_scp)
+            if static_scp
             else PermissionsContainer()
         )
 
+    def _scp_for_principal(
+        self, principal_context: Dict[str, Any]
+    ) -> PermissionsContainer:
+        """SCP for one principal: the static baseline plus its principal-
+        conditional denies, resolved against the principal (applies -> enforced,
+        doesn't -> dropped, undecidable -> kept symbolic)."""
+        if not self._scp_principal_denies:
+            return self.scp_policy
+        resolved: List[PolicyWithSource] = []
+        for source, statement in self._scp_principal_denies:
+            applies = resolve_principal_condition(
+                statement.get("Condition"), principal_context
+            )
+            if applies is False:
+                continue
+            if applies is True:
+                statement = {k: v for k, v in statement.items() if k != "Condition"}
+            resolved.append(
+                PolicyWithSource(
+                    source, {"Version": "2012-10-17", "Statement": [statement]}
+                )
+            )
+        if not resolved:
+            return self.scp_policy
+        return _merge_containers(
+            self.scp_policy, self.policy_expander.expand_policies(resolved)
+        )
+
     def create_json_report(
         self, permissions_container: PermissionsContainer
     ) -> FinalReportT:
@@ -677,6 +846,7 @@ def evaluate(self, arn: str, entity_type: EntityType) -> PermissionsContainer:
             logger.error(f"Error - couldn't find entity with ARN {arn}")
             raise EntityNotFoundException(arn)
 
+        scp_policy = self._scp_for_principal(build_principal_context(arn, entity_obj))
         direct_policies = self.get_direct_policies(entity_obj, entity_type)
         indirect_policies: List[PolicyWithSource] = []
         if entity_type == EntityType.user:
@@ -695,7 +865,7 @@ def evaluate(self, arn: str, entity_type: EntityType) -> PermissionsContainer:
         final_permissions, ineffective_permissions = explicitly_deny(direct_permissions)
 
         denied_permissions = direct_permissions.denied_permissions
-        for boundary in (permission_boundary, self.scp_policy):
+        for boundary in (permission_boundary, scp_policy):
             if boundary.allowed_permissions or boundary.denied_permissions:
                 (
                     final_permissions,

iam-ape/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "iam-ape"
-version = "1.1.7"
+version = "1.1.8"
 description = "IAM AWS Permissions Evaluator"
 authors = ["Tohar Braun, Orca Security <tohar@orca.security>"]
 license = "GPL-3.0-or-later"

iam-ape/tests/test_principal_conditions.py

@@ -0,0 +1,146 @@
+from typing import Any
+
+from iam_ape.evaluator import AuthorizationDetails, EffectivePolicyEvaluator
+from iam_ape.helper_classes import PolicyWithSource
+from iam_ape.helper_types import EntityType
+
+_GUARDRAIL_ACTIONS = [
+    "bedrock:CreateGuardrail",
+    "bedrock:UpdateGuardrail",
+    "bedrock:DeleteGuardrail",
+]
+_SSO_ADMIN_PATTERN = "arn:aws:iam::*:role/aws-reserved/sso.amazonaws.com/*/AWSReservedSSO_AdministratorAccess*"
+
+
+def _policy(*statements: Any) -> Any:
+    return {"Version": "2012-10-17", "Statement": list(statements)}
+
+
+_ADMIN = _policy({"Effect": "Allow", "Action": ["*"], "Resource": ["*"]})
+_FULL_AWS_ACCESS = _policy({"Effect": "Allow", "Action": ["*"], "Resource": ["*"]})
+
+
+def _scp(condition: Any = None) -> Any:
+    statement: Any = {"Effect": "Deny", "Action": _GUARDRAIL_ACTIONS, "Resource": ["*"]}
+    if condition:
+        statement["Condition"] = condition
+    return _policy(statement)
+
+
+def _evaluate(arn: str, scp: Any, tags: Any = None) -> Any:
+    user = {
+        "UserName": "u",
+        "Arn": arn,
+        "UserId": "AIDAEXAMPLE",
+        "Path": "/",
+        "UserPolicyList": [{"PolicyName": "inline-admin", "PolicyDocument": _ADMIN}],
+        "AttachedManagedPolicies": [],
+        "GroupList": [],
+        "Tags": [{"Key": k, "Value": v} for k, v in (tags or {}).items()],
+    }
+    auth = AuthorizationDetails(
+        {
+            "UserDetailList": [user],
+            "GroupDetailList": [],
+            "RoleDetailList": [],
+            "Policies": [],
+        }
+    )
+    scps = [
+        PolicyWithSource("p-FullAWSAccess", _FULL_AWS_ACCESS),
+        PolicyWithSource("p-scp", scp),
+    ]
+    return EffectivePolicyEvaluator(auth, scps).evaluate(
+        arn=arn, entity_type=EntityType.user
+    )
+
+
+def _allows_guardrail(res) -> bool:
+    return "bedrock:CreateGuardrail" in res.allowed_permissions
+
+
+def test_unconditional_deny_removes_action() -> None:
+    assert not _allows_guardrail(
+        _evaluate("arn:aws:iam::111122223333:user/App", _scp())
+    )
+
+
+def test_arn_condition_applies_to_non_exempt_principal() -> None:
+    cond = {"ArnNotLike": {"aws:PrincipalARN": [_SSO_ADMIN_PATTERN]}}
+    assert not _allows_guardrail(
+        _evaluate("arn:aws:iam::111122223333:user/App", _scp(cond))
+    )
+
+
+def test_arn_condition_skipped_for_exempt_principal() -> None:
+    cond = {"ArnNotLike": {"aws:PrincipalARN": [_SSO_ADMIN_PATTERN]}}
+    exempt = (
+        "arn:aws:iam::111122223333:role/aws-reserved/sso.amazonaws.com/"
+        "us-east-1/AWSReservedSSO_AdministratorAccess_abc/admin"
+    )
+    assert _allows_guardrail(_evaluate(exempt, _scp(cond)))
+
+
+def test_account_condition_applies_to_non_allowed_account() -> None:
+    cond = {"StringNotEquals": {"aws:PrincipalAccount": ["999988887777"]}}
+    assert not _allows_guardrail(
+        _evaluate("arn:aws:iam::111122223333:user/App", _scp(cond))
+    )
+
+
+def test_account_condition_skipped_for_allowed_account() -> None:
+    cond = {"StringNotEquals": {"aws:PrincipalAccount": ["111122223333"]}}
+    assert _allows_guardrail(
+        _evaluate("arn:aws:iam::111122223333:user/App", _scp(cond))
+    )
+
+
+def test_mixed_arn_and_account_condition_and_semantics() -> None:
+    # Real p-gh487uoz shape: deny unless allowed account OR exempt SSO role.
+    cond = {
+        "StringNotEquals": {"aws:PrincipalAccount": ["999988887777"]},
+        "ArnNotLike": {"aws:PrincipalARN": [_SSO_ADMIN_PATTERN]},
+    }
+    # non-allowed account AND non-exempt role -> deny applies -> removed
+    assert not _allows_guardrail(
+        _evaluate("arn:aws:iam::111122223333:user/App", _scp(cond))
+    )
+    # allowed account -> account clause false -> deny does NOT apply -> kept
+    assert _allows_guardrail(
+        _evaluate("arn:aws:iam::999988887777:user/App", _scp(cond))
+    )
+
+
+def test_principal_tag_condition_resolved() -> None:
+    cond = {"StringNotEquals": {"aws:PrincipalTag/team": ["ai-platform"]}}
+    denied = _evaluate("arn:aws:iam::111122223333:user/App", _scp(cond), {"team": "x"})
+    kept = _evaluate(
+        "arn:aws:iam::111122223333:user/App", _scp(cond), {"team": "ai-platform"}
+    )
+    assert not _allows_guardrail(denied)
+    assert _allows_guardrail(kept)
+
+
+def test_absent_tag_stays_symbolic() -> None:
+    cond = {"StringNotEquals": {"aws:PrincipalTag/team": ["ai-platform"]}}
+    # principal has no such tag -> undecidable -> kept symbolic (alert still fires)
+    assert _allows_guardrail(
+        _evaluate("arn:aws:iam::111122223333:user/App", _scp(cond))
+    )
+
+
+def test_non_principal_condition_stays_symbolic() -> None:
+    cond = {"StringNotEquals": {"aws:RequestedRegion": ["us-east-1"]}}
+    assert _allows_guardrail(
+        _evaluate("arn:aws:iam::111122223333:user/App", _scp(cond))
+    )
+
+
+def test_mixed_with_unresolvable_key_stays_symbolic() -> None:
+    cond = {
+        "ArnNotLike": {"aws:PrincipalARN": [_SSO_ADMIN_PATTERN]},
+        "StringNotEquals": {"aws:RequestedRegion": ["us-east-1"]},
+    }
+    assert _allows_guardrail(
+        _evaluate("arn:aws:iam::111122223333:user/App", _scp(cond))
+    )