diff --git a/iam-ape/iam_ape/evaluator.py b/iam-ape/iam_ape/evaluator.py index 93daa54..64c1d6f 100644 --- a/iam-ape/iam_ape/evaluator.py +++ b/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, 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_conditions.py b/iam-ape/tests/test_principal_conditions.py new file mode 100644 index 0000000..ace9d0c --- /dev/null +++ b/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)) + )