Map and query
Map an account once, then ask questions of the saved map as often as you like: what faces the internet, what depends on a key, what breaks if a queue goes away.
flowchart LR
M["cloudg map / map_inventory()"] --> F["./inventory/inventory-map.json"]
F --> L["InventoryResult.load()"]
L --> Q["summary, exposure"]
L --> D["DependencyGraph: depends_on, blast_radius"]
L --> O["export_merged(): asset-map.json, compliance-map.json"]
S["scanner findings, any time later"] --> OMap an account¶
Needs read-only AWS credentials and pip install "cloudg[aws]". No scanner runs.
map_only.py
from cloudg import CloudGConfig, CloudGEngine
config = CloudGConfig(providers=["aws"])
config.aws.regions = ["ALL"]
config.inventory.exclude_services = ["kubernetes"]
engine = CloudGEngine(config)
inventory = engine.map_inventory_sync(output_dir="./inventory", tagging_sweep=False)
summary = inventory.summary
print(summary["total_assets"], "assets,", summary["total_edges"], "edges")
print("internet exposed:", summary["internet_exposed"])
print("top services:", list(summary["assets_by_service"].items())[:5])
tagging_sweep=False and --no-sweep both skip the Resource Groups Tagging API sweep. InventoryMapper has the other switches and Inventory mapping explains what each layer collects.
Save, load and summarise a map¶
Offline. An InventoryResult built from hand-made assets, exported to disk and loaded back. The same load() reads a map that cloudg map wrote.
inventory_roundtrip.py
from cloudg import InventoryResult
from cloudg.inventory import RelationshipLinker
from cloudg.schema.models import AssetType, CloudAsset, CloudProvider
ACCOUNT = "123456789012"
REGION = "eu-west-1"
KEY = f"arn:aws:kms:{REGION}:{ACCOUNT}:key/1111aaaa-22bb-33cc-44dd-555555eeeeee"
QUEUE = f"arn:aws:sqs:{REGION}:{ACCOUNT}:orders-inbound"
TABLE = f"arn:aws:dynamodb:{REGION}:{ACCOUNT}:table/orders"
FUNCTION = f"arn:aws:lambda:{REGION}:{ACCOUNT}:function:orders-worker"
def aws(arn, name, asset_type, **metadata):
return CloudAsset(arn=arn, name=name, asset_type=asset_type, provider=CloudProvider.AWS,
region=REGION, account_id=ACCOUNT, metadata=metadata)
assets = [
aws(KEY, "orders-key", AssetType.KMS_KEY),
aws(QUEUE, "orders-inbound", AssetType.MESSAGE_QUEUE, kms_key_id=KEY),
aws(TABLE, "orders", AssetType.DYNAMODB_TABLE, kms_key_id=KEY),
aws(FUNCTION, "orders-worker", AssetType.LAMBDA_FUNCTION, kms_key_id=KEY, relations=[
{"target": QUEUE, "edge": "INVOKES", "relationship": "TRIGGERED_BY", "reverse": True},
{"target": TABLE, "edge": "REFERENCES", "relationship": "WRITES_TO"},
{"target": "arn:aws:iam::999999999999:role/partner-reader", "edge": "ASSUMES_ROLE"},
]),
]
linker = RelationshipLinker(assets)
edges = linker.link()
inventory = InventoryResult(
assets=assets + linker.external_assets,
edges=edges,
providers=["aws"],
regions={"aws": [REGION]},
unresolved_references=linker.unresolved,
)
paths = inventory.export("./inventory")
print(sorted(paths))
again = InventoryResult.load("./inventory") # the directory or inventory-map.json
s = again.summary
print(s["total_assets"], s["total_edges"], s["edges_by_type"])
print("external accounts:", s["external_accounts"], "cross-account edges:", s["cross_account_edges"])
analysis = again.analysis(top=5)
print(list(analysis))
print([(d["name"], d["direct_dependents"]) for d in analysis["shared_dependencies"]])
print(analysis["cross_account_edges"])
$ python inventory_roundtrip.py
['dependencies', 'graph', 'graphml', 'map']
5 6 {'INVOKES': 1, 'REFERENCES': 4, 'ASSUMES_ROLE': 1}
external accounts: 1 cross-account edges: 1
['shared_dependencies', 'largest_blast_radius', 'cross_account_edges', 'security_coverage', 'unresolved_references']
[('orders-key', 3)]
[{'source': 'arn:aws:lambda:eu-west-1:123456789012:function:orders-worker', 'source_account': '123456789012', 'target': 'arn:aws:iam::999999999999:root', 'target_account': '999999999999', 'edge_type': 'ASSUMES_ROLE', 'relationship': None, 'external': True}]
Then read the saved map without calling the cloud:
whats_exposed.py
from cloudg import InventoryResult
inventory = InventoryResult.load("./inventory")
summary = inventory.summary
print(f"{summary['internet_exposed']} of {summary['total_assets']} assets face the internet")
for asset in inventory.assets:
if asset.is_internet_exposed:
print(f" {asset.asset_type.value:16} {asset.account_id or '-':14} {asset.display_id}")
print("unresolved references:", summary["unresolved_references"])
for ref in inventory.unresolved_references[:10]:
print(f" {ref['source_name']} -> {ref['target']} ({ref['edge_type']})")
What breaks if this goes away¶
Offline. The script builds its assets by hand. DependencyGraph answers "what does this need" and "what needs this", with the edge each step follows.
Against the map inventory_roundtrip.py saved in ./inventory:
blast_radius.py
from cloudg.inventory import RelationshipLinker
from cloudg.inventory.dependencies import DependencyGraph
from cloudg.schema.models import AssetType, CloudAsset, CloudProvider
ACCOUNT = "123456789012"
REGION = "eu-west-1"
KEY = f"arn:aws:kms:{REGION}:{ACCOUNT}:key/1111aaaa-22bb-33cc-44dd-555555eeeeee"
ROLE = f"arn:aws:iam::{ACCOUNT}:role/orders-worker-role"
QUEUE = f"arn:aws:sqs:{REGION}:{ACCOUNT}:orders-inbound"
TABLE = f"arn:aws:dynamodb:{REGION}:{ACCOUNT}:table/orders"
FUNCTION = f"arn:aws:lambda:{REGION}:{ACCOUNT}:function:orders-worker"
API = f"arn:aws:apigateway:{REGION}::/restapis/a1b2c3d4e5"
def aws(arn, name, asset_type, *relations, region=REGION, exposed=False):
return CloudAsset(arn=arn, name=name, asset_type=asset_type, provider=CloudProvider.AWS,
region=region, account_id=ACCOUNT, is_internet_exposed=exposed,
metadata={"relations": list(relations)})
def rel(target, edge, relationship=None, reverse=False):
return {"target": target, "edge": edge, "relationship": relationship, "reverse": reverse}
assets = [
aws(KEY, "orders-key", AssetType.KMS_KEY),
aws(ROLE, "orders-worker-role", AssetType.IAM_ROLE, region="global"),
aws(QUEUE, "orders-inbound", AssetType.MESSAGE_QUEUE, rel(KEY, "REFERENCES", "ENCRYPTED_BY_KMS")),
aws(TABLE, "orders", AssetType.DYNAMODB_TABLE, rel(KEY, "REFERENCES", "ENCRYPTED_BY_KMS")),
aws(FUNCTION, "orders-worker", AssetType.LAMBDA_FUNCTION,
rel(ROLE, "ASSUMES_ROLE", "RUNS_ON"),
rel(TABLE, "REFERENCES", "WRITES_TO"),
rel(QUEUE, "INVOKES", "TRIGGERED_BY", reverse=True)),
aws(API, "orders-api", AssetType.API_GATEWAY, rel(FUNCTION, "INVOKES"), exposed=True),
]
edges = RelationshipLinker(assets).link()
graph = DependencyGraph(assets, edges)
key = graph.find("orders-key") # ID, ARN, unique name or unique ARN tail
print("If", key.name, "is disabled:")
for link in graph.dependents(key.id):
asset = graph.assets[link.asset_id]
parent = graph.assets[link.parent_id].name
print(f" depth {link.depth}: {asset.name} ({asset.asset_type.value}) needs {parent} via {link.via_edge}")
worker = graph.find(FUNCTION)
print("orders-worker needs:", [graph.assets[link.asset_id].name for link in graph.depends_on(worker.id)])
print("direct (needs, needed by):", graph.direct_counts(worker.id))
for row in graph.blast_radius(top=3):
print(f"{row['name']:20} {row['transitive_dependents']} dependents, "
f"{row['internet_exposed_dependents']} internet-exposed")
$ python blast_radius.py
If orders-key is disabled:
depth 1: orders-inbound (MESSAGE_QUEUE) needs orders-key via REFERENCES
depth 1: orders (DYNAMODB_TABLE) needs orders-key via REFERENCES
depth 2: orders-worker (LAMBDA_FUNCTION) needs orders-inbound via INVOKES
orders-worker needs: ['orders-worker-role', 'orders', 'orders-inbound', 'orders-api', 'orders-key']
direct (needs, needed by): (4, 0)
orders-key 3 dependents, 0 internet-exposed
orders-api 1 dependents, 0 internet-exposed
orders 1 dependents, 0 internet-exposed
Dependencies and blast radius explains how direction works and how to read the tables cloudg deps prints.
Overlay findings on a saved map¶
Scanner findings can arrive days after the map. Load both and write the overlay next to the map. The script reads the map in ./inventory (from inventory_roundtrip.py above, or cloudg map -o ./inventory) and the sample reports: run make_sample_reports.py in the same directory first, and copy its trivy-image.json to trivy.json.
On the CLI the flag belongs to cloudg map, so it maps again. The Python call reuses the saved map.
overlay_findings.py
from cloudg import CloudGConfig, InventoryMapper, InventoryResult
from cloudg.ingest import ingest_reports
from cloudg.normaliser import FindingsNormaliser
# A map written earlier by `cloudg map -o ./inventory` or InventoryResult.export()
inventory = InventoryResult.load("./inventory")
# Scanner output that arrived later, normalised so compliance mapping is filled in
raw = ingest_reports({"prowler": ["./prowler-output/"], "trivy": ["./trivy.json"]})
findings = FindingsNormaliser().normalise(raw).findings
mapper = InventoryMapper(CloudGConfig(providers=["aws"]))
paths = mapper.export_merged(inventory, findings, "./inventory")
print({name: str(path) for name, path in paths.items()})
compliance = mapper.build_compliance_map(inventory, findings)
for framework, entry in list(compliance["frameworks"].items())[:3]:
print(framework, entry["findings"], entry["affected_assets"])
asset_map = mapper.build_asset_map(inventory, findings)
print(asset_map["total_assets"], "assets,", asset_map["assets_with_findings"], "with findings")
The sample findings name resources that are not on this map, so no asset picks one up, but the compliance map still lists every affected resource. Against a map of the account the scanners looked at, each asset in asset-map.json carries its own findings.
Related¶
-
Every field and method of the saved map.
-
depends_on,dependents,tree,blast_radiusand the asset matcher. -
The same questions from the command line.
-
Every file the map writes, field by field.