Graph, ontology and RAG
The graph, the ontology and the RAG export all start from the same two lists, CloudAsset and NetworkEdge. These scripts build a small estate by hand, so every one of them is offline: swap the hand-made lists for collection.assets and collection.edges from CloudGEngine.collect(), or for an InventoryResult you loaded, and nothing else changes.
flowchart LR
A["CloudAsset + NetworkEdge"] --> G["GraphBuilder"]
G --> P["attack paths, centrality"]
G --> R["ReachabilityAnalyzer"]
R --> F["exposure findings"]
A --> O["CloudOntology"]
O --> Q["SPARQL, Turtle, JSON-LD"]
A --> X["RAGExporter"]
X --> C["rag_chunks.jsonl"]Build and explore the graph¶
GraphBuilder turns the lists into a NetworkX DiGraph, merges parallel rules between the same pair of nodes, and adds a placeholder node for 0.0.0.0/0.
build_graph.py
"""Build a graph from a hand-made estate and look at it from a few angles."""
import json
from pathlib import Path
import community as community_louvain # python-louvain, a cloudg dependency
from cloudg.graph.builder import GraphBuilder
from cloudg.schema.models import AssetType, CloudAsset, CloudProvider, EdgeType, NetworkEdge
ACCOUNT = "123456789012"
def asset(asset_id: str, name: str, asset_type: AssetType, arn_suffix: str, **extra) -> CloudAsset:
return CloudAsset(
id=asset_id,
name=name,
asset_type=asset_type,
provider=CloudProvider.AWS,
region="eu-west-1",
account_id=ACCOUNT,
arn=f"arn:aws:{arn_suffix}",
**extra,
)
assets = [
asset("sg-alb", "sg-alb", AssetType.SECURITY_GROUP, f"ec2:eu-west-1:{ACCOUNT}:security-group/sg-alb"),
asset("sg-web", "sg-web", AssetType.SECURITY_GROUP, f"ec2:eu-west-1:{ACCOUNT}:security-group/sg-web"),
asset("alb", "web-alb", AssetType.LOAD_BALANCER, f"elasticloadbalancing:eu-west-1:{ACCOUNT}:loadbalancer/app/web-alb/1"),
asset("tg", "web-tg", AssetType.TARGET_GROUP, f"elasticloadbalancing:eu-west-1:{ACCOUNT}:targetgroup/web-tg/1"),
asset("web-1", "web-1", AssetType.EC2, f"ec2:eu-west-1:{ACCOUNT}:instance/i-0web1", is_internet_exposed=True),
asset("web-2", "web-2", AssetType.EC2, f"ec2:eu-west-1:{ACCOUNT}:instance/i-0web2"),
asset("app-role", "app-role", AssetType.IAM_ROLE, f"iam::{ACCOUNT}:role/app-role"),
asset("orders-db", "orders-db", AssetType.RDS_INSTANCE, f"rds:eu-west-1:{ACCOUNT}:db:orders-db"),
]
def rule(source: str, target: str, port: str, cidr: str = "", description: str = "") -> NetworkEdge:
return NetworkEdge(
source_id=source,
target_id=target,
edge_type=EdgeType.SECURITY_GROUP_RULE,
port_range=port,
protocol="TCP",
cidr=cidr or None,
direction="ingress",
description=description or None,
)
edges = [
# Two rules on the same pair: GraphBuilder merges them into one edge
rule("0.0.0.0/0", "sg-alb", "443", "0.0.0.0/0", "HTTPS from anywhere"),
rule("0.0.0.0/0", "sg-alb", "80", "0.0.0.0/0", "HTTP redirect"),
rule("sg-alb", "sg-web", "8080", description="ALB to web tier"),
NetworkEdge(source_id="alb", target_id="sg-alb", edge_type=EdgeType.ATTACHED_TO, relationship="PROTECTED_BY_SG"),
NetworkEdge(source_id="web-1", target_id="sg-web", edge_type=EdgeType.ATTACHED_TO, relationship="PROTECTED_BY_SG"),
NetworkEdge(source_id="web-2", target_id="sg-web", edge_type=EdgeType.ATTACHED_TO, relationship="PROTECTED_BY_SG"),
NetworkEdge(source_id="alb", target_id="tg", edge_type=EdgeType.LOAD_BALANCER_TARGET),
NetworkEdge(source_id="tg", target_id="web-1", edge_type=EdgeType.LOAD_BALANCER_TARGET),
NetworkEdge(source_id="tg", target_id="web-2", edge_type=EdgeType.LOAD_BALANCER_TARGET),
NetworkEdge(source_id="web-1", target_id="app-role", edge_type=EdgeType.IAM_TRUST),
NetworkEdge(source_id="app-role", target_id="orders-db", edge_type=EdgeType.GRANTS_ACCESS),
]
builder = GraphBuilder()
graph = builder.build(assets, edges)
print(f"{graph.number_of_nodes()} nodes, {graph.number_of_edges()} edges from {len(edges)} NetworkEdges")
merged = graph.edges["0.0.0.0/0", "sg-alb"]
print("merged rule:", merged["port_range"], "|", merged["description"])
print("placeholder:", graph.nodes["0.0.0.0/0"]["asset_type"], graph.nodes["0.0.0.0/0"]["is_external"])
def names(path: list[str]) -> str:
return " -> ".join(graph.nodes[n]["name"] for n in path)
# Paths follow edges in their stored direction
for path in builder.find_attack_paths("0.0.0.0/0", "sg-web"):
print("internet path:", names(path))
for path in builder.find_attack_paths("web-1", "orders-db"):
print("identity path:", names(path))
for path in builder.find_lateral_movement_paths():
print("lateral path:", names(path))
centrality = builder.compute_centrality()
top = sorted(centrality.items(), key=lambda kv: kv[1]["betweenness"], reverse=True)[:3]
for node_id, scores in top:
print(f"betweenness {scores['betweenness']:.3f} {graph.nodes[node_id]['name']}")
# Louvain works on undirected graphs; the RAG exporter does the same
partition = community_louvain.best_partition(graph.to_undirected(), random_state=42)
groups: dict[int, list[str]] = {}
for node_id, community_id in partition.items():
groups.setdefault(community_id, []).append(graph.nodes[node_id]["name"])
for community_id, members in sorted(groups.items()):
print(f"community {community_id}: {sorted(members)}")
out = Path("graph-out")
builder.save_graphml(out / "estate.graphml")
(out / "estate-d3.json").write_text(builder.to_json_str())
(out / "estate-cytoscape.json").write_text(json.dumps(builder.to_cytoscape_json(), indent=2))
print("wrote", sorted(p.name for p in out.iterdir()))
$ python build_graph.py
9 nodes, 10 edges from 11 NetworkEdges
merged rule: 443,80 | HTTPS from anywhere; HTTP redirect
placeholder: EXTERNAL True
internet path: 0.0.0.0/0 -> sg-alb -> sg-web
identity path: web-1 -> app-role -> orders-db
lateral path: web-1 -> app-role
betweenness 0.080 web-1
betweenness 0.071 web-tg
betweenness 0.054 app-role
community 0: ['0.0.0.0/0', 'sg-alb']
community 1: ['sg-web', 'web-2']
community 2: ['app-role', 'orders-db', 'web-1']
community 3: ['web-alb', 'web-tg']
wrote ['estate-cytoscape.json', 'estate-d3.json', 'estate.graphml']
cloudg run builds the same graph from a live collection and writes it next to the reports:
graph.persist_graphml (on by default) writes topology.graphml, and graph.export_cytoscape (off by default) adds topology-cytoscape.json. Both are in the graph section of config.yaml.
What can the internet reach¶
ReachabilityAnalyzer walks the graph from the internet along the edges traffic can actually take, and gives each finding an id that stays the same from run to run.
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
Query it with SPARQL¶
CloudOntology writes the estate into an rdflib graph with typed relations, then exports it in four RDF serialisations and answers SPARQL queries.
ontology_demo.py
"""Build the RDF ontology for a small estate, save it and query it with SPARQL."""
from pathlib import Path
from rdflib import Graph
from cloudg.graph.ontology import CloudOntology, RelationGroup
from cloudg.schema.models import (
AssetType,
CloudAsset,
CloudProvider,
EdgeType,
Finding,
NetworkEdge,
Severity,
)
ACCOUNT = "123456789012"
REGION = "eu-west-1"
KEY_ARN = f"arn:aws:kms:{REGION}:{ACCOUNT}:key/1111aaaa-22bb-33cc-44dd-555555eeeeee"
def asset(asset_id: str, asset_type: AssetType, arn: str, **extra) -> CloudAsset:
return CloudAsset(
id=asset_id,
name=asset_id,
asset_type=asset_type,
provider=CloudProvider.AWS,
region=REGION,
account_id=ACCOUNT,
arn=arn,
**extra,
)
assets = [
asset("sg-web", AssetType.SECURITY_GROUP, f"arn:aws:ec2:{REGION}:{ACCOUNT}:security-group/sg-web"),
asset("web-1", AssetType.EC2, f"arn:aws:ec2:{REGION}:{ACCOUNT}:instance/i-0web1",
tags={"owner": "payments"}),
asset("orders-key", AssetType.KMS_KEY, KEY_ARN),
asset("orders-db", AssetType.RDS_INSTANCE, f"arn:aws:rds:{REGION}:{ACCOUNT}:db:orders-db",
metadata={"storage_encrypted": True, "kms_key_id": KEY_ARN}),
]
edges = [
NetworkEdge(source_id="0.0.0.0/0", target_id="sg-web", edge_type=EdgeType.SECURITY_GROUP_RULE,
port_range="22", protocol="TCP", cidr="0.0.0.0/0", direction="ingress"),
NetworkEdge(source_id="web-1", target_id="sg-web", edge_type=EdgeType.ATTACHED_TO),
]
# A scanner finding that names the database by ARN, not by asset id
findings = [
Finding(
resource_id=f"arn:aws:rds:{REGION}:{ACCOUNT}:db:orders-db",
severity=Severity.HIGH,
title="RDS instance has no deletion protection",
description="Deletion protection is disabled.",
source_tool="prowler",
compliance_frameworks=["CIS"],
)
]
onto = CloudOntology()
onto.build(assets, edges, findings)
stats = onto.stats()
print(stats["total_triples"], "triples,", stats["classes_used"], "classes,",
stats["individuals"], "individuals")
print("groups:", dict(sorted(stats["relation_group_counts"].items())))
# The graph binds the cm:, cmr: and cmp: prefixes, so queries can use them
rows = onto.query("""
SELECT ?subject ?relation ?object WHERE {
?s ?p ?o .
FILTER(STRSTARTS(STR(?p), STR(cmp:)))
FILTER(!STRSTARTS(STR(?p), STR(cmp:has)))
FILTER(?p != cmp:isInternetExposed)
?s cmp:hasName ?subject .
OPTIONAL { ?o cmp:hasName ?oname }
BIND(STRAFTER(STR(?p), STR(cmp:)) AS ?relation)
BIND(COALESCE(?oname, STR(?o)) AS ?object)
} ORDER BY ?subject ?relation
""")
for row in rows:
print(f" {row['subject']} {row['relation']} {row['object']}")
print("internet reachable:", [r["name"] for r in onto.query_internet_exposed()])
print("security group:", len(onto.query_by_relation_group(RelationGroup.SECURITY)), "triples")
print("compliance gaps:", [(r["severity"], r["resourceName"]) for r in onto.query_compliance_gaps()])
out = Path("ontology-out")
for fmt, name in [("turtle", "ontology.ttl"), ("json-ld", "ontology.jsonld"),
("xml", "ontology.rdf"), ("nt", "ontology.nt")]:
onto.save(out / name, fmt=fmt)
# Any RDF tool can read the files back; here rdflib itself
reloaded = Graph().parse(out / "ontology.ttl", format="turtle")
print("reloaded", len(reloaded), "triples from", sorted(p.name for p in out.iterdir()))
$ python ontology_demo.py
689 triples, 9 classes, 9 individuals
groups: {'GOVERNANCE': 2, 'NETWORK': 3, 'SECURITY': 4}
0.0.0.0/0 INGRESS_ALLOWED sg-web
0.0.0.0/0 INTERNET_REACHABLE sg-web
0.0.0.0/0 ONLY_SSH sg-web
RDS instance has no deletion protection FINDING_AFFECTS orders-db
orders-db ENCRYPTED_BY_KMS orders-key
web-1 OWNED_BY https://cloudg.io/resource/tag_payments
web-1 PROTECTED_BY_SG sg-web
web-1 TAGGED_WITH https://cloudg.io/resource/tag_owner_payments
internet reachable: ['sg-web']
security group: 4 triples
compliance gaps: [('HIGH', 'orders-db')]
reloaded 689 triples from ['ontology.jsonld', 'ontology.nt', 'ontology.rdf', 'ontology.ttl']
Chunks for retrieval-augmented generation¶
RAGExporter cuts the estate into one chunk per asset, per community and per relation group, each with metadata you can filter on before a similarity search.
rag_demo.py
"""Export RAG chunks for a small estate and reshape them for a vector store."""
import json
from pathlib import Path
from cloudg.graph.builder import GraphBuilder
from cloudg.graph.rag_export import RAGExporter
from cloudg.schema.models import (
AssetType,
CloudAsset,
CloudProvider,
EdgeType,
Finding,
NetworkEdge,
Severity,
)
ACCOUNT = "123456789012"
REGION = "eu-west-1"
def asset(asset_id: str, asset_type: AssetType, arn: str, **extra) -> CloudAsset:
return CloudAsset(
id=asset_id,
name=asset_id,
asset_type=asset_type,
provider=CloudProvider.AWS,
region=REGION,
account_id=ACCOUNT,
arn=arn,
**extra,
)
assets = [
asset("sg-web", AssetType.SECURITY_GROUP, f"arn:aws:ec2:{REGION}:{ACCOUNT}:security-group/sg-web"),
asset("web-1", AssetType.EC2, f"arn:aws:ec2:{REGION}:{ACCOUNT}:instance/i-0web1",
is_internet_exposed=True),
asset("web-2", AssetType.EC2, f"arn:aws:ec2:{REGION}:{ACCOUNT}:instance/i-0web2"),
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"),
]
edges = [
NetworkEdge(source_id="0.0.0.0/0", target_id="sg-web", edge_type=EdgeType.SECURITY_GROUP_RULE,
port_range="443", protocol="TCP", cidr="0.0.0.0/0", direction="ingress"),
NetworkEdge(source_id="web-1", target_id="sg-web", edge_type=EdgeType.ATTACHED_TO),
NetworkEdge(source_id="web-2", target_id="sg-web", edge_type=EdgeType.ATTACHED_TO),
NetworkEdge(source_id="web-1", target_id="app-role", edge_type=EdgeType.ASSUMES_ROLE),
NetworkEdge(source_id="app-role", target_id="orders-db", edge_type=EdgeType.GRANTS_ACCESS),
]
findings = [
Finding(
resource_id="web-1",
severity=Severity.HIGH,
title="Instance metadata service v1 enabled",
description="IMDSv2 is not enforced.",
source_tool="prowler",
compliance_frameworks=["CIS"],
)
]
graph = GraphBuilder().build(assets, edges)
paths = RAGExporter().export_all(assets, edges, graph, findings, output_dir="rag-out")
print({kind: path.name for kind, path in paths.items()})
index = json.loads(paths["index"].read_text())
print({k: v for k, v in index.items() if k.endswith("chunks")})
chunks = [json.loads(line) for line in paths["chunks"].read_text().splitlines()]
# The content uses a few symbols; plain words embed and display more reliably
MARKERS = {
"\N{WARNING SIGN}": "WARNING:",
"\N{RIGHTWARDS ARROW}": "->",
"\N{LEFTWARDS ARROW}": "<-",
"\N{BULLET}": "-",
}
def clean(text: str) -> str:
for symbol, word in MARKERS.items():
text = text.replace(symbol, word)
return text
def scalar(value):
"""Most vector stores accept only str, int, float and bool metadata."""
if isinstance(value, (str, int, float, bool)):
return value
if isinstance(value, list):
return ",".join(map(str, value))
return json.dumps(value, sort_keys=True)
records = [
{
"id": c["chunk_id"],
"text": clean(c["content"]),
"metadata": {"chunk_type": c["chunk_type"],
**{k: scalar(v) for k, v in c["metadata"].items()}},
}
for c in chunks
]
out = Path("rag-out") / "vector-records.jsonl"
out.write_text("".join(json.dumps(r) + "\n" for r in records))
print(len(records), "records ->", out)
web_1 = next(r for r in records if r["id"] == "entity::web-1")
print(web_1["text"])
print(web_1["metadata"])
# Pre-filter on metadata before any similarity search
urgent = [r["id"] for r in records if r["metadata"].get("severity_max") in ("CRITICAL", "HIGH")]
print("high or critical:", urgent)
$ python rag_demo.py
{'chunks': 'rag_chunks.jsonl', 'index': 'rag_metadata_index.json'}
{'total_chunks': 11, 'entity_chunks': 5, 'community_chunks': 2, 'relation_group_chunks': 4}
11 records -> rag-out/vector-records.jsonl
Resource: web-1
Type: EC2
Provider: AWS
Region: eu-west-1
ARN: arn:aws:ec2:eu-west-1:123456789012:instance/i-0web1
Account: 123456789012
WARNING: INTERNET EXPOSED
Relations (2):
-> PROTECTED_BY_SG: sg-web
-> RUNS_ON: app-role
Findings (1):
[HIGH] Instance metadata service v1 enabled
{'chunk_type': 'entity', 'asset_type': 'EC2', 'provider': 'AWS', 'region': 'eu-west-1', 'account_id': '123456789012', 'is_internet_exposed': True, 'relation_types': 'PROTECTED_BY_SG,RUNS_ON', 'severity_max': 'HIGH', 'compliance_frameworks': 'CIS', 'neighbour_count': 2, 'finding_count': 1, 'arn': 'arn:aws:ec2:eu-west-1:123456789012:instance/i-0web1'}
high or critical: ['entity::web-1', 'community::1']
Related¶
-
The guide: what each export contains and how to load it elsewhere.
-
Every method, from
build()tosave_graphml(). -
The relations, the query helpers and the serialisations.
-
Chunk types, metadata fields and size limits.