From c4a9732d6ecbae0fd4bc1b64c87864d2be4db887 Mon Sep 17 00:00:00 2001 From: Yana Sidanych Date: Wed, 8 Jul 2026 12:13:54 +0300 Subject: [PATCH 1/4] iam-ape: resolve aws:PrincipalARN-only conditions on denies (CIEM-685) Effective-permissions evaluation now resolves deny-statement conditions that depend solely on aws:PrincipalARN against the evaluated principal, instead of carrying them symbolically. A provable deny becomes unconditional (action removed); a provably inapplicable one is skipped; conditions referencing any other key are left untouched (prior behavior). Bump 1.1.7 -> 1.1.8; add tests. --- iam-ape/iam_ape/evaluator.py | 81 ++++++++++++++++--- iam-ape/pyproject.toml | 2 +- .../tests/test_principal_arn_conditions.py | 81 +++++++++++++++++++ 3 files changed, 152 insertions(+), 12 deletions(-) create mode 100644 iam-ape/tests/test_principal_arn_conditions.py diff --git a/iam-ape/iam_ape/evaluator.py b/iam-ape/iam_ape/evaluator.py index 93daa54..e756d0f 100644 --- a/iam-ape/iam_ape/evaluator.py +++ b/iam-ape/iam_ape/evaluator.py @@ -24,14 +24,53 @@ logger = logging.getLogger("IAM-APE:evaluator") +_PRINCIPAL_ARN_KEY = "aws:principalarn" +_ARN_LIKE_OPERATORS = {"ArnLike", "StringLike"} +_ARN_EQUALS_OPERATORS = {"ArnEquals", "StringEquals"} +_ARN_NOTLIKE_OPERATORS = {"ArnNotLike", "StringNotLike"} +_ARN_NOTEQUALS_OPERATORS = {"ArnNotEquals", "StringNotEquals"} + + +def resolve_principal_arn_condition( + condition: Optional[Dict[str, Any]], principal_arn: str +) -> Optional[bool]: + """Resolve a Condition keyed only on aws:PrincipalARN against the evaluated + principal. Returns True (applies), False (doesn't), or None (can't resolve + statically -> caller keeps symbolic handling).""" + if not condition: + return None + result = True + for operator, key_values in condition.items(): + if not isinstance(key_values, dict): + return None + for key, values in key_values.items(): + if key.lower() != _PRINCIPAL_ARN_KEY: + return None + patterns = list(values) if isinstance(values, (list, tuple)) else [values] + if operator in _ARN_LIKE_OPERATORS: + satisfied = any(wildcard_match(principal_arn, p) for p in patterns) + elif operator in _ARN_EQUALS_OPERATORS: + satisfied = principal_arn in patterns + elif operator in _ARN_NOTLIKE_OPERATORS: + satisfied = not any(wildcard_match(principal_arn, p) for p in patterns) + elif operator in _ARN_NOTEQUALS_OPERATORS: + satisfied = principal_arn not in patterns + else: + return None + result = result and satisfied + return result + def should_deny( - iam_action: Action, denied_actions: Dict[str, Set[Action]] + iam_action: Action, + denied_actions: Dict[str, Set[Action]], + principal_arn: Optional[str] = None, ) -> Tuple[bool, Set[Action], Optional[str]]: """ Check if an action is denied by a list of denied actions :param iam_action: :param denied_actions: + :param principal_arn: enables resolving aws:PrincipalARN-only deny conditions :return: denied, partially_denied_actions, source """ res = set() @@ -39,6 +78,17 @@ def should_deny( for denied_action in denied_actions.get(iam_action.action, []): + # A provable aws:PrincipalARN deny becomes unconditional; a provably + # inapplicable one is skipped; anything else keeps its condition. + if principal_arn is not None and denied_action.condition: + applies = resolve_principal_arn_condition( + denied_action.condition, principal_arn + ) + if applies is False: + continue + if applies is True: + denied_action = replace(denied_action, condition=None) + if iam_action == denied_action: return True, set(), denied_action.source @@ -84,9 +134,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 +151,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 @@ -227,13 +281,14 @@ def should_deny( def explicitly_deny( permissions: PermissionsContainer, + principal_arn: Optional[str] = None, ) -> Tuple[Dict[str, Set[Action]], Set[IneffectiveAction]]: final_actions_dict: Dict[str, Set[Action]] = defaultdict(set) ineffective_permissions: Set[IneffectiveAction] = set() for action_key, action_values in permissions.allowed_permissions.items(): for action_value in action_values: denied, new_action_values, denied_by = should_deny( - action_value, permissions.denied_permissions + action_value, permissions.denied_permissions, principal_arn ) if not denied: final_actions_dict[action_key].update(new_action_values) @@ -255,6 +310,7 @@ def explicitly_deny( def apply_permission_boundary( allow_actions: Dict[str, Set[Action]], permission_boundary: PermissionsContainer, + principal_arn: Optional[str] = None, ) -> Tuple[Dict[str, Set[Action]], Set[IneffectiveAction]]: def permit(at: Action, bt: Action) -> Action: return replace( @@ -455,7 +511,8 @@ def get_pb_id() -> str: PermissionsContainer( allowed_permissions=new_allow_actions, denied_permissions=permission_boundary.denied_permissions, - ) + ), + principal_arn, ) ineffective_permissions.update(denied_ineffective) return allowed, ineffective_permissions @@ -692,7 +749,9 @@ def evaluate(self, arn: str, entity_type: EntityType) -> PermissionsContainer: ) permission_boundary = self.get_permission_boundary(entity_obj) - final_permissions, ineffective_permissions = explicitly_deny(direct_permissions) + final_permissions, ineffective_permissions = explicitly_deny( + direct_permissions, arn + ) denied_permissions = direct_permissions.denied_permissions for boundary in (permission_boundary, self.scp_policy): @@ -700,7 +759,7 @@ def evaluate(self, arn: str, entity_type: EntityType) -> PermissionsContainer: ( final_permissions, more_ineffective_permissions, - ) = apply_permission_boundary(final_permissions, boundary) + ) = apply_permission_boundary(final_permissions, boundary, arn) ineffective_permissions.update(more_ineffective_permissions) denied_permissions = deep_update( diff --git a/iam-ape/pyproject.toml b/iam-ape/pyproject.toml index 138ba13..befba7c 100644 --- a/iam-ape/pyproject.toml +++ b/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 "] license = "GPL-3.0-or-later" diff --git a/iam-ape/tests/test_principal_arn_conditions.py b/iam-ape/tests/test_principal_arn_conditions.py new file mode 100644 index 0000000..d7144d2 --- /dev/null +++ b/iam-ape/tests/test_principal_arn_conditions.py @@ -0,0 +1,81 @@ +from iam_ape.evaluator import AuthorizationDetails, EffectivePolicyEvaluator +from iam_ape.helper_classes import PolicyWithSource +from iam_ape.helper_types import AwsPolicyType, EntityType + +_ADMIN: AwsPolicyType = { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": "*", "Resource": "*"}], +} +_FULL_AWS_ACCESS: AwsPolicyType = { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": "*", "Resource": "*"}], +} +_GUARDRAIL_ACTIONS = [ + "bedrock:CreateGuardrail", + "bedrock:UpdateGuardrail", + "bedrock:DeleteGuardrail", +] +_SSO_ADMIN_PATTERN = "arn:aws:iam::*:role/aws-reserved/sso.amazonaws.com/*/AWSReservedSSO_AdministratorAccess*" + + +def _scp(condition=None) -> AwsPolicyType: + statement = {"Effect": "Deny", "Action": _GUARDRAIL_ACTIONS, "Resource": ["*"]} + if condition: + statement["Condition"] = condition + return {"Version": "2012-10-17", "Statement": [statement]} + + +def _evaluate(arn: str, scp: AwsPolicyType): + auth = AuthorizationDetails( + { + "UserDetailList": [ + { + "UserName": "u", + "Arn": arn, + "UserId": "AIDAEXAMPLE", + "Path": "/", + "UserPolicyList": [ + {"PolicyName": "inline-admin", "PolicyDocument": _ADMIN} + ], + "AttachedManagedPolicies": [], + "GroupList": [], + } + ], + "GroupDetailList": [], + "RoleDetailList": [], + "Policies": [], + } + ) + scp_policies = [ + PolicyWithSource("p-FullAWSAccess", _FULL_AWS_ACCESS), + PolicyWithSource("p-scp", scp), + ] + evaluator = EffectivePolicyEvaluator(auth, scp_policies) + return evaluator.evaluate(arn=arn, entity_type=EntityType.user) + + +def test_unconditional_deny_removes_action() -> None: + res = _evaluate("arn:aws:iam::111122223333:role/AppRole", _scp()) + assert "bedrock:CreateGuardrail" not in res.allowed_permissions + + +def test_conditional_deny_applies_to_non_exempt_principal() -> None: + condition = {"ArnNotLike": {"aws:PrincipalARN": [_SSO_ADMIN_PATTERN]}} + res = _evaluate("arn:aws:iam::111122223333:role/AppRole", _scp(condition)) + assert "bedrock:CreateGuardrail" not in res.allowed_permissions + + +def test_conditional_deny_skipped_for_exempt_principal() -> None: + condition = {"ArnNotLike": {"aws:PrincipalARN": [_SSO_ADMIN_PATTERN]}} + exempt = ( + "arn:aws:iam::111122223333:role/aws-reserved/sso.amazonaws.com/" + "us-east-1/AWSReservedSSO_AdministratorAccess_abc/admin" + ) + res = _evaluate(exempt, _scp(condition)) + assert "bedrock:CreateGuardrail" in res.allowed_permissions + + +def test_non_principal_condition_stays_symbolic() -> None: + condition = {"StringNotEquals": {"aws:RequestedRegion": ["us-east-1"]}} + res = _evaluate("arn:aws:iam::111122223333:role/AppRole", _scp(condition)) + assert "bedrock:CreateGuardrail" in res.allowed_permissions From af85c196277092a9ac3eb89c820ea466f6cd10f8 Mon Sep 17 00:00:00 2001 From: Yana Sidanych Date: Wed, 8 Jul 2026 15:17:04 +0300 Subject: [PATCH 2/4] iam-ape: resolve principal conditions before expansion (ARN + account + tag) Widen deny-condition resolution to aws:PrincipalARN, aws:PrincipalAccount and aws:PrincipalTag/*, and resolve the whole Condition BEFORE the policy expander splits multi-operator blocks into independent entries -- so AND semantics are preserved (a mixed account+arn deny like the real p-gh487uoz resolves correctly). SCPs are partitioned in __init__: the static baseline (incl. FullAWSAccess *) is expanded once; only principal-conditional Deny statements are resolved+expanded per principal, so no per-principal re-expansion of *. should_deny/boundary hot path is unchanged. Undecidable clauses (region/MFA/tags-absent/...) stay symbolic. --- iam-ape/iam_ape/evaluator.py | 192 +++++++++++++----- .../tests/test_principal_arn_conditions.py | 81 -------- iam-ape/tests/test_principal_conditions.py | 144 +++++++++++++ 3 files changed, 288 insertions(+), 129 deletions(-) delete mode 100644 iam-ape/tests/test_principal_arn_conditions.py create mode 100644 iam-ape/tests/test_principal_conditions.py diff --git a/iam-ape/iam_ape/evaluator.py b/iam-ape/iam_ape/evaluator.py index e756d0f..272ac8e 100644 --- a/iam-ape/iam_ape/evaluator.py +++ b/iam-ape/iam_ape/evaluator.py @@ -24,53 +24,133 @@ logger = logging.getLogger("IAM-APE:evaluator") -_PRINCIPAL_ARN_KEY = "aws:principalarn" -_ARN_LIKE_OPERATORS = {"ArnLike", "StringLike"} -_ARN_EQUALS_OPERATORS = {"ArnEquals", "StringEquals"} -_ARN_NOTLIKE_OPERATORS = {"ArnNotLike", "StringNotLike"} -_ARN_NOTEQUALS_OPERATORS = {"ArnNotEquals", "StringNotEquals"} +_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_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 { + "aws:principalarn": arn, + "aws:principalaccount": 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 == "aws:principalarn": + return context["aws:principalarn"] + if key_lower == "aws:principalaccount": + return context["aws:principalaccount"] or _UNKNOWN + if key_lower.startswith(_PRINCIPAL_TAG_PREFIX): + return context["tags"].get(key.split("/", 1)[1], _UNKNOWN) + return _UNKNOWN -def resolve_principal_arn_condition( - condition: Optional[Dict[str, Any]], principal_arn: str +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 ("aws:principalarn", "aws:principalaccount"): + continue + if key_lower.startswith(_PRINCIPAL_TAG_PREFIX): + continue + return False + return True + + +def resolve_principal_condition( + condition: Optional[Dict[str, Any]], context: Dict[str, Any] ) -> Optional[bool]: - """Resolve a Condition keyed only on aws:PrincipalARN against the evaluated - principal. Returns True (applies), False (doesn't), or None (can't resolve - statically -> caller keeps symbolic handling).""" + """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 not isinstance(key_values, dict): + if operator not in _MODELED_OPERATORS or not isinstance(key_values, dict): return None for key, values in key_values.items(): - if key.lower() != _PRINCIPAL_ARN_KEY: + value = _principal_condition_value(context, key) + if value is _UNKNOWN: return None patterns = list(values) if isinstance(values, (list, tuple)) else [values] - if operator in _ARN_LIKE_OPERATORS: - satisfied = any(wildcard_match(principal_arn, p) for p in patterns) - elif operator in _ARN_EQUALS_OPERATORS: - satisfied = principal_arn in patterns - elif operator in _ARN_NOTLIKE_OPERATORS: - satisfied = not any(wildcard_match(principal_arn, p) for p in patterns) - elif operator in _ARN_NOTEQUALS_OPERATORS: - satisfied = principal_arn not in patterns + if operator in _LIKE_OPERATORS: + satisfied = any(wildcard_match(value, p) for p in patterns) + elif operator in _EQUALS_OPERATORS: + satisfied = value in patterns + elif operator in _NOTLIKE_OPERATORS: + satisfied = not any(wildcard_match(value, p) for p in patterns) else: - return None + satisfied = value not in patterns result = result and satisfied 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]], - principal_arn: Optional[str] = None, + iam_action: Action, denied_actions: Dict[str, Set[Action]] ) -> Tuple[bool, Set[Action], Optional[str]]: """ Check if an action is denied by a list of denied actions :param iam_action: :param denied_actions: - :param principal_arn: enables resolving aws:PrincipalARN-only deny conditions :return: denied, partially_denied_actions, source """ res = set() @@ -78,17 +158,6 @@ def should_deny( for denied_action in denied_actions.get(iam_action.action, []): - # A provable aws:PrincipalARN deny becomes unconditional; a provably - # inapplicable one is skipped; anything else keeps its condition. - if principal_arn is not None and denied_action.condition: - applies = resolve_principal_arn_condition( - denied_action.condition, principal_arn - ) - if applies is False: - continue - if applies is True: - denied_action = replace(denied_action, condition=None) - if iam_action == denied_action: return True, set(), denied_action.source @@ -281,14 +350,13 @@ def should_deny( def explicitly_deny( permissions: PermissionsContainer, - principal_arn: Optional[str] = None, ) -> Tuple[Dict[str, Set[Action]], Set[IneffectiveAction]]: final_actions_dict: Dict[str, Set[Action]] = defaultdict(set) ineffective_permissions: Set[IneffectiveAction] = set() for action_key, action_values in permissions.allowed_permissions.items(): for action_value in action_values: denied, new_action_values, denied_by = should_deny( - action_value, permissions.denied_permissions, principal_arn + action_value, permissions.denied_permissions ) if not denied: final_actions_dict[action_key].update(new_action_values) @@ -310,7 +378,6 @@ def explicitly_deny( def apply_permission_boundary( allow_actions: Dict[str, Set[Action]], permission_boundary: PermissionsContainer, - principal_arn: Optional[str] = None, ) -> Tuple[Dict[str, Set[Action]], Set[IneffectiveAction]]: def permit(at: Action, bt: Action) -> Action: return replace( @@ -511,8 +578,7 @@ def get_pb_id() -> str: PermissionsContainer( allowed_permissions=new_allow_actions, denied_permissions=permission_boundary.denied_permissions, - ), - principal_arn, + ) ) ineffective_permissions.update(denied_ineffective) return allowed, ineffective_permissions @@ -547,12 +613,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: @@ -734,6 +831,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: @@ -749,17 +847,15 @@ def evaluate(self, arn: str, entity_type: EntityType) -> PermissionsContainer: ) permission_boundary = self.get_permission_boundary(entity_obj) - final_permissions, ineffective_permissions = explicitly_deny( - direct_permissions, arn - ) + 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, more_ineffective_permissions, - ) = apply_permission_boundary(final_permissions, boundary, arn) + ) = apply_permission_boundary(final_permissions, boundary) ineffective_permissions.update(more_ineffective_permissions) denied_permissions = deep_update( diff --git a/iam-ape/tests/test_principal_arn_conditions.py b/iam-ape/tests/test_principal_arn_conditions.py deleted file mode 100644 index d7144d2..0000000 --- a/iam-ape/tests/test_principal_arn_conditions.py +++ /dev/null @@ -1,81 +0,0 @@ -from iam_ape.evaluator import AuthorizationDetails, EffectivePolicyEvaluator -from iam_ape.helper_classes import PolicyWithSource -from iam_ape.helper_types import AwsPolicyType, EntityType - -_ADMIN: AwsPolicyType = { - "Version": "2012-10-17", - "Statement": [{"Effect": "Allow", "Action": "*", "Resource": "*"}], -} -_FULL_AWS_ACCESS: AwsPolicyType = { - "Version": "2012-10-17", - "Statement": [{"Effect": "Allow", "Action": "*", "Resource": "*"}], -} -_GUARDRAIL_ACTIONS = [ - "bedrock:CreateGuardrail", - "bedrock:UpdateGuardrail", - "bedrock:DeleteGuardrail", -] -_SSO_ADMIN_PATTERN = "arn:aws:iam::*:role/aws-reserved/sso.amazonaws.com/*/AWSReservedSSO_AdministratorAccess*" - - -def _scp(condition=None) -> AwsPolicyType: - statement = {"Effect": "Deny", "Action": _GUARDRAIL_ACTIONS, "Resource": ["*"]} - if condition: - statement["Condition"] = condition - return {"Version": "2012-10-17", "Statement": [statement]} - - -def _evaluate(arn: str, scp: AwsPolicyType): - auth = AuthorizationDetails( - { - "UserDetailList": [ - { - "UserName": "u", - "Arn": arn, - "UserId": "AIDAEXAMPLE", - "Path": "/", - "UserPolicyList": [ - {"PolicyName": "inline-admin", "PolicyDocument": _ADMIN} - ], - "AttachedManagedPolicies": [], - "GroupList": [], - } - ], - "GroupDetailList": [], - "RoleDetailList": [], - "Policies": [], - } - ) - scp_policies = [ - PolicyWithSource("p-FullAWSAccess", _FULL_AWS_ACCESS), - PolicyWithSource("p-scp", scp), - ] - evaluator = EffectivePolicyEvaluator(auth, scp_policies) - return evaluator.evaluate(arn=arn, entity_type=EntityType.user) - - -def test_unconditional_deny_removes_action() -> None: - res = _evaluate("arn:aws:iam::111122223333:role/AppRole", _scp()) - assert "bedrock:CreateGuardrail" not in res.allowed_permissions - - -def test_conditional_deny_applies_to_non_exempt_principal() -> None: - condition = {"ArnNotLike": {"aws:PrincipalARN": [_SSO_ADMIN_PATTERN]}} - res = _evaluate("arn:aws:iam::111122223333:role/AppRole", _scp(condition)) - assert "bedrock:CreateGuardrail" not in res.allowed_permissions - - -def test_conditional_deny_skipped_for_exempt_principal() -> None: - condition = {"ArnNotLike": {"aws:PrincipalARN": [_SSO_ADMIN_PATTERN]}} - exempt = ( - "arn:aws:iam::111122223333:role/aws-reserved/sso.amazonaws.com/" - "us-east-1/AWSReservedSSO_AdministratorAccess_abc/admin" - ) - res = _evaluate(exempt, _scp(condition)) - assert "bedrock:CreateGuardrail" in res.allowed_permissions - - -def test_non_principal_condition_stays_symbolic() -> None: - condition = {"StringNotEquals": {"aws:RequestedRegion": ["us-east-1"]}} - res = _evaluate("arn:aws:iam::111122223333:role/AppRole", _scp(condition)) - assert "bedrock:CreateGuardrail" in res.allowed_permissions diff --git a/iam-ape/tests/test_principal_conditions.py b/iam-ape/tests/test_principal_conditions.py new file mode 100644 index 0000000..335f182 --- /dev/null +++ b/iam-ape/tests/test_principal_conditions.py @@ -0,0 +1,144 @@ +from iam_ape.evaluator import AuthorizationDetails, EffectivePolicyEvaluator +from iam_ape.helper_classes import PolicyWithSource +from iam_ape.helper_types import AwsPolicyType, EntityType + +_ADMIN: AwsPolicyType = { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": "*", "Resource": "*"}], +} +_FULL_AWS_ACCESS: AwsPolicyType = { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": "*", "Resource": "*"}], +} +_GUARDRAIL_ACTIONS = [ + "bedrock:CreateGuardrail", + "bedrock:UpdateGuardrail", + "bedrock:DeleteGuardrail", +] +_SSO_ADMIN_PATTERN = "arn:aws:iam::*:role/aws-reserved/sso.amazonaws.com/*/AWSReservedSSO_AdministratorAccess*" + + +def _scp(condition=None) -> AwsPolicyType: + statement = {"Effect": "Deny", "Action": _GUARDRAIL_ACTIONS, "Resource": ["*"]} + if condition: + statement["Condition"] = condition + return {"Version": "2012-10-17", "Statement": [statement]} + + +def _evaluate(arn, scp, tags=None): + 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)) + ) From 3d88a9c9502bb566ae5f4cb50baaf441d24bd722 Mon Sep 17 00:00:00 2001 From: Yana Sidanych Date: Wed, 8 Jul 2026 17:16:47 +0300 Subject: [PATCH 3/4] iam-ape: fix mypy in tests (avoid strict AwsPolicyType on dynamic policies) --- iam-ape/tests/test_principal_conditions.py | 30 ++++++++++++---------- 1 file changed, 16 insertions(+), 14 deletions(-) diff --git a/iam-ape/tests/test_principal_conditions.py b/iam-ape/tests/test_principal_conditions.py index 335f182..ace9d0c 100644 --- a/iam-ape/tests/test_principal_conditions.py +++ b/iam-ape/tests/test_principal_conditions.py @@ -1,15 +1,9 @@ +from typing import Any + from iam_ape.evaluator import AuthorizationDetails, EffectivePolicyEvaluator from iam_ape.helper_classes import PolicyWithSource -from iam_ape.helper_types import AwsPolicyType, EntityType - -_ADMIN: AwsPolicyType = { - "Version": "2012-10-17", - "Statement": [{"Effect": "Allow", "Action": "*", "Resource": "*"}], -} -_FULL_AWS_ACCESS: AwsPolicyType = { - "Version": "2012-10-17", - "Statement": [{"Effect": "Allow", "Action": "*", "Resource": "*"}], -} +from iam_ape.helper_types import EntityType + _GUARDRAIL_ACTIONS = [ "bedrock:CreateGuardrail", "bedrock:UpdateGuardrail", @@ -18,14 +12,22 @@ _SSO_ADMIN_PATTERN = "arn:aws:iam::*:role/aws-reserved/sso.amazonaws.com/*/AWSReservedSSO_AdministratorAccess*" -def _scp(condition=None) -> AwsPolicyType: - statement = {"Effect": "Deny", "Action": _GUARDRAIL_ACTIONS, "Resource": ["*"]} +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 {"Version": "2012-10-17", "Statement": [statement]} + return _policy(statement) -def _evaluate(arn, scp, tags=None): +def _evaluate(arn: str, scp: Any, tags: Any = None) -> Any: user = { "UserName": "u", "Arn": arn, From af99b80ed35a69b64eca6e8a2a5b0c1b34d4778f Mon Sep 17 00:00:00 2001 From: Yana Sidanych Date: Wed, 8 Jul 2026 17:40:17 +0300 Subject: [PATCH 4/4] iam-ape: address SonarCloud criticals (constants for principal keys, reduce resolver complexity) --- iam-ape/iam_ape/evaluator.py | 57 +++++++++++++++++++++++------------- 1 file changed, 36 insertions(+), 21 deletions(-) diff --git a/iam-ape/iam_ape/evaluator.py b/iam-ape/iam_ape/evaluator.py index 272ac8e..64c1d6f 100644 --- a/iam-ape/iam_ape/evaluator.py +++ b/iam-ape/iam_ape/evaluator.py @@ -31,6 +31,8 @@ _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() @@ -39,18 +41,18 @@ 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 { - "aws:principalarn": arn, - "aws:principalaccount": parts[4] if len(parts) > 4 and parts[4] else None, + _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 == "aws:principalarn": - return context["aws:principalarn"] - if key_lower == "aws:principalaccount": - return context["aws:principalaccount"] or _UNKNOWN + 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 @@ -63,7 +65,7 @@ def condition_is_principal_resolvable(condition: Dict[str, Any]) -> bool: return False for key in key_values: key_lower = key.lower() - if key_lower in ("aws:principalarn", "aws:principalaccount"): + if key_lower in (_PRINCIPAL_ARN_KEY, _PRINCIPAL_ACCOUNT_KEY): continue if key_lower.startswith(_PRINCIPAL_TAG_PREFIX): continue @@ -71,6 +73,29 @@ def condition_is_principal_resolvable(condition: Dict[str, Any]) -> bool: 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]: @@ -83,20 +108,10 @@ def resolve_principal_condition( for operator, key_values in condition.items(): if operator not in _MODELED_OPERATORS or not isinstance(key_values, dict): return None - 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] - if operator in _LIKE_OPERATORS: - satisfied = any(wildcard_match(value, p) for p in patterns) - elif operator in _EQUALS_OPERATORS: - satisfied = value in patterns - elif operator in _NOTLIKE_OPERATORS: - satisfied = not any(wildcard_match(value, p) for p in patterns) - else: - satisfied = value not in patterns - result = result and satisfied + group = _operator_group(operator, key_values, context) + if group is None: + return None + result = result and group return result