From b03c164b2d14cf724d01316bcb8d56bbfc72210a Mon Sep 17 00:00:00 2001 From: Eric Conklin Date: Thu, 4 Jun 2026 15:59:39 -0500 Subject: [PATCH] Fix admin reachability clean path tracking --- iamscope/reasoner/admin_reachability.py | 113 ++++++++++++++++------ tests/test_admin_reachability_reasoner.py | 103 ++++++++++++++++++++ 2 files changed, 186 insertions(+), 30 deletions(-) diff --git a/iamscope/reasoner/admin_reachability.py b/iamscope/reasoner/admin_reachability.py index 92e0613..881500e 100644 --- a/iamscope/reasoner/admin_reachability.py +++ b/iamscope/reasoner/admin_reachability.py @@ -23,8 +23,8 @@ that question. Verdict shape: - validated all reachable chains have clean witnesses - inconclusive at least one reachable chain traverses a hyperedge + validated at least one reachable admin chain has clean witnesses + inconclusive admin reachability exists only through a hyperedge or wildcard ambiguity No blocked verdict — SCP analysis is per-chain, and this reasoner is @@ -125,24 +125,28 @@ def _compute_reachability( """BFS from source, collecting all reachable admin-equivalent roles. Returns a Finding if at least one admin is reached, else None. - Tracks all visited roles, all per-hop edges, and whether the - walk encountered any hyperedge/wildcard ambiguity (drives - the inconclusive verdict). + Tracks all visited roles and per-hop edges. Clean versus + ambiguous reachability is tracked per BFS path so an unrelated + wildcard branch cannot poison a separate clean admin proof. """ reachable_admins: set[str] = set() # admin role ARNs + clean_reachable_admins: set[str] = set() # reached via non-ambiguous path + ambiguous_reachable_admins: set[str] = set() # reached via ambiguous path admin_witness_edges: list[Edge] = [] # admin equivalence proof edges - visited: set[str] = {source.provider_id} + clean_admin_witness_edges: list[Edge] = [] # admin witnesses for clean proofs + visited: set[tuple[str, bool]] = {(source.provider_id, True)} all_walk_edges: list[Edge] = [] # every permission + trust edge traversed + clean_proof_walk_edges: list[Edge] = [] # edges from clean admin-reaching path(s) all_visited_roles: list[Node] = [] # in BFS order any_hyperedge_traversed = False walk_hit_depth_limit = False - # Frontier: (current_arn, depth) - frontier: deque[tuple[str, int]] = deque() - frontier.append((source.provider_id, 0)) + # Frontier: (current_arn, depth, path_is_clean, path_edges) + frontier: deque[tuple[str, int, bool, tuple[Edge, ...]]] = deque() + frontier.append((source.provider_id, 0, True, ())) while frontier: - current_arn, depth = frontier.popleft() + current_arn, depth, path_is_clean, path_edges = frontier.popleft() # If current is a role and depth >= 1, check admin equivalence. if depth >= 1: @@ -154,9 +158,15 @@ def _compute_reachability( facts, current_node, ) - if admin_witness is not None and current_arn not in reachable_admins: + if admin_witness is not None: reachable_admins.add(current_arn) - admin_witness_edges.append(admin_witness) + self._append_unique_edge(admin_witness_edges, admin_witness) + if path_is_clean: + clean_reachable_admins.add(current_arn) + self._append_unique_edge(clean_admin_witness_edges, admin_witness) + self._append_unique_edges(clean_proof_walk_edges, path_edges) + else: + ambiguous_reachable_admins.add(current_arn) # Stop walking deeper if depth limit hit. if depth >= _MAX_DEPTH: @@ -169,8 +179,6 @@ def _compute_reachability( current_arn, ): next_arn = perm_edge.dst.provider_id - if next_arn in visited: - continue trust_edge = self._find_admitting_trust_edge( facts, @@ -182,15 +190,23 @@ def _compute_reachability( # Track ambiguity: if either edge is a hyperedge witness, # the walk has touched ambiguous ground. + hop_is_clean = True if self._is_ambiguous_edge(perm_edge): any_hyperedge_traversed = True + hop_is_clean = False if self._is_ambiguous_edge(trust_edge): any_hyperedge_traversed = True + hop_is_clean = False + + next_path_is_clean = path_is_clean and hop_is_clean + visited_key = (next_arn, next_path_is_clean) + if visited_key in visited: + continue - all_walk_edges.append(perm_edge) - all_walk_edges.append(trust_edge) - visited.add(next_arn) - frontier.append((next_arn, depth + 1)) + self._append_unique_edge(all_walk_edges, perm_edge) + self._append_unique_edge(all_walk_edges, trust_edge) + visited.add(visited_key) + frontier.append((next_arn, depth + 1, next_path_is_clean, path_edges + (perm_edge, trust_edge))) # No reachable admins → no finding if not reachable_admins: @@ -200,8 +216,12 @@ def _compute_reachability( facts=facts, source=source, reachable_admins=sorted(reachable_admins), + clean_reachable_admins=sorted(clean_reachable_admins), + ambiguous_reachable_admins=sorted(ambiguous_reachable_admins), admin_witness_edges=admin_witness_edges, + clean_admin_witness_edges=clean_admin_witness_edges, all_walk_edges=all_walk_edges, + clean_proof_walk_edges=clean_proof_walk_edges, all_visited_roles=all_visited_roles, any_hyperedge_traversed=any_hyperedge_traversed, walk_hit_depth_limit=walk_hit_depth_limit, @@ -257,6 +277,16 @@ def _is_ambiguous_edge(self, edge: Edge) -> bool: return _is_unknown_witness(edge) + def _append_unique_edge(self, edges: list[Edge], edge: Edge) -> None: + """Append edge once, preserving first-seen deterministic order.""" + if edge.edge_id not in {existing.edge_id for existing in edges}: + edges.append(edge) + + def _append_unique_edges(self, edges: list[Edge], new_edges: tuple[Edge, ...]) -> None: + """Append each edge once, preserving first-seen deterministic order.""" + for edge in new_edges: + self._append_unique_edge(edges, edge) + def _check_boundary_blockers_on_walk( self, facts: FactGraph, @@ -369,8 +399,12 @@ def _build_finding( facts: FactGraph, source: Node, reachable_admins: list[str], + clean_reachable_admins: list[str], + ambiguous_reachable_admins: list[str], admin_witness_edges: list[Edge], + clean_admin_witness_edges: list[Edge], all_walk_edges: list[Edge], + clean_proof_walk_edges: list[Edge], all_visited_roles: list[Node], any_hyperedge_traversed: bool, walk_hit_depth_limit: bool, @@ -479,9 +513,18 @@ def _build_finding( ) # ---- Check 3: at least one reachable chain uses clean witnesses - # If any hyperedge was traversed during the walk, the result is - # ambiguous (we can't prove which specific chains are real). - check_3_state = CheckState.UNKNOWN if any_hyperedge_traversed else CheckState.PASS + # Ambiguous alternate branches are still retained in evidence, but + # do not downgrade a separate clean proof path to an admin endpoint. + has_clean_admin_path = bool(clean_reachable_admins) + check_3_state = CheckState.PASS if has_clean_admin_path else CheckState.UNKNOWN + if has_clean_admin_path and any_hyperedge_traversed: + check_3_reason = ( + "at least one clean reachable admin path exists; ambiguous alternate walk evidence also observed" + ) + elif has_clean_admin_path: + check_3_reason = "all BFS paths use clean witness edges" + else: + check_3_reason = "BFS walk traversed at least one wildcard/hyperedge edge" check_results.append( Check( name="at_least_one_reachable_chain_uses_clean_witnesses", @@ -491,20 +534,28 @@ def _build_finding( ), state=check_3_state, evidence_refs=tuple(edge_refs), - reason=( - "BFS walk traversed at least one wildcard/hyperedge edge" - if any_hyperedge_traversed - else "all BFS paths use clean witness edges" - ), + reason=check_3_reason, ) ) trace.append( TraceEntry( step=3, action="check_at_least_one_reachable_chain_uses_clean_witnesses", - inputs=(str(any_hyperedge_traversed),), + inputs=( + ( + str(any_hyperedge_traversed), + str(len(clean_reachable_admins)), + str(len(ambiguous_reachable_admins)), + ) + if has_clean_admin_path and any_hyperedge_traversed + else (str(any_hyperedge_traversed),) + ), result=check_3_state.value.upper(), - reason="ambiguity in walk" if any_hyperedge_traversed else "clean walk", + reason=( + "clean path with ambiguous alternate walk" + if has_clean_admin_path and any_hyperedge_traversed + else ("clean walk" if has_clean_admin_path else "ambiguity in walk") + ), ) ) @@ -541,9 +592,11 @@ def _build_finding( # ---- Check 5: no permission boundary blocks reachable walk scp_constraint_refs: set[str] = set() scp_edge_constraint_refs: set[str] = set() + blocker_walk_edges = clean_proof_walk_edges if clean_reachable_admins else all_walk_edges + blocker_admin_witness_edges = clean_admin_witness_edges if clean_reachable_admins else admin_witness_edges check_5_state, check_5_reason, check_5_blockers = self._check_scp_blockers_on_walk( facts, - all_walk_edges, + blocker_walk_edges, scp_constraint_refs, scp_edge_constraint_refs, ) @@ -575,7 +628,7 @@ def _build_finding( boundary_edge_constraint_refs: set[str] = set() check_6_state, check_6_reason, check_6_blockers = self._check_boundary_blockers_on_walk( facts, - all_walk_edges + admin_witness_edges, + blocker_walk_edges + blocker_admin_witness_edges, boundary_constraint_refs, boundary_edge_constraint_refs, ) diff --git a/tests/test_admin_reachability_reasoner.py b/tests/test_admin_reachability_reasoner.py index c1fca0c..8798061 100644 --- a/tests/test_admin_reachability_reasoner.py +++ b/tests/test_admin_reachability_reasoner.py @@ -28,6 +28,7 @@ from tests.test_assume_role_chain_reasoner import ( # noqa: I001 _ADMIN_ARN, _ALICE_ARN, + _DEPLOY_ARN, _DEVOPS_ARN, _NON_ADMIN_ARN, _PROD_ARN, @@ -441,6 +442,63 @@ def test_complete_scp_block_downgrades_reachable_admin_to_blocked(self) -> None: class TestHyperedgeInconclusive: + def test_clean_admin_path_plus_unrelated_ambiguous_branch_stays_validated(self) -> None: + """An ambiguous alternate branch must not poison a separate clean admin proof.""" + facts = _build_two_hop_chain() + non_admin = _role(_NON_ADMIN_ARN) + ambiguous_perm = _assume_perm_edge( + src_arn=_ALICE_ARN, + dst_arn=_NON_ADMIN_ARN, + digest="8" * 64, + is_wildcard_resource=True, + ) + ambiguous_trust = _trust_edge( + principal_arn=_ALICE_ARN, + target_arn=_NON_ADMIN_ARN, + digest="9" * 64, + ) + branched = _make_facts( + nodes=(*facts.nodes, non_admin), + edges=(*facts.edges, ambiguous_perm, ambiguous_trust), + ) + + findings = AdminReachabilityReasoner().run(branched) + alice_f = next(f for f in findings if f.source.provider_id == _ALICE_ARN) + + assert alice_f.verdict.value == "validated" + check = next( + c for c in alice_f.required_checks if c.name == "at_least_one_reachable_chain_uses_clean_witnesses" + ) + assert check.state.value == "pass" + assert "ambiguous alternate walk evidence" in check.reason + assert ambiguous_perm.edge_id in alice_f.evidence.edge_refs + assert ambiguous_trust.edge_id in alice_f.evidence.edge_refs + + def test_only_ambiguous_path_to_admin_remains_inconclusive(self) -> None: + """Admin reachability through only a wildcard hop remains inconclusive.""" + alice = _user(_ALICE_ARN) + admin = _role(_ADMIN_ARN) + wildcard_perm = _assume_perm_edge( + src_arn=_ALICE_ARN, + dst_arn=_ADMIN_ARN, + is_wildcard_resource=True, + ) + trust = _trust_edge(principal_arn=_ALICE_ARN, target_arn=_ADMIN_ARN) + admin_grant = _admin_grant_edge(_ADMIN_ARN) + facts = _make_facts( + nodes=(alice, admin), + edges=(wildcard_perm, trust, admin_grant), + ) + + findings = AdminReachabilityReasoner().run(facts) + alice_f = next(f for f in findings if f.source.provider_id == _ALICE_ARN) + + assert alice_f.verdict.value == "inconclusive" + check = next( + c for c in alice_f.required_checks if c.name == "at_least_one_reachable_chain_uses_clean_witnesses" + ) + assert check.state.value == "unknown" + def test_wildcard_hop_produces_inconclusive(self) -> None: """Wildcard sts:AssumeRole on first hop → check 3 UNKNOWN.""" alice = _user(_ALICE_ARN) @@ -467,6 +525,51 @@ def test_wildcard_hop_produces_inconclusive(self) -> None: assert c.state.value == "unknown" +class TestDepthLimitConservative: + def test_clean_admin_path_still_inconclusive_when_alternate_walk_hits_depth_limit(self) -> None: + alice = _user(_ALICE_ARN) + admin = _role(_ADMIN_ARN) + deploy = _role(_DEPLOY_ARN) + devops = _role(_DEVOPS_ARN) + prod = _role(_PROD_ARN) + non_admin = _role(_NON_ADMIN_ARN) + + direct_perm = _assume_perm_edge(src_arn=_ALICE_ARN, dst_arn=_ADMIN_ARN, digest="1" * 64) + direct_trust = _trust_edge(principal_arn=_ALICE_ARN, target_arn=_ADMIN_ARN, digest="2" * 64) + branch_1_perm = _assume_perm_edge(src_arn=_ALICE_ARN, dst_arn=_DEPLOY_ARN, digest="3" * 64) + branch_1_trust = _trust_edge(principal_arn=_ALICE_ARN, target_arn=_DEPLOY_ARN, digest="4" * 64) + branch_2_perm = _assume_perm_edge(src_arn=_DEPLOY_ARN, dst_arn=_DEVOPS_ARN, digest="5" * 64) + branch_2_trust = _trust_edge(principal_arn=_DEPLOY_ARN, target_arn=_DEVOPS_ARN, digest="6" * 64) + branch_3_perm = _assume_perm_edge(src_arn=_DEVOPS_ARN, dst_arn=_PROD_ARN, digest="7" * 64) + branch_3_trust = _trust_edge(principal_arn=_DEVOPS_ARN, target_arn=_PROD_ARN, digest="8" * 64) + branch_4_perm = _assume_perm_edge(src_arn=_PROD_ARN, dst_arn=_NON_ADMIN_ARN, digest="9" * 64) + branch_4_trust = _trust_edge(principal_arn=_PROD_ARN, target_arn=_NON_ADMIN_ARN, digest="0" * 64) + admin_grant = _admin_grant_edge(_ADMIN_ARN) + facts = _make_facts( + nodes=(alice, admin, deploy, devops, prod, non_admin), + edges=( + direct_perm, + direct_trust, + branch_1_perm, + branch_1_trust, + branch_2_perm, + branch_2_trust, + branch_3_perm, + branch_3_trust, + branch_4_perm, + branch_4_trust, + admin_grant, + ), + ) + + findings = AdminReachabilityReasoner().run(facts) + alice_f = next(f for f in findings if f.source.provider_id == _ALICE_ARN) + + assert alice_f.verdict.value == "inconclusive" + check = next(c for c in alice_f.required_checks if c.name == "walk_terminated_within_depth_limit") + assert check.state.value == "unknown" + + # --------------------------------------------------------------------------- # 3-hop chain — Alice reaches Admin via DevOps → Prod (intermediate not admin) # ---------------------------------------------------------------------------