mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-09 21:14:22 +00:00
feat(attack-paths): Add 4 IAM privilege escalation detection queries (#11460)
Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com> Co-authored-by: Josema Camacho <josema@prowler.com>
This commit is contained in:
Paramanand Mallik
co-authored by
Daniel Barranquero
Josema Camacho
Daniel Barranquero
Josema Camacho
parent
7275b46707
commit
88c666a0d2
@@ -0,0 +1 @@
|
||||
Attack Paths adds four AWS privilege-escalation detection queries from pathfinding.cloud: cross-account role trust (STS-002), wildcard role trust (STS-003), user permissions-boundary removal (IAM-022), and IAM Identity Center permission-set escalation (SSO-001)
|
||||
@@ -3536,6 +3536,160 @@ AWS_STS_PRIVESC_ASSUME_ROLE = AttackPathsQueryDefinition(
|
||||
parameters=[],
|
||||
)
|
||||
|
||||
# STS-002
|
||||
AWS_STS_PRIVESC_CROSS_ACCOUNT_TRUST = AttackPathsQueryDefinition(
|
||||
id="aws-sts-privesc-cross-account-trust",
|
||||
name="Cross-Account Role Trust for Privilege Escalation (STS-002)",
|
||||
short_description="Roles that trust an external account's root principal can be assumed by any principal in that account, enabling confused-deputy escalation.",
|
||||
description="Detect IAM roles whose trust policy allows an external AWS account root principal (arn:aws:iam::<account-id>:root) to assume them. Any principal in the trusted external account that holds sts:AssumeRole can assume the role and gain its permissions, which is the confused-deputy escalation surface. The ingested graph does not record trust-policy conditions, so roles protected by an sts:ExternalId condition cannot be filtered out automatically and are surfaced here for manual review.",
|
||||
attribution=AttackPathsQueryAttribution(
|
||||
text="pathfinding.cloud - STS-002 - sts:AssumeRole",
|
||||
link="https://pathfinding.cloud/paths/sts-002",
|
||||
),
|
||||
provider="aws",
|
||||
cypher=f"""
|
||||
// Find roles that trust an external account's root principal (cross-account trust)
|
||||
MATCH path_target = (aws:AWSAccount {{id: $provider_uid}})--(target_role:AWSRole)-[:TRUSTS_AWS_PRINCIPAL]->(trusted:AWSRootPrincipal)
|
||||
WHERE trusted.arn CONTAINS ':root'
|
||||
AND NOT trusted.arn CONTAINS aws.id
|
||||
|
||||
WITH DISTINCT path_target
|
||||
WITH collect(path_target) AS paths
|
||||
UNWIND paths AS p
|
||||
UNWIND nodes(p) AS n
|
||||
|
||||
WITH paths, collect(DISTINCT n) AS unique_nodes
|
||||
UNWIND unique_nodes AS n
|
||||
|
||||
OPTIONAL MATCH (n)-[pfr:HAS_FINDING]-(pf:{PROWLER_FINDING_LABEL} {{status: 'FAIL'}})
|
||||
|
||||
RETURN paths, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr
|
||||
""",
|
||||
parameters=[],
|
||||
)
|
||||
|
||||
# STS-003
|
||||
AWS_STS_PRIVESC_WILDCARD_TRUST = AttackPathsQueryDefinition(
|
||||
id="aws-sts-privesc-wildcard-trust",
|
||||
name="Potential Wildcard Role Trust (STS-003)",
|
||||
short_description="Potential wildcard role trusts that need manual review before they are treated as assumable.",
|
||||
description='Find IAM roles linked to a wildcard principal ("AWS": "*"). The ingested graph does not preserve trust-policy Effect or Condition fields, so a match can come from a Deny statement or a restricted Allow statement. Treat each result as a candidate for manual review, not as a confirmed assumable role.',
|
||||
attribution=AttackPathsQueryAttribution(
|
||||
text="pathfinding.cloud - STS-003 - sts:AssumeRole",
|
||||
link="https://pathfinding.cloud/paths/sts-003",
|
||||
),
|
||||
provider="aws",
|
||||
cypher=f"""
|
||||
// Find roles linked to a wildcard principal for manual review
|
||||
MATCH path_target = (aws:AWSAccount {{id: $provider_uid}})--(target_role:AWSRole)-[:TRUSTS_AWS_PRINCIPAL]->(trusted:AWSPrincipal)
|
||||
WHERE trusted.arn = '*'
|
||||
|
||||
WITH DISTINCT path_target
|
||||
WITH collect(path_target) AS paths
|
||||
UNWIND paths AS p
|
||||
UNWIND nodes(p) AS n
|
||||
|
||||
WITH paths, collect(DISTINCT n) AS unique_nodes
|
||||
UNWIND unique_nodes AS n
|
||||
|
||||
OPTIONAL MATCH (n)-[pfr:HAS_FINDING]-(pf:{PROWLER_FINDING_LABEL} {{status: 'FAIL'}})
|
||||
|
||||
RETURN paths, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr
|
||||
""",
|
||||
parameters=[],
|
||||
)
|
||||
|
||||
# IAM-022
|
||||
AWS_IAM_PRIVESC_DELETE_USER_PERMISSIONS_BOUNDARY = AttackPathsQueryDefinition(
|
||||
id="aws-iam-privesc-delete-user-permissions-boundary",
|
||||
name="Permissions Boundary Removal for Self-Escalation (IAM-022)",
|
||||
short_description="IAM users that can remove their own permissions boundary, if one is attached.",
|
||||
description="Find IAM users whose policies allow iam:DeleteUserPermissionsBoundary on their own user ARN. The graph does not record whether a boundary is attached or whether removing it grants more access, so each result needs manual review.",
|
||||
attribution=AttackPathsQueryAttribution(
|
||||
text="pathfinding.cloud - IAM-022 - iam:DeleteUserPermissionsBoundary",
|
||||
link="https://pathfinding.cloud/paths/iam-022",
|
||||
),
|
||||
provider="aws",
|
||||
cypher=f"""
|
||||
// Find IAM users with iam:DeleteUserPermissionsBoundary permission
|
||||
MATCH path_principal = (aws:AWSAccount {{id: $provider_uid}})--(principal:AWSUser)-[:POLICY]->(policy:AWSPolicy)-[:STATEMENT]->(stmt:AWSPolicyStatement {{effect: 'Allow'}})
|
||||
MATCH (stmt)-[:HAS_ACTION]->(act:AWSPolicyStatementActionItem)
|
||||
WHERE toLower(act.value) IN ['iam:*', 'iam:deleteuserpermissionsboundary']
|
||||
OR act.value = '*'
|
||||
WITH DISTINCT principal, stmt, path_principal
|
||||
|
||||
// Keep only users that can remove the boundary from their own user ARN
|
||||
MATCH (stmt)-[:HAS_RESOURCE]->(res:AWSPolicyStatementResourceItem)
|
||||
WHERE res.value = '*'
|
||||
OR res.value = principal.arn
|
||||
OR (res.value ENDS WITH '*' AND principal.arn STARTS WITH replace(res.value, '*', ''))
|
||||
|
||||
WITH DISTINCT path_principal
|
||||
WITH collect(path_principal) AS paths
|
||||
UNWIND paths AS p
|
||||
UNWIND nodes(p) AS n
|
||||
|
||||
WITH paths, collect(DISTINCT n) AS unique_nodes
|
||||
UNWIND unique_nodes AS n
|
||||
|
||||
OPTIONAL MATCH (n)-[pfr:HAS_FINDING]-(pf:{PROWLER_FINDING_LABEL} {{status: 'FAIL'}})
|
||||
|
||||
RETURN paths, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr
|
||||
""",
|
||||
parameters=[],
|
||||
)
|
||||
|
||||
# SSO-001
|
||||
AWS_SSO_PRIVESC_PERMISSION_SET_ESCALATION = AttackPathsQueryDefinition(
|
||||
id="aws-sso-privesc-permission-set-escalation",
|
||||
name="Identity Center Permission Set Escalation (SSO-001)",
|
||||
short_description="Create an administrative Identity Center permission set and assign it to gain organization-wide admin access.",
|
||||
description="Detect principals that hold sso:CreatePermissionSet, sso:AttachManagedPolicyToPermissionSet, and sso:CreateAccountAssignment together. With all three, a principal can create a new IAM Identity Center permission set, attach the AdministratorAccess managed policy to it, and assign it to their own user or group for any account in the organization, gaining administrative access across the organization through the Identity Center portal.",
|
||||
attribution=AttackPathsQueryAttribution(
|
||||
text="pathfinding.cloud - SSO-001 - sso:CreatePermissionSet + sso:AttachManagedPolicyToPermissionSet + sso:CreateAccountAssignment",
|
||||
link="https://pathfinding.cloud/paths/sso-001",
|
||||
),
|
||||
provider="aws",
|
||||
cypher=f"""
|
||||
// Find principals with sso:CreatePermissionSet permission
|
||||
MATCH path_principal = (aws:AWSAccount {{id: $provider_uid}})--(principal:AWSPrincipal)-[:POLICY]->(policy:AWSPolicy)-[:STATEMENT]->(stmt:AWSPolicyStatement {{effect: 'Allow'}})
|
||||
MATCH (stmt)-[:HAS_ACTION]->(act:AWSPolicyStatementActionItem)
|
||||
WHERE toLower(act.value) IN ['sso:*', 'sso:createpermissionset']
|
||||
OR act.value = '*'
|
||||
MATCH (stmt)-[:HAS_RESOURCE]->(res:AWSPolicyStatementResourceItem)
|
||||
WHERE res.value = '*'
|
||||
WITH DISTINCT aws, principal, path_principal
|
||||
|
||||
// Find sso:AttachManagedPolicyToPermissionSet permission on the same principal
|
||||
MATCH (principal)-[:POLICY]->(:AWSPolicy)-[:STATEMENT]->(stmt2:AWSPolicyStatement {{effect: 'Allow'}})-[:HAS_ACTION]->(act2:AWSPolicyStatementActionItem)
|
||||
WHERE toLower(act2.value) IN ['sso:*', 'sso:attachmanagedpolicytopermissionset']
|
||||
OR act2.value = '*'
|
||||
MATCH (stmt2)-[:HAS_RESOURCE]->(res2:AWSPolicyStatementResourceItem)
|
||||
WHERE res2.value = '*'
|
||||
WITH DISTINCT principal, path_principal
|
||||
|
||||
// Find sso:CreateAccountAssignment permission on the same principal
|
||||
MATCH (principal)-[:POLICY]->(:AWSPolicy)-[:STATEMENT]->(stmt3:AWSPolicyStatement {{effect: 'Allow'}})-[:HAS_ACTION]->(act3:AWSPolicyStatementActionItem)
|
||||
WHERE toLower(act3.value) IN ['sso:*', 'sso:createaccountassignment']
|
||||
OR act3.value = '*'
|
||||
MATCH (stmt3)-[:HAS_RESOURCE]->(res3:AWSPolicyStatementResourceItem)
|
||||
WHERE res3.value = '*'
|
||||
|
||||
WITH DISTINCT path_principal
|
||||
WITH collect(path_principal) AS paths
|
||||
UNWIND paths AS p
|
||||
UNWIND nodes(p) AS n
|
||||
|
||||
WITH paths, collect(DISTINCT n) AS unique_nodes
|
||||
UNWIND unique_nodes AS n
|
||||
|
||||
OPTIONAL MATCH (n)-[pfr:HAS_FINDING]-(pf:{PROWLER_FINDING_LABEL} {{status: 'FAIL'}})
|
||||
|
||||
RETURN paths, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr
|
||||
""",
|
||||
parameters=[],
|
||||
)
|
||||
|
||||
# AWS Queries List
|
||||
|
||||
AWS_QUERIES: list[AttackPathsQueryDefinition] = [
|
||||
@@ -3617,4 +3771,8 @@ AWS_QUERIES: list[AttackPathsQueryDefinition] = [
|
||||
AWS_SSM_PRIVESC_START_SESSION,
|
||||
AWS_SSM_PRIVESC_SEND_COMMAND,
|
||||
AWS_STS_PRIVESC_ASSUME_ROLE,
|
||||
AWS_STS_PRIVESC_CROSS_ACCOUNT_TRUST,
|
||||
AWS_STS_PRIVESC_WILDCARD_TRUST,
|
||||
AWS_IAM_PRIVESC_DELETE_USER_PERMISSIONS_BOUNDARY,
|
||||
AWS_SSO_PRIVESC_PERMISSION_SET_ESCALATION,
|
||||
]
|
||||
|
||||
@@ -0,0 +1,245 @@
|
||||
"""
|
||||
Structural validation tests for Attack Paths query definitions.
|
||||
|
||||
These tests verify that each query in the AWS_QUERIES registry meets the
|
||||
schema and convention requirements documented in
|
||||
`docs/developer-guide/attack-paths-queries.mdx` without requiring a live
|
||||
graph connection. They deliberately assert the conventions that keep queries
|
||||
functional and Neptune-compatible: list-typed policy properties are reached
|
||||
through `HAS_*` child-item traversals (never read as node fields), predicate
|
||||
functions unsupported on Neptune (`any`/`all`/`none`, regex `=~`) are absent,
|
||||
the finding probe is typed and filters only on `status`, and the `RETURN`
|
||||
shape preserves the `paths, dpf, dpfr` contract.
|
||||
"""
|
||||
|
||||
import re
|
||||
|
||||
import pytest
|
||||
from api.attack_paths.queries.aws import (
|
||||
AWS_IAM_PRIVESC_DELETE_USER_PERMISSIONS_BOUNDARY,
|
||||
AWS_QUERIES,
|
||||
AWS_SSO_PRIVESC_PERMISSION_SET_ESCALATION,
|
||||
AWS_STS_PRIVESC_CROSS_ACCOUNT_TRUST,
|
||||
AWS_STS_PRIVESC_WILDCARD_TRUST,
|
||||
)
|
||||
from api.attack_paths.queries.types import AttackPathsQueryDefinition
|
||||
|
||||
# The pathfinding.cloud privilege-escalation queries added for PROWLER-2278.
|
||||
NEW_PATHFINDING_QUERIES = [
|
||||
AWS_STS_PRIVESC_CROSS_ACCOUNT_TRUST,
|
||||
AWS_STS_PRIVESC_WILDCARD_TRUST,
|
||||
AWS_IAM_PRIVESC_DELETE_USER_PERMISSIONS_BOUNDARY,
|
||||
AWS_SSO_PRIVESC_PERMISSION_SET_ESCALATION,
|
||||
]
|
||||
|
||||
# Cypher keywords that indicate a mutating query (not allowed; queries are read-only).
|
||||
MUTATING_KEYWORDS = re.compile(
|
||||
r"\b(CREATE|MERGE|SET|DELETE|REMOVE|DETACH)\b", re.IGNORECASE
|
||||
)
|
||||
|
||||
# CALL subquery: unsupported by Neptune openCypher.
|
||||
CALL_SUBQUERY_PATTERN = re.compile(r"\bCALL\s*\{", re.IGNORECASE)
|
||||
|
||||
# Predicate functions that are not part of the openCypher spec and fail on Neptune.
|
||||
NEPTUNE_UNSUPPORTED_PREDICATES = re.compile(r"\b(any|all|none)\s*\(", re.IGNORECASE)
|
||||
|
||||
# The list-typed policy properties that are exploded into child item nodes at sync
|
||||
# time and popped off the parent, so reading them as a field always yields null.
|
||||
NORMALIZED_STATEMENT_FIELDS = ("action", "resource", "notaction", "notresource")
|
||||
|
||||
|
||||
class TestNewPathfindingQueriesRegistered:
|
||||
"""Every new query is present in the AWS_QUERIES registry."""
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_query_in_registry(self, query):
|
||||
assert query in AWS_QUERIES
|
||||
|
||||
|
||||
class TestNewPathfindingQueriesSchema:
|
||||
"""Required fields and naming conventions for each new query."""
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_is_query_definition_instance(self, query):
|
||||
assert isinstance(query, AttackPathsQueryDefinition)
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_id_is_kebab_case(self, query):
|
||||
assert re.match(r"^[a-z0-9]+(-[a-z0-9]+)*$", query.id), (
|
||||
f"Query id '{query.id}' is not kebab-case"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_id_starts_with_aws(self, query):
|
||||
assert query.id.startswith("aws-")
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_provider_is_aws(self, query):
|
||||
assert query.provider == "aws"
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_has_name(self, query):
|
||||
assert query.name and len(query.name) > 5
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_has_short_description(self, query):
|
||||
assert query.short_description and len(query.short_description) > 10
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_has_description(self, query):
|
||||
assert query.description and len(query.description) > 20
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_has_attribution(self, query):
|
||||
assert query.attribution is not None
|
||||
assert "pathfinding.cloud" in query.attribution.text
|
||||
assert query.attribution.link.startswith("https://pathfinding.cloud/paths/")
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_parameters_is_list(self, query):
|
||||
assert isinstance(query.parameters, list)
|
||||
|
||||
|
||||
class TestNewPathfindingQueriesCypher:
|
||||
"""Cypher content, conventions, and Neptune compatibility."""
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_cypher_not_empty(self, query):
|
||||
assert query.cypher and len(query.cypher.strip()) > 0
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_cypher_under_10000_chars(self, query):
|
||||
assert len(query.cypher) < 10000, (
|
||||
f"Query {query.id} exceeds 10,000 character limit "
|
||||
f"({len(query.cypher)} chars)"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_cypher_uses_provider_uid_parameter(self, query):
|
||||
assert "$provider_uid" in query.cypher, (
|
||||
f"Query {query.id} missing $provider_uid parameter"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_finding_label_interpolated(self, query):
|
||||
# The f-string should have interpolated PROWLER_FINDING_LABEL already.
|
||||
assert "PROWLER_FINDING_LABEL" not in query.cypher, (
|
||||
f"Query {query.id} has unresolved PROWLER_FINDING_LABEL "
|
||||
"(f-string not applied)"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_finding_probe_is_typed_and_status_scoped(self, query):
|
||||
# The finding probe must be typed HAS_FINDING (so Neptune applies an inline
|
||||
# edge filter) and gate on FAIL status only. ProwlerFinding nodes carry no
|
||||
# provider_uid property, so a probe that filters on it never matches.
|
||||
assert re.search(
|
||||
r"-\[pfr:HAS_FINDING\]-\(pf:ProwlerFinding \{status: 'FAIL'\}\)",
|
||||
query.cypher,
|
||||
), f"Query {query.id} does not use the typed, status-scoped finding probe"
|
||||
assert "provider_uid:$provider_uid}" not in query.cypher.replace(" ", ""), (
|
||||
f"Query {query.id} filters the finding node on a non-existent "
|
||||
"provider_uid property"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_cypher_is_read_only(self, query):
|
||||
cypher_no_comments = _strip_comment_lines(query.cypher)
|
||||
match = MUTATING_KEYWORDS.search(cypher_no_comments)
|
||||
assert match is None, (
|
||||
f"Query {query.id} contains mutating keyword: '{match.group()}'"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_cypher_no_call_subquery(self, query):
|
||||
assert not CALL_SUBQUERY_PATTERN.search(query.cypher), (
|
||||
f"Query {query.id} uses a CALL subquery (not Neptune-compatible)"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_cypher_no_neptune_unsupported_predicates(self, query):
|
||||
match = NEPTUNE_UNSUPPORTED_PREDICATES.search(query.cypher)
|
||||
assert match is None, (
|
||||
f"Query {query.id} uses '{match.group().strip()}' predicate function; "
|
||||
"use size([x IN list WHERE pred]) > 0 for Neptune compatibility"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_cypher_no_regex_operator(self, query):
|
||||
assert "=~" not in query.cypher, (
|
||||
f"Query {query.id} uses the regex operator '=~'; "
|
||||
"use CONTAINS / STARTS WITH for Neptune compatibility"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_cypher_does_not_read_normalized_list_fields(self, query):
|
||||
# action/resource/notaction/notresource are materialized as child item nodes
|
||||
# and popped off AWSPolicyStatement, so `stmt.action` etc. are always null.
|
||||
for field in NORMALIZED_STATEMENT_FIELDS:
|
||||
assert not re.search(rf"\.{field}\b", query.cypher), (
|
||||
f"Query {query.id} reads the normalized list field "
|
||||
f"'.{field}' as a node property; traverse the HAS_"
|
||||
f"{field.upper()} edge to the child item node instead"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_cypher_preserves_return_contract(self, query):
|
||||
assert re.search(
|
||||
r"RETURN paths, collect\(DISTINCT pf\) as dpf, "
|
||||
r"collect\(DISTINCT pfr\) as dpfr",
|
||||
query.cypher,
|
||||
), f"Query {query.id} does not preserve the 'paths, dpf, dpfr' RETURN contract"
|
||||
|
||||
@pytest.mark.parametrize("query", NEW_PATHFINDING_QUERIES, ids=lambda q: q.id)
|
||||
def test_cypher_anchored_on_account(self, query):
|
||||
assert "(aws:AWSAccount {id: $provider_uid})" in query.cypher, (
|
||||
f"Query {query.id} is not anchored on the AWSAccount node"
|
||||
)
|
||||
|
||||
|
||||
class TestNewPathfindingQueriesAccuracy:
|
||||
"""Query-specific contracts that prevent known false positives."""
|
||||
|
||||
def test_wildcard_trust_is_presented_as_a_manual_review_candidate(self):
|
||||
query = AWS_STS_PRIVESC_WILDCARD_TRUST
|
||||
text = f"{query.name} {query.short_description} {query.description}".lower()
|
||||
assert all(
|
||||
word in text
|
||||
for word in ("potential", "effect", "condition", "manual review")
|
||||
)
|
||||
|
||||
def test_permissions_boundary_removal_is_scoped_to_the_same_user(self):
|
||||
query = AWS_IAM_PRIVESC_DELETE_USER_PERMISSIONS_BOUNDARY
|
||||
assert "(principal:AWSUser)" in query.cypher
|
||||
assert (
|
||||
"(stmt)-[:HAS_RESOURCE]->(res:AWSPolicyStatementResourceItem)"
|
||||
in query.cypher
|
||||
)
|
||||
assert "principal.arn" in query.cypher
|
||||
assert "manual review" in query.description.lower()
|
||||
|
||||
def test_permission_set_escalation_requires_global_resources(self):
|
||||
query = AWS_SSO_PRIVESC_PERMISSION_SET_ESCALATION
|
||||
for suffix in ("", "2", "3"):
|
||||
resource_match = (
|
||||
f"(stmt{suffix})-[:HAS_RESOURCE]->"
|
||||
f"(res{suffix}:AWSPolicyStatementResourceItem)"
|
||||
)
|
||||
assert resource_match in query.cypher
|
||||
assert f"WHERE res{suffix}.value = '*'" in query.cypher
|
||||
|
||||
|
||||
class TestAllQueriesUniqueIds:
|
||||
"""No duplicate IDs in the full registry."""
|
||||
|
||||
def test_no_duplicate_ids_in_aws_queries(self):
|
||||
ids = [q.id for q in AWS_QUERIES]
|
||||
duplicates = sorted({qid for qid in ids if ids.count(qid) > 1})
|
||||
assert not duplicates, f"Duplicate query IDs found: {duplicates}"
|
||||
|
||||
|
||||
def _strip_comment_lines(cypher: str) -> str:
|
||||
"""Drop `//` comment lines so keyword scans ignore prose in comments."""
|
||||
return "\n".join(
|
||||
line for line in cypher.split("\n") if not line.strip().startswith("//")
|
||||
)
|
||||
Reference in New Issue
Block a user