Map the cloud. Scan it. Merge everything.

Installation

# core: graph engine, normaliser, renderers, ingest
pip install cloudg

# cloud SDK extras, pulled in only for collection
pip install cloudg[aws]        # boto3, aioboto3
pip install cloudg[azure]      # Azure management SDKs
pip install cloudg[gcp]        # google-cloud-asset, google-auth
pip install cloudg[all]

The core package alone ships the graph engine, ontology builder, findings normaliser, report renderers and cloudg ingest. That is enough to aggregate existing scan results on a machine with no cloud access at all.

Requires Python 3.11 or newer.

The pipeline

collect -> graph -> scan -> normalise -> report
                \-> ontology / RAG / terraform

Collection runs all configured providers concurrently with asyncio, iterating accounts and regions per provider. Regions auto-discover with --regions all. Assets and network edges go into a directed NetworkX graph: BFS from the internet node finds exposed resources, and blast-radius scoring estimates what an attacker could reach from each node.

The enabled scanners run in parallel over the same inventory, each in its own thread with its own timeout. Their findings, the graph's reachability findings and the built-in IAM linter's results all flow into the normaliser, which deduplicates within and across scanners, rescores against CVSS, and maps everything to compliance controls.

Every phase degrades gracefully. A missing binary, an unreachable provider or a failed exporter logs a warning; nothing else stops.

Alongside the pipeline sits an independent function: inventory mapping. It runs no scanners at all. It deep-collects everything deployed or default across every service, links it into one map, and lets scanner findings merge into that map later.

cloudg run

The full pipeline: collect, graph, scan, normalise, report.

cloudg run -p aws --regions us-east-1
cloudg run -p all --regions all
cloudg run -p aws --scanners prowler,trivy
cloudg run -p aws --regions us-east-1 --terraform

When no IaC directory is configured but --terraform is on, the IaC scanners target the generated Terraform recreation of the live infrastructure. Checkov then audits your actual cloud, not whatever directory cloudg happens to run from. There is deliberately no fallback to .: a zero-finding scan of an unrelated directory reads like a clean bill of health, and cloudg refuses to produce one.

collect & scan

The phases alone. collect writes assets and edges without running a scanner; scan runs the selected scanners concurrently and writes raw-findings.json, no collection.

cloudg collect -p aws --profile prod --region eu-west-1
cloudg scan -p aws --scanners prowler,checkov --iac-dir ./terraform

cloudg map

Scanner-independent inventory mapping: what exists, and how is it wired together. What is wrong stays the scanners' business. No scanner runs and none needs to be installed; read credentials are enough.

cloudg map -p aws --regions all
cloudg map -p all --regions all
cloudg map -p aws --findings ./reports/raw-findings.json   # overlay scanner output

Coverage comes in three layers:

Outputs: inventory-map.json (assets, interconnections, summary), inventory-map.graphml and inventory-graph.json. With --findings, additionally asset-map.json (per-asset risk, riskiest first) and compliance-map.json (framework → affected assets). The map never depends on the scanners; findings merge in whenever they exist.

cloudg ingest

Aggregation without execution. Feed cloudg the native output files of scans that already ran (in CI, on another host, on a schedule) and it runs the normalisation half of the pipeline: cross-scanner dedupe via the check-equivalence rulesets, compliance mapping and report generation. No scanner executes and no cloud credentials are needed.

cloudg ingest \
  --prowler ./prowler-output/ \
  --scoutsuite ./scoutsuite-report/ \
  --checkov ./results_json.json \
  --trivy ./trivy-image.json --trivy ./trivy-fs.json \
  -o ./reports

Every input flag is repeatable and any combination or subset of tools works: one tool alone, or all four together. The exact formats are pinned down under Input requirements.

cloudg report

Re-render reports from a previous run's findings.json. No cloud, no scanners. --format takes html, json, svg or all.

cloudg report -i ./reports/findings.json --format html

Input requirements

What each ingest path expects, and the command that produces it.

Prowler

prowler aws -M json-asff -o ./prowler-output

cloudg accepts the output directory or any single file from it, as a JSON array or JSONL. The fields read from each ASFF finding:

{
  "Id": "prowler-aws-s3_bucket_default_encryption-123456789012-eu-west-1-...",
  "Title": "S3 bucket default encryption",
  "Severity": {"Label": "HIGH"},
  "Resources": [{"Id": "arn:aws:s3:::my-bucket"}],
  "Compliance": {"Status": "FAILED", "RelatedRequirements": ["CIS 2.1.1"]},
  "Remediation": {"Recommendation": {"Text": "..."}}
}

Findings with a PASSED status are skipped. The check name embedded in Id becomes the dedupe key.

ScoutSuite

scout aws --report-dir ./scoutsuite-report --no-browser

Accepts the report directory (searched recursively for scoutsuite_results*.js) or the file itself. The file is JavaScript, a JSON object assigned to a variable; cloudg strips the assignment and parses the JSON. Each entry under services.<service>.findings with flagged_items > 0 yields one finding per flagged item. level maps to severity: danger is CRITICAL, warning HIGH, caution MEDIUM.

Checkov

checkov -d ./iac --output json > results_json.json

Accepts the JSON file or a directory containing results_json.json. Both output shapes parse: a single {check_type, results} object, or a list of them when several frameworks ran. Findings come from results.failed_checks[], reading check_id, check_name, severity, resource, file_path and guideline.

Trivy

trivy image --format json myrepo/app:latest > trivy-image.json
trivy fs --format json --scanners vuln,misconfig,secret ./iac > trivy-fs.json

Accepts a single JSON file or a directory of them. The top-level ArtifactType decides the parser: container_image goes through the image parser (vulnerabilities and secrets), everything else through the filesystem parser, which also handles misconfigurations. ArtifactName becomes the resource identifier, and the CVSS v3 score is read when present.

Output contract

{
  "metadata":   {"scan_id", "provider", "account_id", "region", "started_at", "completed_at"},
  "summary":    {"total_assets", "total_findings", "severity_breakdown", "compliance_frameworks"},
  "assets":     [CloudAsset, ...],
  "findings":   [Finding, ...],
  "compliance": [ComplianceResult, ...],
  "graph":      {"nodes": [], "links": []}
}

Data models

Everything flowing through cloudg is a Pydantic v2 model from cloudg.schema.models. All models allow extra fields, so scanner-specific attributes survive serialisation.

Finding

The rest

CloudAsset carries arn, name, a 45-value asset_type taxonomy (EC2, S3_BUCKET, IAM_ROLE, VPC, KMS_KEY, ...), provider, region, tags, metadata and is_internet_exposed. NetworkEdge is a directed edge with an edge_type (SECURITY_GROUP_RULE, IAM_TRUST, CONTAINS, INTERNET_EXPOSED, ATTACHED_TO, REFERENCES, ...), ports, protocol and CIDR. ComplianceResult is one control-level verdict with framework, control_id, status and the finding_ids behind it. ScanResult is the aggregate container the normaliser returns and the renderers consume, with a computed summary.

Configuration

config.yaml mirrors every CLI flag and adds persistent settings. Everything is optional. The shipped config.yaml documents each field inline; the structure, with defaults:

providers: [aws]

aws:
  regions: [us-east-1]          # or [all]
  accounts: []                  # multi-account fan-out
  role_name: null
  role_arn: null
  external_id: null
  web_identity_token_file: null

scanners:
  enabled: [prowler, scoutsuite, checkov, trivy, iam]
  timeout_seconds: 3600
  iac_directories: []
  trivy_images: []
  prowler_extra_args: []        # passthrough, per scanner

inventory: { tagging_sweep: true, link_references: true }  # cloudg map
ontology:  { enabled: true, export_formats: [turtle, json-ld] }
rag:       { enabled: true, chunk_strategy: hybrid, max_chunk_tokens: 2000 }
terraform: { enabled: false }
rulesets:  { rules_dir: null }  # defaults to the packaged cloudg/rules/
concurrency_limit: 5

In code, the same structure is CloudGConfig:

from cloudg import CloudGConfig, load_config

config = load_config("config.yaml")
config = CloudGConfig(providers=["aws"])
config.scanners.enabled = ["prowler", "iam"]

Python API

CloudGEngine in cloudg.api is the integration entry point: async-first, sync wrappers included, built for embedding in larger systems (SIEM pipelines, orchestration platforms, command extensions such as hol-guard).

from cloudg import CloudGConfig, CloudGEngine

engine = CloudGEngine(CloudGConfig(providers=["aws"]))
engine.on_finding = lambda f: siem.send(f.model_dump(mode="json"))
engine.on_error = lambda phase, exc: alerting.notify(phase, exc)

result = engine.run_pipeline_sync()
print(result.to_summary())

Event hooks: on_phase_start(phase), on_collection_complete(result), on_finding(finding), on_scan_complete(findings), on_analysis_complete(result), on_error(phase, exc). Exceptions raised inside a hook are swallowed, so a broken callback cannot take the pipeline down. PipelineResult carries the data, report paths, coverage and a to_summary() dict ready to serialise.

Inventory mapping from code

from cloudg.inventory import InventoryMapper, RelationshipLinker

mapper = InventoryMapper(config)
inventory = mapper.map_inventory_sync()                 # no scanners
inventory.export("./reports")                          # map + graphml + graph json

asset_map = mapper.build_asset_map(inventory, findings) # asset -> risk
compliance = mapper.build_compliance_map(inventory, findings)

edges = RelationshipLinker(assets).link()               # works on ANY CloudAsset list

The deeper toolkit

The CLI is a thin layer; the modules the pipeline uses internally are public, importable API. The markdown reference documents each in full.

Integration recipes

Aggregate existing scan results, get reports

engine = CloudGEngine(CloudGConfig())
result = engine.run_from_reports_sync(
    {"prowler": ["./prowler-output/"], "trivy": ["./trivy-image.json"]},
    output_dir="./reports",
)
print(result.total_findings, result.severity_breakdown)
print(result.report_paths["html"])

Parse first, decide later

from cloudg.ingest import ingest_reports

findings = ingest_reports({"prowler": ["./out/"], "checkov": ["results_json.json"]})
scan_result = engine.normalise_findings(findings)

for c in scan_result.compliance:
    if c.status.value == "FAIL":
        print(c.framework, c.control_id, len(c.finding_ids))

The per-scanner parsers are also public: TrivyScanner.parse_report("./trivy.json") and friends, one tool with no engine.

Map the inventory, overlay findings later

inventory = engine.map_inventory_sync(output_dir="./reports")   # no scanners
print(inventory.summary["assets_by_service"])   # what exists
print(inventory.summary["internet_exposed"])    # what faces the internet

# hours or days later, when scanner output exists:
findings = engine.ingest_reports({"prowler": ["./ci-prowler-output/"]})
InventoryMapper(config).export_merged(inventory, findings, "./reports")
# -> asset-map.json, compliance-map.json

Combine live collection with ingested reports

collection = await engine.collect()
ingested = engine.ingest_reports({"prowler": ["./ci-prowler-output/"]})
scan_result = engine.normalise_findings(ingested, assets=collection.assets)
await engine.analyze(collection.assets, collection.edges, scan_result.findings)

Gate a CI job on severity

result = CloudGEngine(CloudGConfig()).run_from_reports_sync(
    {"checkov": ["results_json.json"]}
)
if result.severity_breakdown.get("CRITICAL", 0):
    sys.exit(1)

Plugins

Custom collectors and scanners register through setuptools entry points; no core changes needed. PluginRegistry discovers them at startup and falls back to the built-ins for any name not overridden.

# my_package/scanner.py
from cloudg.schema.models import Finding, Severity

class MyScanner:
    def run(self) -> list[Finding]:
        return [Finding(
            resource_id="arn:aws:s3:::example",
            severity=Severity.HIGH,
            title="Example finding",
            description="What was found and why it matters",
            source_tool="myscanner",
            source_finding_id="MY_CHECK_001",
        )]

[project.entry-points."cloudg.scanners"]
myscanner = "my_package.scanner:MyScanner"

[project.entry-points."cloudg.collectors"]
mycloud = "my_package.collector:MyCollector"

After pip install, the name works in scanners.enabled and --scanners. If your scanner emits a stable check ID in source_finding_id, add it to rules/check_equivalence.yaml so its findings merge with equivalent checks from other tools.

Authentication

Every provider supports several methods, resolved in fixed priority order, so the same config works on a laptop, in CI and on cloud compute.

Docker

docker pull ghcr.io/morpheuslord/cloudg:latest
docker compose run --rm cloudg run -p aws --regions us-east-1

The image bundles all four scanner binaries, so it is the shortest path to the full pipeline.

Changelog

Major releases only. The full log lives in CHANGELOG.md.