Fix low-blast reasoner correctness bugs

iamscope/reasoner/cross_account_trust.py

@@ -61,6 +61,7 @@
 from typing import Any
 
 from iamscope.constants import (
+    CONSTRAINT_TYPE_SCP,
     NAKED_BROAD,
     NAKED_CONDITIONED,
     NAKED_CRITICAL,
@@ -429,13 +430,18 @@ def _evaluate_edge(
         )
 
         # ---- Check 4: no_scp_blocks_sts_assumerole.
-        # Walk all SCP bindings on this edge. In V1, all bindings on
-        # trust edges are SCPs (permission boundaries don't bind to
-        # trust edges), so we don't filter by constraint_type — every
-        # binding is in scope for this check.
+        # Walk only SCP bindings on this edge. The binding sidecar can
+        # attach SCP, TRUST_CONDITION, PERMISSION_BOUNDARY, and other
+        # constraints to trust edges; non-SCP constraints must not be
+        # treated as SCP blockers or SCP ambiguity.
         check_4_state = CheckState.PASS
         check_4_reason = "no SCP bindings observed on this edge"
-        bindings = facts.bindings_for_edge(edge_id)
+        bindings = [
+            binding
+            for binding in facts.bindings_for_edge(edge_id)
+            if (constraint := facts.constraint_by_id(binding.constraint_id)) is not None
+            and constraint.constraint_type == CONSTRAINT_TYPE_SCP
+        ]
         if bindings:
             edge_constraint_refs.update(f"{b.edge_id}|{b.constraint_id}" for b in bindings)
             constraint_refs.update(b.constraint_id for b in bindings)

iamscope/reasoner/passrole_ecs.py

@@ -66,6 +66,7 @@
 
 from __future__ import annotations
 
+import fnmatch
 import logging
 from typing import Any
 
@@ -1109,8 +1110,7 @@ def _check_passed_to_service(
                     found_in_supported_op = True
                     values = value if isinstance(value, list) else [value]
                     for v in values:
-                        v_lower = str(v).lower()
-                        if v_lower == _ECS_TASKS_SERVICE_PRINCIPAL:
+                        if self._passed_to_service_matches(operator, str(v), _ECS_TASKS_SERVICE_PRINCIPAL):
                             scoped_to_ecs_tasks = True
                         else:
                             other_services.append(str(v))
@@ -1145,6 +1145,26 @@ def _check_passed_to_service(
             f"(not ECS) — PassRole cannot pass to an ECS task role",
         )
 
+    def _passed_to_service_matches(
+        self,
+        operator: str,
+        condition_value: str,
+        service_principal: str,
+    ) -> bool:
+        """Evaluate supported iam:PassedToService operators.
+
+        StringEquals preserves exact string semantics. StringLike uses
+        AWS-style glob matching, case-insensitively, so values such as
+        `ecs-tasks.*`, `*.amazonaws.com`, and `*` match ECS tasks.
+        """
+        condition_value_lower = condition_value.lower()
+        service_principal_lower = service_principal.lower()
+        if operator == "StringEquals":
+            return condition_value_lower == service_principal_lower
+        if operator == "StringLike":
+            return fnmatch.fnmatchcase(service_principal_lower, condition_value_lower)
+        return False
+
     # ---------------------------------------------------------------
     # Admin-equivalence read
     # ---------------------------------------------------------------

iamscope/reasoner/passrole_lambda.py

@@ -66,6 +66,7 @@
 
 from __future__ import annotations
 
+import fnmatch
 import logging
 from typing import Any
 
@@ -1089,8 +1090,7 @@ def _check_passed_to_service(
                     found_in_supported_op = True
                     values = value if isinstance(value, list) else [value]
                     for v in values:
-                        v_lower = str(v).lower()
-                        if v_lower == _LAMBDA_SERVICE_PRINCIPAL:
+                        if self._passed_to_service_matches(operator, str(v), _LAMBDA_SERVICE_PRINCIPAL):
                             scoped_to_lambda = True
                         else:
                             other_services.append(str(v))
@@ -1125,6 +1125,26 @@ def _check_passed_to_service(
             f"(not Lambda) — PassRole cannot pass to a Lambda execution role",
         )
 
+    def _passed_to_service_matches(
+        self,
+        operator: str,
+        condition_value: str,
+        service_principal: str,
+    ) -> bool:
+        """Evaluate supported iam:PassedToService operators.
+
+        StringEquals preserves exact string semantics. StringLike uses
+        AWS-style glob matching, case-insensitively, so values such as
+        `lambda.*`, `*.amazonaws.com`, and `*` match Lambda.
+        """
+        condition_value_lower = condition_value.lower()
+        service_principal_lower = service_principal.lower()
+        if operator == "StringEquals":
+            return condition_value_lower == service_principal_lower
+        if operator == "StringLike":
+            return fnmatch.fnmatchcase(service_principal_lower, condition_value_lower)
+        return False
+
     # ---------------------------------------------------------------
     # Admin-equivalence read
     # ---------------------------------------------------------------

iamscope/reasoner/s3_bucket_takeover.py

@@ -229,43 +229,45 @@ def _build_finding(
             )
         )
 
-        # ---- Check 3: no_scp_blocks_put_bucket_policy
-        # Each check gets its own constraint_refs accumulator so the
-        # evidence_refs attributed to check 3 are ONLY the SCPs it
-        # evaluated, not a contaminated mix of SCPs and boundaries.
-        check_3_constraint_refs: set[str] = set()
-        check_3_state, check_3_reason, check_3_blockers = self._check_scp_blockers(
-            facts,
-            witness_edge,
-            check_3_constraint_refs,
-            edge_constraint_refs,
+        # ---- Check 3: target_bucket_collected
+        # Dangling bucket nodes are materialized from policy references
+        # when collection did not directly return the resource. Preserve
+        # the signal but demote to inconclusive rather than claiming a
+        # validated takeover against an uncollected bucket.
+        is_dangling_bucket = bool(bucket.properties.get("is_dangling_reference", False))
+        check_3_state = CheckState.UNKNOWN if is_dangling_bucket else CheckState.PASS
+        check_3_reason = (
+            "target bucket was referenced by policy but not directly collected"
+            if is_dangling_bucket
+            else "target bucket was directly collected"
         )
-        blockers.extend(check_3_blockers)
         check_results.append(
             Check(
-                name="no_scp_blocks_put_bucket_policy",
-                description=("No SCP blocks s3:PutBucketPolicy on this edge with complete governance confidence"),
-                state=check_3_state,
-                evidence_refs=(
-                    tuple(sorted(check_3_constraint_refs)) if check_3_constraint_refs else (witness_edge.edge_id,)
+                name="target_bucket_collected",
+                description=(
+                    "Target S3 bucket was directly collected, not only materialized from a dangling policy reference"
                 ),
+                state=check_3_state,
+                evidence_refs=(witness_edge.edge_id,),
                 reason=check_3_reason,
             )
         )
         trace.append(
             TraceEntry(
                 step=3,
-                action="check_no_scp_blocks_put_bucket_policy",
-                inputs=(witness_edge.edge_id,),
+                action="check_target_bucket_collected",
+                inputs=(bucket.provider_id,),
                 result=check_3_state.value.upper(),
                 reason=check_3_reason,
             )
         )
-        constraint_refs.update(check_3_constraint_refs)
 
-        # ---- Check 4: no_boundary_blocks_put_bucket_policy
+        # ---- Check 4: no_scp_blocks_put_bucket_policy
+        # Each check gets its own constraint_refs accumulator so the
+        # evidence_refs attributed to this check are ONLY the SCPs it
+        # evaluated, not a contaminated mix of SCPs and boundaries.
         check_4_constraint_refs: set[str] = set()
-        check_4_state, check_4_reason, check_4_blockers = self._check_boundary_blockers(
+        check_4_state, check_4_reason, check_4_blockers = self._check_scp_blockers(
             facts,
             witness_edge,
             check_4_constraint_refs,
@@ -274,8 +276,8 @@ def _build_finding(
         blockers.extend(check_4_blockers)
         check_results.append(
             Check(
-                name="no_boundary_blocks_put_bucket_policy",
-                description=("No permission boundary blocks s3:PutBucketPolicy on this edge"),
+                name="no_scp_blocks_put_bucket_policy",
+                description=("No SCP blocks s3:PutBucketPolicy on this edge with complete governance confidence"),
                 state=check_4_state,
                 evidence_refs=(
                     tuple(sorted(check_4_constraint_refs)) if check_4_constraint_refs else (witness_edge.edge_id,)
@@ -286,15 +288,46 @@ def _build_finding(
         trace.append(
             TraceEntry(
                 step=4,
-                action="check_no_boundary_blocks_put_bucket_policy",
+                action="check_no_scp_blocks_put_bucket_policy",
                 inputs=(witness_edge.edge_id,),
                 result=check_4_state.value.upper(),
                 reason=check_4_reason,
             )
         )
         constraint_refs.update(check_4_constraint_refs)
 
-        # ---- Check 5: principal_is_actionable
+        # ---- Check 5: no_boundary_blocks_put_bucket_policy
+        check_5_constraint_refs: set[str] = set()
+        check_5_state, check_5_reason, check_5_blockers = self._check_boundary_blockers(
+            facts,
+            witness_edge,
+            check_5_constraint_refs,
+            edge_constraint_refs,
+        )
+        blockers.extend(check_5_blockers)
+        check_results.append(
+            Check(
+                name="no_boundary_blocks_put_bucket_policy",
+                description=("No permission boundary blocks s3:PutBucketPolicy on this edge"),
+                state=check_5_state,
+                evidence_refs=(
+                    tuple(sorted(check_5_constraint_refs)) if check_5_constraint_refs else (witness_edge.edge_id,)
+                ),
+                reason=check_5_reason,
+            )
+        )
+        trace.append(
+            TraceEntry(
+                step=5,
+                action="check_no_boundary_blocks_put_bucket_policy",
+                inputs=(witness_edge.edge_id,),
+                result=check_5_state.value.upper(),
+                reason=check_5_reason,
+            )
+        )
+        constraint_refs.update(check_5_constraint_refs)
+
+        # ---- Check 6: principal_is_actionable
         # Filter out service principals and root. Service principals
         # (*.amazonaws.com) represent infrastructure, not attacker-
         # controlled entities. Root accounts are always "admin" and
@@ -316,7 +349,7 @@ def _build_finding(
         )
         trace.append(
             TraceEntry(
-                step=5,
+                step=6,
                 action="check_principal_is_actionable",
                 inputs=(principal.provider_id,),
                 result="PASS",