Skip to content

Fix admin reachability clean path tracking #49

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
InfoSecHack merged 1 commit into from
Jun 4, 2026
Merged

Fix admin reachability clean path tracking #49

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
113 changes: 83 additions & 30 deletions iamscope/reasoner/admin_reachability.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand All @@ -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,
Expand All @@ -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:
Expand All @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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",
Expand All @@ -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")
),
)
)

Expand Down Expand Up @@ -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,
)
Expand Down Expand Up @@ -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,
)
Expand Down
103 changes: 103 additions & 0 deletions tests/test_admin_reachability_reasoner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)
Expand All @@ -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)
# ---------------------------------------------------------------------------
Expand Down