ReachabilityAnalyzer

Walks a GraphBuilder graph from the internet along the edges traffic can actually take, and reports what it reaches with stable finding ids.

Last changed v0.6.0

cloudg run calls ReachabilityAnalyzer(graph).generate_findings() right after it builds the graph, and the results show up in findings.json with source_tool set to cloudg-reachability. The class answers two different questions, and they follow different edges:

• What can the internet reach? find_internet_exposed() walks network-flow edges only, with the rules below.
• What could a compromised resource reach? compute_blast_radius(node) follows every edge, identity included, and stops at the internet placeholders.

The analyzer works on the graph you pass in and changes it: find_internet_exposed() sets is_internet_exposed=True on every node it reaches. Pass graph.copy() if you need the original untouched (the MCP layer does).

The network-flow walk

Release 0.6.0 changed the walk. Before it, the analysis followed every edge type out of an exposed node, so a database whose only link to the internet was an IAM role granted access to it came out as a CRITICAL "Internet-exposed RDS_INSTANCE". Now an edge is followed only when traffic can travel along it, and in the direction it travels.

flowchart LR
  I["0.0.0.0/0, ::/0, Internet tag"] -->|"SG or NACL rule, ingress"| G["Security group, NSG, NACL"]
  G -->|"ATTACHED_TO, walked backwards"| R["Resource in the group"]
  I -->|"INTERNET_EXPOSED"| P["Public resource"]
  R -->|"LOAD_BALANCER_TARGET"| T["Target group"]
  T -->|"LOAD_BALANCER_TARGET"| W["Targets"]
  E["ENI or Elastic IP"] -->|"ATTACHED_TO"| V["Instance"]
  N["VPC, VNet, subnet"] -->|"CONTAINS"| M["Resource placed in it"]
  S["Route or peering source"] -->|"ROUTE, PEERING"| D["Destination"]

Edges the exposure walk follows, in the direction traffic moves

Edge type	Followed	Direction
INTERNET_EXPOSED, LOAD_BALANCER_TARGET, ROUTE, PEERING	always	source to target
SECURITY_GROUP_RULE, NACL_RULE	unless direction is egress	source to target
ATTACHED_TO	when the target is a SECURITY_GROUP, NSG or NACL, or the edge's relationship is PROTECTED_BY_SG or PROTECTED_BY_NACL	backwards, from the group to the resource attached to it
ATTACHED_TO	when the source is a NETWORK_INTERFACE or ELASTIC_IP	source to target
CONTAINS	only from a VPC, VNET or SUBNET	parent to child
IAM_TRUST, IAM_POLICY_ATTACHMENT, ASSUMES_ROLE, GRANTS_ACCESS, INVOKES, REFERENCES, USES_IMAGE, LOGS_TO, PROTECTS, MONITORS, MANAGES, GOVERNS	never

The backwards ATTACHED_TO hop is the one people miss. Collectors and the linker write web-1 ATTACHED_TO sg-web, pointing from the resource to its group, while traffic flows the other way: whatever the group admits reaches web-1.

INVOKES is deliberately not followed, so a Lambda function behind a public API Gateway is not reported as exposed. The gateway is the exposed resource; the function is only invoked by it. Disk, route-table and gateway attachments are not traffic paths either, and organization, account, cluster and namespace containment are not traversed.

Where the walk starts: every node named 0.0.0.0/0 or ::/0, plus the source of every non-egress edge whose cidr stands for the whole internet. That test (cloudg.graph.ports.is_internet_source) also accepts the Azure NSG service tags Internet, Any and * in any letter case. Private CIDRs such as 10.0.0.0/8, unresolved references and a group with an egress rule to 0.0.0.0/0 are not entry points.

The walk over-approximates. It does not intersect ports across hops, and a rule whose source is another security group is followed like any other ingress rule, so a database whose group admits the web tier's group is reported exposed once the web tier is. Treat a finding as "there is a network path", then check the ports.

What generate_findings() reports

Rule (finding_id input)	Raised for	Severity	Frameworks
internet-exposed-sensitive-asset	a reached RDS_INSTANCE, AURORA_CLUSTER, AZURE_SQL, CLOUD_SQL or DYNAMODB_TABLE	CRITICAL	CIS, NIST-800-53
internet-exposed-unexpected-asset	any other reached resource, except load balancers, CloudFront, CDNs and internet gateways	HIGH	none
internet-open-sensitive-port:<port>	a non-egress edge from an internet source whose ports include one of 22, 3389, 3306, 5432, 1433, 27017, 6379, 9200, 5601, 8080 or 8443	CRITICAL	CIS, NIST-800-53, PCI-DSS

Security groups, NSGs, NACLs and target groups are hops: the walk marks them exposed, but they get no exposure finding of their own. A group's open rules show up as open-port findings instead, with the group as resource_id, and a target group's exposure shows up on the targets behind it. Placeholder nodes (is_external) never get an exposure finding.

Ports are compared as numbers after cloudg.graph.ports parses port_range, so 0-65535 contains 22, 2200-2300 does not, and an Azure * covers all eleven ports. An edge yields at most one finding per sensitive port, however many of its merged rules overlap it.

Deterministic ids

Each finding id is a UUID5 of the rule name and a stable key, so the same exposure gets the same id on every scan. You can dedupe, track and suppress reachability findings across runs.

finding_id(rule, asset_key(graph, node_id))                     # exposure findings
finding_id(f"{RULE_OPEN_PORT}:{port}", f"{src_key}->{dst_key}") # open-port findings

The key is the node's ARN, or the node id when there is no ARN (CIDR placeholders). The graph node id itself would not do: collectors give assets a fresh random UUID on every collection. Finding.resource_id is still that per-run node id, so match findings to assets of a later run by resource_arn.

Examples

Exposure, findings and blast radius

Ten assets, fourteen edges, no credentials. The estate has a public load balancer in front of web-1, an SSH rule left open on the web tier's group, a database reachable only from a private subnet and through an IAM grant, and a Lambda function behind a public API Gateway.

exposure.py

"""Internet exposure, flow hops, finding ids and blast radius on a small estate."""

from cloudg.graph.builder import GraphBuilder
from cloudg.graph.reachability import (
    RULE_UNEXPECTED_EXPOSURE,
    ReachabilityAnalyzer,
    asset_key,
    finding_id,
    network_flow_graph,
)
from cloudg.schema.models import AssetType, CloudAsset, CloudProvider, EdgeType, NetworkEdge

ACCOUNT = "123456789012"
REGION = "eu-west-1"

def asset(asset_id: str, asset_type: AssetType, arn: str) -> CloudAsset:
    return CloudAsset(
        id=asset_id,
        name=asset_id,
        asset_type=asset_type,
        provider=CloudProvider.AWS,
        region=REGION,
        account_id=ACCOUNT,
        arn=arn,
    )

assets = [
    asset("sg-alb", AssetType.SECURITY_GROUP, f"arn:aws:ec2:{REGION}:{ACCOUNT}:security-group/sg-alb"),
    asset("sg-web", AssetType.SECURITY_GROUP, f"arn:aws:ec2:{REGION}:{ACCOUNT}:security-group/sg-web"),
    asset("sg-db", AssetType.SECURITY_GROUP, f"arn:aws:ec2:{REGION}:{ACCOUNT}:security-group/sg-db"),
    asset("web-alb", AssetType.LOAD_BALANCER, f"arn:aws:elasticloadbalancing:{REGION}:{ACCOUNT}:loadbalancer/app/web-alb/1"),
    asset("web-tg", AssetType.TARGET_GROUP, f"arn:aws:elasticloadbalancing:{REGION}:{ACCOUNT}:targetgroup/web-tg/1"),
    asset("web-1", AssetType.EC2, f"arn:aws:ec2:{REGION}:{ACCOUNT}:instance/i-0web1"),
    asset("app-role", AssetType.IAM_ROLE, f"arn:aws:iam::{ACCOUNT}:role/app-role"),
    asset("orders-db", AssetType.RDS_INSTANCE, f"arn:aws:rds:{REGION}:{ACCOUNT}:db:orders-db"),
    asset("orders-api", AssetType.API_GATEWAY, f"arn:aws:apigateway:{REGION}::/restapis/a1b2c3"),
    asset("orders-fn", AssetType.LAMBDA_FUNCTION, f"arn:aws:lambda:{REGION}:{ACCOUNT}:function:orders-fn"),
]

def ingress(source: str, target: str, ports: str, cidr: str | None = None) -> NetworkEdge:
    return NetworkEdge(
        source_id=source,
        target_id=target,
        edge_type=EdgeType.SECURITY_GROUP_RULE,
        port_range=ports,
        protocol="TCP",
        cidr=cidr,
        direction="ingress",
    )

def edge(source: str, target: str, edge_type: EdgeType, **extra) -> NetworkEdge:
    return NetworkEdge(source_id=source, target_id=target, edge_type=edge_type, **extra)

edges = [
    ingress("0.0.0.0/0", "sg-alb", "443", cidr="0.0.0.0/0"),
    ingress("0.0.0.0/0", "sg-web", "22", cidr="0.0.0.0/0"),  # SSH left open
    ingress("sg-alb", "sg-web", "8080"),
    ingress("10.0.1.0/24", "sg-db", "5432", cidr="10.0.1.0/24"),  # private source
    # The default egress rule points from the group to the internet
    edge("sg-web", "0.0.0.0/0", EdgeType.SECURITY_GROUP_RULE, port_range="0-65535",
         protocol="ALL", cidr="0.0.0.0/0", direction="egress"),
    edge("web-alb", "sg-alb", EdgeType.ATTACHED_TO, relationship="PROTECTED_BY_SG"),
    edge("web-1", "sg-web", EdgeType.ATTACHED_TO, relationship="PROTECTED_BY_SG"),
    edge("orders-db", "sg-db", EdgeType.ATTACHED_TO, relationship="PROTECTED_BY_SG"),
    edge("web-alb", "web-tg", EdgeType.LOAD_BALANCER_TARGET),
    edge("web-tg", "web-1", EdgeType.LOAD_BALANCER_TARGET),
    edge("web-1", "app-role", EdgeType.ASSUMES_ROLE),
    edge("app-role", "orders-db", EdgeType.GRANTS_ACCESS),
    edge("0.0.0.0/0", "orders-api", EdgeType.INTERNET_EXPOSED, cidr="0.0.0.0/0"),
    edge("orders-api", "orders-fn", EdgeType.INVOKES),
]

graph = GraphBuilder().build(assets, edges)
analyzer = ReachabilityAnalyzer(graph)

print("entry points:", sorted(analyzer.internet_entry_points()))
print("exposed:", sorted(analyzer.find_internet_exposed()))

print("hops out of sg-web:")
for nxt, data, walked_back in analyzer.flow_hops("sg-web"):
    print(f"  -> {nxt} via {data['edge_type']}{' (reversed)' if walked_back else ''}")

for finding in analyzer.generate_findings():
    print(f"{finding.severity.value:8} {finding.title}  id={finding.id[:8]}")

# The id can be recomputed from the rule name and the asset's ARN
expected = finding_id(RULE_UNEXPECTED_EXPOSURE, asset_key(graph, "web-1"))
print("web-1 exposure id:", expected[:8])

flow = network_flow_graph(graph)
print("flow edges:", flow.number_of_edges(), "of", graph.number_of_edges())
print("flow reversed:", sorted((u, v) for u, v, d in flow.edges(data=True) if d["reversed"]))

radius = analyzer.compute_blast_radius("web-1")
print("blast radius of web-1:", sorted(radius["reachable_nodes"]),
      "depth", radius["depth"], "risk", radius["risk_score"])

$ python exposure.py
entry points: ['0.0.0.0/0']
exposed: ['orders-api', 'sg-alb', 'sg-web', 'web-1', 'web-alb', 'web-tg']
hops out of sg-web:
  -> web-1 via ATTACHED_TO (reversed)
HIGH     Unexpected internet-exposed resource: orders-api  id=b5fbe235
HIGH     Unexpected internet-exposed resource: web-1  id=4ec3a339
CRITICAL Security group allows SSH (port 22) from 0.0.0.0/0  id=17cd5168
web-1 exposure id: 4ec3a339
flow edges: 10 of 14
flow reversed: [('sg-alb', 'web-alb'), ('sg-db', 'orders-db'), ('sg-web', 'web-1')]
blast radius of web-1: ['0.0.0.0/0', 'app-role', 'orders-db', 'sg-db', 'sg-web'] depth 3 risk 4.5

What the output shows:

• 10.0.1.0/24 and the group with the egress rule are not entry points. orders-db is not exposed: its only inbound rule comes from a private CIDR, and the IAM path through app-role does not count.
• orders-fn is not exposed, because INVOKES is not a network hop. The API Gateway is reported instead.
• web-alb is reached but gets no finding (load balancers are expected to face the internet), and the groups and the target group are hops only.
• The ids are the same on every run. Run the script twice and compare.
• The only edge leaving sg-web in the graph is its egress rule, which the walk skips. Its one flow hop is the attachment from web-1, walked backwards.

The blast radius is a different, wider question. From web-1 it follows every edge type, so the IAM edges lead it to orders-db. It also follows the egress rule to the 0.0.0.0/0 placeholder but stops there: being able to send traffic to the internet does not put everything the internet reaches in the blast radius. Read it as an upper bound all the same.

Flow-aware paths with NetworkX

network_flow_graph() returns a new DiGraph with an edge u -> v exactly where the exposure walk would hop, so ordinary NetworkX path functions follow traffic. Save this next to exposure.py; the import runs that script first, so its output is printed too.

flow_paths.py

import networkx as nx

from cloudg.graph.reachability import network_flow_graph
from exposure import graph

flow = network_flow_graph(graph)
for path in nx.all_simple_paths(flow, "0.0.0.0/0", "web-1", cutoff=6):
    hops = [flow.edges[u, v]["edge_type"] + (" (rev)" if flow.edges[u, v]["reversed"] else "")
            for u, v in zip(path, path[1:])]
    print(" -> ".join(path), "|", ", ".join(hops))

$ python flow_paths.py
...
0.0.0.0/0 -> sg-alb -> sg-web -> web-1 | SECURITY_GROUP_RULE, SECURITY_GROUP_RULE, ATTACHED_TO (rev)
0.0.0.0/0 -> sg-alb -> web-alb -> web-tg -> web-1 | SECURITY_GROUP_RULE, ATTACHED_TO (rev), LOAD_BALANCER_TARGET, LOAD_BALANCER_TARGET
0.0.0.0/0 -> sg-web -> web-1 | SECURITY_GROUP_RULE, ATTACHED_TO (rev)

The first path is the over-approximation at work: sg-alb admits 443 from the internet and sg-web admits 8080 from sg-alb, and the walk does not check that the ports line up.

Notes

• compute_blast_radius() returns reachable_nodes (every node a breadth-first walk over every edge type reaches), depth (the longest shortest-path distance) and risk_score: 0.5 per reachable node, plus 2.0 for each reachable RDS_INSTANCE, AURORA_CLUSTER, AZURE_SQL, CLOUD_SQL or DYNAMODB_TABLE, capped at 10.0. The walk reaches the internet placeholders 0.0.0.0/0 and ::/0 but does not continue through them; only a walk that starts at a placeholder follows its edges. An unknown node gives {"reachable_nodes": [], "depth": 0, "risk_score": 0.0}.
• flow_hops() can yield the same next node more than once when several edges lead there, and yields nothing for an unknown node. flow_successors() is the same walk without the edge data.
• network_flow_graph() copies every node with its attributes and leaves the input graph alone. Each edge carries edge_type and reversed.
• Graphs built by hand can carry a ports list on an edge, and the open-port check reads it next to port_range. GraphBuilder never copies NetworkEdge.ports onto the graph.
• The open-port title keeps the historical 0.0.0.0/0 wording for both 0.0.0.0/0 and ::/0, and names the Azure tag as written (Internet, Any, *) otherwise.
• Stored findings from 0.5.x carry random ids. They will not match the 0.6.0 ids; re-baseline suppressions after upgrading.

Reference

ReachabilityAnalyzer

ReachabilityAnalyzer(graph: DiGraph)

Analyses network reachability via BFS on the infrastructure graph.

1. Find all nodes reachable from the internet over network-flow edges ("internet-exposed"); see the module docstring for the traversal rules.
2. From any node, BFS over every edge type to find what it can reach ("blast radius"), stopping at the internet placeholders.
3. Cross-reference with asset type to generate severity-scored findings.

Source code in cloudg/graph/reachability.py

def __init__(self, graph: nx.DiGraph) -> None:
    self._graph = graph

internet_entry_points

internet_entry_points() -> set[str]

Nodes that represent the internet as a traffic source.

Nodes named 0.0.0.0/0 or ::/0, plus the source of every non-egress edge whose cidr passes cloudg.graph.ports.is_internet_source (which adds the Azure service tags Internet, Any and *).

Source code in cloudg/graph/reachability.py

def internet_entry_points(self) -> set[str]:
    """Nodes that represent the internet as a traffic source.

    Nodes named ``0.0.0.0/0`` or ``::/0``, plus the source of every
    non-egress edge whose ``cidr`` passes
    :func:`~cloudg.graph.ports.is_internet_source` (which adds the Azure
    service tags ``Internet``, ``Any`` and ``*``).
    """
    entry_points = {
        node_id
        for node_id, data in self._graph.nodes(data=True)
        if data.get("name", node_id) in INTERNET_CIDRS
    }
    for source, _target, data in self._graph.edges(data=True):
        if is_internet_source(data.get("cidr")) and not is_egress(data):
            entry_points.add(source)
    return entry_points

find_internet_exposed

find_internet_exposed() -> set[str]

Find all nodes reachable from the internet over network-flow edges.

Entry points are nodes named 0.0.0.0/0 or ::/0 and the source of every non-egress edge whose CIDR stands for the internet. The walk follows only the network-flow hops described in the module docstring, so resources reached only through IAM or other non-network edges are not reported. Every node returned is marked is_internet_exposed in the graph, rule containers and target groups included.

Returns:

Type	Description
set[str]	Set of node IDs that are internet-exposed.

Source code in cloudg/graph/reachability.py

def find_internet_exposed(self) -> set[str]:
    """Find all nodes reachable from the internet over network-flow edges.

    Entry points are nodes named ``0.0.0.0/0`` or ``::/0`` and the source
    of every non-egress edge whose CIDR stands for the internet. The walk
    follows only the network-flow hops described in the module docstring,
    so resources reached only through IAM or other non-network edges are
    not reported. Every node returned is marked ``is_internet_exposed``
    in the graph, rule containers and target groups included.

    Returns:
        Set of node IDs that are internet-exposed.
    """
    exposed: set[str] = set()
    queue = deque(self.internet_entry_points())
    seen = set(queue)
    while queue:
        node_id = queue.popleft()
        for nxt in self.flow_successors(node_id):
            exposed.add(nxt)
            if nxt not in seen:
                seen.add(nxt)
                queue.append(nxt)

    # Mark nodes in graph
    for node_id in exposed:
        self._graph.nodes[node_id]["is_internet_exposed"] = True

    logger.info("Found %d internet-exposed nodes", len(exposed))
    return exposed

flow_hops

flow_hops(node_id: str) -> Iterator[FlowHop]

The network-flow hops out of node_id, with the edge each uses.

Yields (next_node, edge_data, reversed): out-edges that carry traffic forward (reversed False) and ATTACHED_TO in-edges walked back from a traffic filter to the resource attached to it (reversed True). A node may appear more than once when several edges lead to it. Unknown nodes yield nothing.

Source code in cloudg/graph/reachability.py

def flow_hops(self, node_id: str) -> Iterator[FlowHop]:
    """The network-flow hops out of ``node_id``, with the edge each uses.

    Yields ``(next_node, edge_data, reversed)``: out-edges that carry
    traffic forward (``reversed`` False) and ``ATTACHED_TO`` in-edges
    walked back from a traffic filter to the resource attached to it
    (``reversed`` True). A node may appear more than once when several
    edges lead to it. Unknown nodes yield nothing.
    """
    if node_id not in self._graph:
        return
    node_type = self._graph.nodes[node_id].get("asset_type", "")
    for _source, target, data in self._graph.out_edges(node_id, data=True):
        if _is_forward_hop(data, node_type):
            yield target, data, False
    # Traffic admitted by a security group / NSG / NACL reaches the
    # resources attached to it: walk those ATTACHED_TO edges backwards.
    is_filter = node_type in _TRAFFIC_FILTER_TYPES
    for source, _target, data in self._graph.in_edges(node_id, data=True):
        if _is_reverse_hop(data, is_filter):
            yield source, data, True

flow_successors

flow_successors(node_id: str) -> Iterator[str]

Nodes that traffic arriving at node_id can travel on to.

This is the single-step form of the exposure walk; the rules are in the module docstring. See flow_hops for the edges used and network_flow_graph for the whole flow graph at once.

Source code in cloudg/graph/reachability.py

def flow_successors(self, node_id: str) -> Iterator[str]:
    """Nodes that traffic arriving at ``node_id`` can travel on to.

    This is the single-step form of the exposure walk; the rules are in
    the module docstring. See :meth:`flow_hops` for the edges used and
    :func:`network_flow_graph` for the whole flow graph at once.
    """
    for nxt, _data, _reversed in self.flow_hops(node_id):
        yield nxt

generate_findings

generate_findings() -> list[Finding]

Generate security findings based on reachability analysis.

Returns:

Type	Description
list[Finding]	List of Finding objects for reachability-related security issues.

Source code in cloudg/graph/reachability.py

def generate_findings(self) -> list[Finding]:
    """Generate security findings based on reachability analysis.

    Returns:
        List of Finding objects for reachability-related security issues.
    """
    findings: list[Finding] = []

    # Findings for internet-exposed resources, in a stable order
    for node_id in sorted(self.find_internet_exposed()):
        finding = self._exposure_finding(node_id)
        if finding:
            findings.append(finding)

    # Findings for sensitive port exposure on security group edges
    findings.extend(self._sensitive_port_findings())

    logger.info("Generated %d reachability findings", len(findings))
    return findings

compute_blast_radius

compute_blast_radius(node_id: str) -> dict[str, Any]

Compute blast radius: all nodes reachable from a given node.

A breadth-first walk over every out-edge, identity and inventory edges included. The internet placeholders (0.0.0.0/0, ::/0) are reached but not walked through: an egress rule to the internet means the node can send traffic out, not that every resource the internet can reach is in its blast radius. Starting the walk at a placeholder still follows its edges.

Parameters:

Name	Type	Description	Default
node_id	str	Starting node for BFS.	required

Returns:

Type	Description
dict[str, Any]	Dict with 'reachable_nodes', 'depth', and 'risk_score'.

Source code in cloudg/graph/reachability.py

def compute_blast_radius(self, node_id: str) -> dict[str, Any]:
    """Compute blast radius: all nodes reachable from a given node.

    A breadth-first walk over every out-edge, identity and inventory
    edges included. The internet placeholders (``0.0.0.0/0``,
    ``::/0``) are reached but not walked through: an egress rule to
    the internet means the node can send traffic out, not that every
    resource the internet can reach is in its blast radius. Starting
    the walk at a placeholder still follows its edges.

    Args:
        node_id: Starting node for BFS.

    Returns:
        Dict with 'reachable_nodes', 'depth', and 'risk_score'.
    """
    if node_id not in self._graph:
        return {"reachable_nodes": [], "depth": 0, "risk_score": 0.0}

    distance = {node_id: 0}
    queue = deque([node_id])
    while queue:
        current = queue.popleft()
        if current != node_id and self._is_internet_node(current):
            continue  # stop at the internet placeholder
        for nxt in self._graph.successors(current):
            if nxt not in distance:
                distance[nxt] = distance[current] + 1
                queue.append(nxt)
    reachable = set(distance) - {node_id}
    max_depth = max(distance.values())

    # Risk score based on number of reachable nodes and their types
    risk_score = min(10.0, len(reachable) * 0.5)

    # Increase risk if sensitive assets are reachable
    for r_node in reachable:
        node_data = self._graph.nodes.get(r_node, {})
        asset_type_str = node_data.get("asset_type", "")
        try:
            asset_type = AssetType(asset_type_str)
            if asset_type in SENSITIVE_ASSET_TYPES:
                risk_score = min(10.0, risk_score + 2.0)
        except ValueError:
            pass

    return {
        "reachable_nodes": list(reachable),
        "depth": max_depth,
        "risk_score": round(risk_score, 1),
    }

finding_id

finding_id(rule: str, asset: str) -> str

Deterministic finding id: a UUID5 hash of the rule and the asset key.

The same rule firing on the same asset gets the same id on every run, so reachability findings can be deduplicated and tracked across scans. Pass asset_key of the node, not the graph node id: collectors give assets a fresh random id on every collection.

Source code in cloudg/graph/reachability.py

def finding_id(rule: str, asset: str) -> str:
    """Deterministic finding id: a UUID5 hash of the rule and the asset key.

    The same rule firing on the same asset gets the same id on every run,
    so reachability findings can be deduplicated and tracked across scans.
    Pass :func:`asset_key` of the node, not the graph node id: collectors
    give assets a fresh random id on every collection.
    """
    return str(uuid.uuid5(_FINDING_ID_NAMESPACE, f"{rule}|{asset}"))

network_flow_graph

network_flow_graph(graph: DiGraph) -> DiGraph

The network-flow hops of graph as a new directed graph.

Every node of graph is copied with its attributes. There is an edge u -> v exactly when ReachabilityAnalyzer.flow_successors yields v for u, so walking this graph forward follows the same rules as the internet-exposure analysis (see the module docstring). Each edge carries edge_type (the type of the underlying edge) and reversed (True for an ATTACHED_TO edge walked from the filter back to the resource attached to it). graph is not modified.

Source code in cloudg/graph/reachability.py

def network_flow_graph(graph: nx.DiGraph) -> nx.DiGraph:
    """The network-flow hops of ``graph`` as a new directed graph.

    Every node of ``graph`` is copied with its attributes. There is an edge
    ``u -> v`` exactly when :meth:`ReachabilityAnalyzer.flow_successors`
    yields ``v`` for ``u``, so walking this graph forward follows the same
    rules as the internet-exposure analysis (see the module docstring).
    Each edge carries ``edge_type`` (the type of the underlying edge) and
    ``reversed`` (True for an ``ATTACHED_TO`` edge walked from the filter
    back to the resource attached to it). ``graph`` is not modified.
    """
    analyzer = ReachabilityAnalyzer(graph)
    flow = nx.DiGraph()
    flow.add_nodes_from(graph.nodes(data=True))
    for node_id in graph.nodes:
        for nxt, data, walked_back in analyzer.flow_hops(node_id):
            if not flow.has_edge(node_id, nxt):
                flow.add_edge(
                    node_id, nxt, edge_type=data.get("edge_type", ""), reversed=walked_back
                )
    return flow

Related

• GraphBuilder

  Builds the graph this class walks.

• Dependencies and blast radius

  The inventory-level blast radius behind cloudg deps.

• DependencyGraph

  Dependency walks over a saved map.

• Data models

  The Finding model these results use.

• Graph, ontology and RAG

  Where reachability sits in the analysis phase.