feat(api): strip account nodes and surface attack outcome metadata

- Drop account/root-labeled nodes (AWSAccount, AzureSubscription, AzureTenant, GCPProject, KubernetesCluster, GitHubAccount) and their relationships from the serialized attack-paths graph. The account stays the Cypher MATCH anchor; tenant/provider isolation is unaffected.
- Add AttackPathsQueryOutcome (label, description, severity) and an outcome field on AttackPathsQueryDefinition. Assign outcomes to every real attack-path query by id pattern in aws.py (privesc-passrole and code-exec to 'Code execution', IAM/STS privesc to 'Privilege escalation', internet-exposed chain to 'Data exfiltration'). The 11 inventory queries keep outcome=None.
- execute_query attaches outcome to the run-query response; expose it via new AttackPathsQueryOutcomeSerializer in AttackPathsQuerySerializer and AttackPathsQueryResultSerializer (allow_null).
- attack_paths_queries action filters the catalog to queries with an outcome (78 -> 67 surfaced; 11 inventory hidden).
- Existing tests that used AWSAccount as a generic node updated to AWSRole; new tests cover account stripping and outcome passthrough.
This commit is contained in:
Hugo P.Brito committed 2026-05-26 09:10:58 +02:00
1 parent 6f01041178
commit 99a11ecfb6
7 files changed
+166 -6

No files matched your search

@@ -1,5 +1,6 @@
from api.attack_paths.queries.types import (
AttackPathsQueryDefinition,
AttackPathsQueryOutcome,
AttackPathsQueryParameterDefinition,
)
from api.attack_paths.queries.registry import (
@@ -10,6 +11,7 @@ from api.attack_paths.queries.registry import (
__all__ = [
"AttackPathsQueryDefinition",
"AttackPathsQueryOutcome",
"AttackPathsQueryParameterDefinition",
"get_queries_for_provider",
"get_query_by_id",
@@ -1,6 +1,7 @@
from api.attack_paths.queries.types import (
AttackPathsQueryAttribution,
AttackPathsQueryDefinition,
AttackPathsQueryOutcome,
AttackPathsQueryParameterDefinition,
)
from tasks.jobs.attack_paths.config import PROWLER_FINDING_LABEL
@@ -3809,3 +3810,43 @@ AWS_QUERIES: list[AttackPathsQueryDefinition] = [
AWS_SSM_PRIVESC_SEND_COMMAND,
AWS_STS_PRIVESC_ASSUME_ROLE,
]
# Attack outcomes
# ---------------
# The end impact of each real attack path, rendered as a terminal "Outcome" node
# in the graph. Outcomes are assigned by query id pattern so the catalog stays in
# one place. Inventory queries (resource listings, misconfiguration lookups) have
# no attack chain and intentionally keep `outcome = None`.
_OUTCOME_CODE_EXECUTION = AttackPathsQueryOutcome(
label="Code execution",
description="Run arbitrary code with the privileges of the role available to the compute service.",
severity="high",
)
_OUTCOME_PRIVILEGE_ESCALATION = AttackPathsQueryOutcome(
label="Privilege escalation",
description="Obtain higher privileges by altering IAM policies or trust, or by assuming a more privileged role.",
severity="high",
)
_OUTCOME_DATA_EXFILTRATION = AttackPathsQueryOutcome(
label="Data exfiltration",
description="Reach and read sensitive data starting from an internet-exposed entry point.",
severity="critical",
)
def _resolve_outcome(query_id: str) -> AttackPathsQueryOutcome | None:
if query_id == "aws-internet-exposed-ec2-sensitive-s3-access":
return _OUTCOME_DATA_EXFILTRATION
if query_id == "aws-sts-privesc-assume-role" or query_id.startswith(
"aws-iam-privesc-"
):
return _OUTCOME_PRIVILEGE_ESCALATION
if "-privesc-" in query_id:
return _OUTCOME_CODE_EXECUTION
return None
for _query in AWS_QUERIES:
_query.outcome = _resolve_outcome(_query.id)
@@ -9,6 +9,20 @@ class AttackPathsQueryAttribution:
link: str
@dataclass
class AttackPathsQueryOutcome:
"""
Describes the end impact of an attack path (the result of the chain).
Rendered as a terminal "Outcome" node in the graph so the visualization
shows not just the resources involved but what an attacker achieves.
"""
label: str # Short node label, e.g. "Code execution"
description: str # One-line impact, no permission jargon
severity: str = "high" # critical|high|medium - drives outcome node color
@dataclass
class AttackPathsQueryParameterDefinition:
"""
@@ -36,4 +50,5 @@ class AttackPathsQueryDefinition:
provider: str
cypher: str
attribution: AttackPathsQueryAttribution | None = None
outcome: AttackPathsQueryOutcome | None = None
parameters: list[AttackPathsQueryParameterDefinition] = field(default_factory=list)
@@ -1,5 +1,6 @@
import logging
from dataclasses import asdict
from typing import Any, Iterable
import neo4j
@@ -27,6 +28,22 @@ from tasks.jobs.attack_paths.config import (
logger = logging.getLogger(BackendLogger.API)
# Account/root nodes are dropped from the serialized graph: every resource is
# already scoped to a single account, so the account node only adds a hub of
# noise edges. It remains the MATCH anchor in Cypher (isolation is unaffected),
# we only omit it from the output.
ACCOUNT_NODE_LABELS = frozenset(
{
"AWSAccount",
"AzureSubscription",
"AzureTenant",
"GCPProject",
"KubernetesCluster",
"GitHubAccount",
}
)
# Predefined query helpers
@@ -110,7 +127,11 @@ def execute_query(
cypher=definition.cypher,
parameters=parameters,
)
return _serialize_graph(graph, provider_id)
result = _serialize_graph(graph, provider_id)
result["outcome"] = (
asdict(definition.outcome) if definition.outcome else None
)
return result
except graph_database.WriteQueryNotAllowedException:
raise PermissionDenied(
@@ -239,6 +260,11 @@ def _serialize_graph(graph, provider_id: str) -> dict[str, Any]:
if provider_label not in node.labels:
continue
# Drop account/root nodes (and, below, their relationships): they only
# add a noisy hub. Isolation still relies on the account as MATCH anchor.
if ACCOUNT_NODE_LABELS.intersection(node.labels):
continue
kept_node_ids.add(node.element_id)
nodes.append(
{
+61 -4
View File
@@ -110,7 +110,7 @@ def test_execute_query_serializes_graph(
plabel = get_provider_label(provider_id)
node = attack_paths_graph_stub_classes.Node(
element_id="node-1",
labels=["AWSAccount", plabel],
labels=["AWSRole", plabel],
properties={
"name": "account",
"complex": {
@@ -153,6 +153,8 @@ def test_execute_query_serializes_graph(
assert result["nodes"][0]["id"] == "node-1"
assert result["nodes"][0]["properties"]["complex"]["items"][0] == "value"
assert result["relationships"][0]["label"] == "OWNS"
# No outcome defined on this query definition
assert result["outcome"] is None
def test_execute_query_wraps_graph_errors(
@@ -216,9 +218,9 @@ def test_serialize_graph_filters_by_provider_label(attack_paths_graph_stub_class
plabel = get_provider_label(provider_id)
other_label = get_provider_label("provider-other")
node_keep = attack_paths_graph_stub_classes.Node("n1", ["AWSAccount", plabel], {})
node_keep = attack_paths_graph_stub_classes.Node("n1", ["AWSRole", plabel], {})
node_drop = attack_paths_graph_stub_classes.Node(
"n2", ["AWSAccount", other_label], {}
"n2", ["AWSRole", other_label], {}
)
rel_keep = attack_paths_graph_stub_classes.Relationship(
@@ -242,6 +244,61 @@ def test_serialize_graph_filters_by_provider_label(attack_paths_graph_stub_class
assert result["relationships"][0]["id"] == "r1"
def test_serialize_graph_drops_account_nodes(attack_paths_graph_stub_classes):
provider_id = "provider-keep"
plabel = get_provider_label(provider_id)
account = attack_paths_graph_stub_classes.Node("acc", ["AWSAccount", plabel], {})
resource = attack_paths_graph_stub_classes.Node("res", ["AWSRole", plabel], {})
# Account hub edge — must be dropped along with the account node
account_edge = attack_paths_graph_stub_classes.Relationship(
"r-acc", "RESOURCE", account, resource, {}
)
graph = SimpleNamespace(
nodes=[account, resource], relationships=[account_edge]
)
result = views_helpers._serialize_graph(graph, provider_id)
assert [node["id"] for node in result["nodes"]] == ["res"]
assert result["relationships"] == []
def test_execute_query_includes_outcome(
attack_paths_query_definition_factory, attack_paths_graph_stub_classes
):
from api.attack_paths.queries.types import AttackPathsQueryOutcome
outcome = AttackPathsQueryOutcome(
label="Code execution", description="Run code", severity="high"
)
definition = attack_paths_query_definition_factory(
id="aws-ec2-privesc-passrole-iam", outcome=outcome, parameters=[]
)
provider_id = "test-provider-123"
plabel = get_provider_label(provider_id)
node = attack_paths_graph_stub_classes.Node("n1", ["AWSRole", plabel], {})
graph_result = MagicMock()
graph_result.nodes = [node]
graph_result.relationships = []
with patch(
"api.attack_paths.views_helpers.graph_database.execute_read_query",
return_value=graph_result,
):
result = views_helpers.execute_query(
"db-tenant-x", definition, {"provider_uid": "123"}, provider_id=provider_id
)
assert result["outcome"] == {
"label": "Code execution",
"description": "Run code",
"severity": "high",
}
# -- serialize_graph_as_text -------------------------------------------------------
@@ -445,7 +502,7 @@ def test_execute_custom_query_serializes_graph(
):
provider_id = "test-provider-123"
plabel = get_provider_label(provider_id)
node_1 = attack_paths_graph_stub_classes.Node("node-1", ["AWSAccount", plabel], {})
node_1 = attack_paths_graph_stub_classes.Node("node-1", ["AWSRole", plabel], {})
node_2 = attack_paths_graph_stub_classes.Node("node-2", ["RDSInstance", plabel], {})
relationship = attack_paths_graph_stub_classes.Relationship(
"rel-1", "OWNS", node_1, node_2, {}
+11
View File
@@ -1217,12 +1217,22 @@ class AttackPathsQueryParameterSerializer(BaseSerializerV1):
resource_name = "attack-paths-query-parameters"
class AttackPathsQueryOutcomeSerializer(BaseSerializerV1):
label = serializers.CharField()
description = serializers.CharField()
severity = serializers.CharField()
class JSONAPIMeta:
resource_name = "attack-paths-query-outcomes"
class AttackPathsQuerySerializer(BaseSerializerV1):
id = serializers.CharField()
name = serializers.CharField()
short_description = serializers.CharField()
description = serializers.CharField()
attribution = AttackPathsQueryAttributionSerializer(allow_null=True, required=False)
outcome = AttackPathsQueryOutcomeSerializer(allow_null=True, required=False)
provider = serializers.CharField()
parameters = AttackPathsQueryParameterSerializer(many=True)
@@ -1270,6 +1280,7 @@ class AttackPathsRelationshipSerializer(BaseSerializerV1):
class AttackPathsQueryResultSerializer(BaseSerializerV1):
nodes = AttackPathsNodeSerializer(many=True)
relationships = AttackPathsRelationshipSerializer(many=True)
outcome = AttackPathsQueryOutcomeSerializer(allow_null=True, required=False)
total_nodes = serializers.IntegerField()
truncated = serializers.BooleanField()
+9 -1
View File
@@ -2812,7 +2812,15 @@ class AttackPathsScanViewSet(BaseRLSViewSet):
)
def attack_paths_queries(self, request, pk=None):
attack_paths_scan = self.get_object()
queries = get_queries_for_provider(attack_paths_scan.provider.provider)
# Only surface real attack paths (those with a defined outcome). Inventory
# / posture queries without an outcome are hidden from the catalog.
queries = [
query
for query in get_queries_for_provider(
attack_paths_scan.provider.provider
)
if query.outcome is not None
]
if not queries:
return Response(