cloudg reference

cloudg (cloud graphing) maps AWS, Azure and GCP infrastructure into a graph, runs security scanners over that inventory, and merges every finding into one deduplicated, compliance-mapped result set. It can also skip the scanning entirely and aggregate outputs you already have.

This is the complete reference: every CLI command, the exact input and output formats, the data models, the full Python API with integration examples, configuration, plugins and authentication. Release history lives in the changelog.

The inventory mapper (cloudg map, cloudg deps, cloudg.inventory) has three deeper companion documents:

• Inventory reference: the structure of every return value and exported file, field by field, with annotated examples (CloudAsset, NetworkEdge, relation objects, InventoryResult, summary, coverage records, dependency trees, blast radius, security coverage, inventory-map.json, inventory-graph.json, GraphML, inventory-organization.json, asset-map.json, compliance-map.json, cloudg deps --json).
• Inventory catalog: all 156 asset types with the native resource types mapped to each, the metadata keys and relations each one carries, and the full relationship matrix.
• Inventory internals: how a mapping run works, every module, the 133 AWS collector tasks, the linker's resolution rules, the Azure and GCP extractor frameworks, dependency semantics, catalogs, and recipes for adding collectors, extractors and asset types.

Contents:

• Installation
• The pipeline
• CLI reference
• Input requirements
• Output contract
• Data models
• Configuration
• Python API
• Integration recipes
• Plugin system
• Authentication
• Docker

Installation

From PyPI:

pip install cloudg            # core
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]       # every provider SDK

The cloud SDKs are optional extras, pulled in only when you collect from that provider. The core package alone ships the graph engine, the ontology builder, the findings normaliser, the report renderers and cloudg ingest. That is enough to aggregate existing scan results on a machine with no cloud access at all.

The external scanners are separate executables, not Python dependencies: Prowler (pip install prowler), ScoutSuite (pip install scoutsuite), Checkov (pip install checkov) and Trivy (binary install). Install whichever subset you want. cloudg checks PATH at run time and skips anything missing with a warning; the rest of the pipeline is unaffected. install.sh (Linux/macOS) and install.bat (Windows) set up all of them plus the cloud CLIs, and the Docker image bundles the four scanner binaries.

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 when you pass --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 then run in parallel over the same inventory, each in its own thread with its own timeout. Their findings, plus the graph's own 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. The renderers turn the final ScanResult into the report files listed under Output contract.

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

Alongside the pipeline sits an independent function: inventory mapping (cloudg map / CloudGEngine.map_inventory()). It runs no scanners at all. It deep-collects everything deployed or default in the account, sweeps every service for whatever the dedicated collectors miss, and links the lot into one map. Scanner findings can be merged into that map later.

CLI reference

Global options go before the subcommand: cloudg -v -c my-config.yaml run ...

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

Credential flags are listed under Authentication.

A detail worth knowing: 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 rather than 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.

cloudg collect

Asset collection alone, one provider at a time. Writes assets and edges without running any scanner.

cloudg collect -p aws --profile prod --region eu-west-1 -o ./reports
cloudg collect -p azure --subscription-id <id>
cloudg collect -p gcp --project-id <id>

cloudg map

Scanner-independent inventory mapping: everything deployed or default, linked into one map of typed interdependencies. No scanner runs and none needs to be installed.

cloudg map -p aws --regions all
cloudg map -p all --regions all
cloudg map -p aws --org --regions all                       # every account in the Organization / Control Tower
cloudg map -p aws --org --ou Workloads --exclude-account 111122223333
cloudg map -p aws --services containers,serverless,security  # narrow the run
cloudg map -p aws --findings ./reports/raw-findings.json     # merge scanner output

Coverage comes in four layers.

1. Dedicated deep collectors. On AWS, 133 collectors grouped into service families you can select with --services / --exclude-services (family or collector names):

   Family	Collected
   network	VPCs, subnets, security groups, route tables, internet / NAT / egress-only gateways, ENIs, Elastic IPs, NACLs, VPC peering, transit gateways, attachments, route tables and peering, VPC endpoints and endpoint services (PrivateLink, allowed principals, consumers), managed prefix lists, ALB/NLB listeners, rules, certificates and target groups, classic ELB, CloudFront (origins, aliases, certificates, Lambda@Edge, functions, OAC, logging), Global Accelerator, VPC Lattice
   hybrid	site-to-site VPN, customer and virtual private gateways, Direct Connect connections, VIFs and gateways, Cloud WAN / Network Manager
   compute	EC2, Auto Scaling groups, launch templates, AMIs (with sharing), Batch, App Runner, Elastic Beanstalk
   containers	ECR (+ public), ECS clusters, services, task definitions, capacity providers, container instances, Cloud Map, EKS clusters, nodegroups, Fargate profiles, addons, access entries, pod identity
   kubernetes	inside EKS clusters: namespaces, Deployments, StatefulSets, DaemonSets, CronJobs, Services, Ingresses, ServiceAccounts
   serverless	Lambda (triggers, URLs, images, layers, DLQs, aliases), API Gateway REST / HTTP APIs, authorizers, VPC links and custom domains, Step Functions
   integration	SQS, SNS, EventBridge buses, rules, archives, API destinations, Scheduler, Pipes, Kinesis, Firehose, MSK, Amazon MQ
   storage	S3 (notifications, replication, logging, account and bucket Block Public Access), access points and multi-region access points, EBS volumes and snapshots, EFS, FSx, Transfer Family, DataSync
   data	RDS, Aurora, proxies, global clusters, manual snapshot sharing, DynamoDB (KMS, streams, replicas, PITR, policies), ElastiCache, MemoryDB, DAX, OpenSearch (+ Serverless), Redshift (+ Serverless)
   analytics, ml	Glue, Lake Formation permissions, EMR (+ Serverless), Athena; SageMaker, Bedrock agents, knowledge bases, guardrails and invocation logging
   identity	the IAM graph (users, groups, roles, trust, grants, instance profiles, OIDC and SAML providers, access key metadata), IAM Identity Center (instances, permission sets, assignments, users, groups), Roles Anywhere, Cognito user and identity pools, KMS (aliases, rotation, key policy and grant principals), Secrets Manager, ACM
   governance	RAM resource shares, Service Catalog portfolios and provisioned products (Account Factory), CloudFormation StackSets
   security	GuardDuty, Security Hub, Inspector (per-resource coverage), Macie, AWS Config, IAM Access Analyzer, Detective, WAF, Network Firewall, Shield Advanced
   logging, operations, backup	CloudTrail, flow logs, log groups, subscription filters and destinations, cross-account observability (OAM); CloudWatch alarms, SSM parameters (metadata only), managed instances, documents, associations, maintenance windows; AWS Backup plans, selections and vaults
   dns, iac, cicd	Route 53 zones and records, Resolver endpoints, rules and DNS Firewall; CloudFormation stacks and the resources they manage; CodePipeline, CodeBuild, CodeDeploy, CodeConnections

   Account-wide services are collected once per account, in the primary region, so --regions all does not repeat them. Security services that are not enabled appear as placeholder nodes with enabled: false, so detection gaps show on the map. Values that could be secret (parameter values, environment variables, passwords, VPN pre-shared keys, Direct Connect auth keys) are never collected.

   Azure is collected through Azure Resource Graph (azure-mgmt-resourcegraph, in the azure extra) with every resource's full properties, so PaaS services get real relationships: managed identities and role assignments (principal → scope, with role names), private endpoints, AKS node pools, kubelet identity and ACR pulls, App Service plans, VNet integration and container images, Container Apps, subnets, peering, NICs, disks and disk encryption sets, firewalls and route-table next hops, Application Gateway and Front Door backends and WAF policies, Key Vault access policies, storage and SQL network rules, Defender for Cloud plans per subscription. Every Enabled subscription is collected when none are configured (azure.all_subscriptions), and the management group → subscription → resource group hierarchy, Azure Landing Zone archetypes and Azure Policy assignments and exemptions are mapped (azure.map_management_groups). Without the Resource Graph SDK, cloudg falls back to the per-service SDK collectors plus the ARM sweep. Permissions: Reader at the tenant root management group (Resource Graph scopes to what the credential can read).

   GCP is collected through Cloud Asset Inventory list_assets with the full resource JSON, once at organizations/<organization_id> when that is set (gcp.collection_scope), otherwise per project. Relations cover instances (service accounts, subnets, disks, external IPs), firewall semantics (target tags and service accounts, source ranges, priorities), the load-balancing chain down to NEGs and Cloud Armor, managed instance groups and templates, GKE clusters and node pools with Workload Identity, Cloud Run and Functions v2 (service accounts, images in Artifact Registry, secrets, VPC connectors, triggers), Pub/Sub (subscriptions, push endpoints, dead-letter and export targets), Eventarc, logging sinks, CMEK keys, Shared VPC, Cloud SQL / Redis / Filestore networks and DNS zones. IAM bindings resolve to service accounts or to principal nodes (users, groups, workload identity pools); only allUsers / allAuthenticatedUsers mark a resource internet-exposed. With an organization ID, the organization → folder → project tree, organization policies and VPC Service Controls perimeters are mapped too (gcp.map_hierarchy, org_policies, vpc_service_controls). Permissions: roles/cloudasset.viewer on the organization (or each project) and the Cloud Asset API enabled on the quota project.

2. Breadth sweeps, so services without a dedicated collector still appear on the map.

   • AWS Cloud Control API: every CloudFormation resource type with a list handler (800+ types) is listed, tagged or not. Types already mapped by a deep collector are skipped, account-wide types run once per account, types that need a parent identifier or fail are skipped and counted in coverage (cloud_control_skipped_types). The listable-type catalogue is discovered once and cached for a week ($CLOUDG_CACHE_DIR, default ~/.cache/cloudg). Secret-bearing properties (passwords, tokens, SSM parameter values, environment variables, user data) are never copied. Turn it off with --no-cloud-control, or narrow it with inventory.cloud_control_types / cloud_control_exclude (type names or prefixes such as AWS::Glue::).
   • AWS Resource Groups Tagging API: tagged resources only (AWS never returns resources that were never tagged), kept for tag enrichment.
   • Azure Resource Manager's full resources.list() and GCP Cloud Asset Inventory enumerate their scopes.

   Precedence when the same resource is found twice: dedicated collector, then Cloud Control, then the tagging API.

3. The relationship linker. Collectors declare what each asset talks to; the linker resolves those identifiers across services, regions and accounts into typed edges, then applies provider rules and a generic reference pass. See Relationships below.

4. Account hierarchy. Every account gets a CLOUD_ACCOUNT node that contains its top-level resources. With --org, accounts hang under the OU tree, next to SCPs, the Control Tower landing zone and its enabled controls.

Outputs: inventory-map.json (assets, edges, unresolved references, and a summary covering services, types, regions, accounts, relationship counts, cross-account edges, external accounts, security service gaps, internet exposure and unlinked assets), inventory-map.graphml, inventory-graph.json (D3, with account and relationship attributes), inventory-dependencies.json (most shared dependencies, largest blast radius, cross-account edges, security coverage including workloads no vulnerability scanner covers and internet-facing endpoints without a WAF), and with --org inventory-organization.json. With --findings, additionally asset-map.json (per-asset finding counts and severity breakdowns, riskiest first) and compliance-map.json (framework → findings and affected assets). The map does not depend on the scanners; findings from any earlier run merge in whenever they exist.

Relationships

Every edge has a coarse edge_type and, when known, a fine-grained relationship (an ontology relation name) plus properties. Direction always reads "source verb target".

Identifier resolution is scoped: an ambiguous name such as default resolves within the referencing asset's account and region, and is left unresolved rather than guessed when it could belong to several accounts. References into accounts that were not mapped become CLOUD_ACCOUNT nodes with external: true, so a role trusted by a vendor account or a bucket replicating to another account stays visible. References that resolve to nothing (a deleted queue still configured as a DLQ) are listed under unresolved_references.

cloudg deps

Interdependency queries over a saved map; no cloud access needed.

cloudg deps arn:aws:iam::123456789012:role/app-role        # both directions
cloudg deps jobs-queue --direction down --depth 5           # blast radius
cloudg deps --map ./reports                                 # overview
cloudg deps my-function --direction up --json               # machine-readable

Each edge is read as "dependent needs dependency": a function needs its role, image, key and the queue that triggers it; an ALB needs the WAF protecting it; a resource needs the stack managing it. --direction up shows what an asset needs, down what needs it (what breaks or changes with it), both the two trees. Without an asset, deps prints the most shared dependencies (one KMS key behind forty resources), the largest blast radius, and every cross-account edge. Assets are matched by ARN or resource ID, internal ID, or a unique name.

cloudg scan

Scanners alone, no collection. Writes raw-findings.json.

cloudg scan -p aws --scanners prowler,checkov --iac-dir ./terraform

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. See Input requirements for the exact formats.

cloudg report

Re-render reports from a previous run's findings.json. No cloud, no scanners.

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

--format takes html, json, svg or all.

Input requirements

This section pins down exactly what each ingest path expects, and the command that produces it.

Prowler

Produce:

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

cloudg accepts the output directory itself or any single file from it. Files may be a JSON array or JSONL (one ASFF object per line); both parse. The fields cloudg reads from each ASFF finding:

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

Findings with Compliance.Status of PASSED are skipped. The check name embedded in Id (the third dash-separated token) becomes the dedupe key.

ScoutSuite

Produce:

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

cloudg accepts the report directory (searched recursively for scoutsuite_results*.js) or the file itself. The file is JavaScript, a single JSON object assigned to a variable; cloudg strips the assignment and parses the JSON. Findings come from services.<service>.findings.<key>, where each entry with flagged_items > 0 yields one finding per item in items. The level field maps to severity: danger becomes CRITICAL, warning HIGH, caution MEDIUM.

Checkov

Produce:

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

cloudg accepts the JSON file or a directory containing results_json.json (falling back to any *.json in the directory). Both shapes Checkov emits 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, file_line_range and guideline.

Trivy

Produce either or both:

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

cloudg accepts a single JSON file or a directory of them. Each report's top-level ArtifactType decides the parser: container_image goes through the image parser (vulnerabilities and secrets), everything else (filesystem, repository) goes through the filesystem parser, which also handles misconfigurations. ArtifactName becomes the resource identifier. From each vulnerability cloudg reads VulnerabilityID, PkgName, InstalledVersion, FixedVersion, Severity, Description and the CVSS v3 score when present.

Output contract

All commands write into the output directory (./reports by default).

The inventory files are specified field by field in the inventory reference.

findings.json has this shape:

{
  "metadata": {
    "scan_id": "uuid",
    "provider": "AWS",
    "account_id": "...",
    "region": "...",
    "started_at": "...",
    "completed_at": "..."
  },
  "summary": {
    "total_assets": 0,
    "total_findings": 0,
    "total_edges": 0,
    "severity_breakdown": {"CRITICAL": 0, "HIGH": 0},
    "compliance_frameworks": ["CIS", "NIST-800-53"]
  },
  "assets": [],
  "findings": [],
  "compliance": [],
  "graph": {"nodes": [], "links": []}
}

assets, findings and compliance are lists of the models below, serialised with Pydantic's JSON mode. graph is D3 force-layout data. cloudg report -i accepts this same file back, so the format round-trips.

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 unit every scanner and parser produces.

CloudAsset

id is regenerated on every run; correlate assets across runs by arn.

NetworkEdge

Directed edge between two asset IDs (source_id, target_id); direction always reads "source verb target".

edge_type values:

What each one means, its typical endpoints and how it counts for dependency analysis is in the inventory reference; every observed source type → relationship → target type combination is in the relationship matrix.

ComplianceResult

One control-level verdict: framework (for example CIS, NIST-800-53), control_id, control_title, status (PASS, FAIL, NOT_APPLICABLE, MANUAL) and the finding_ids behind it.

ScanResult

The aggregate container the normaliser returns and the renderers consume: assets, findings, edges, compliance, timing fields, and a computed summary dict with totals and the severity breakdown.

Configuration

config.yaml mirrors every CLI flag and adds persistent settings; pass it with -c or keep it next to where you run. Everything is optional, defaults are sensible. The shipped config.yaml documents each field inline. The structure, with defaults:

providers: [aws]                # aws, azure, gcp; several at once is fine

aws:
  regions: [us-east-1]          # or [all]
  accounts: []                  # multi-account fan-out
  role_name: null               # role to assume in each account
  role_arn: null                # single-account AssumeRole or OIDC target
  external_id: null
  web_identity_token_file: null
  profile: null
  max_retries: 10
  retry_mode: adaptive
  organization:                 # AWS Organizations / Control Tower (cloudg map)
    enabled: false
    role_name: null             # default AWSControlTowerExecution
    include_ous: []             # IDs, ARNs or names; nested OUs included
    exclude_accounts: []
    include_management_account: true
    include_suspended: false
    use_governed_regions: true  # regions [all] -> Control Tower governed regions
    control_tower: true
    home_region: null           # auto-detected
    map_structure: true         # org, OU, account, SCP, control nodes

azure:
  subscription_ids: []
  tenant_id: null
  client_id: null
  client_secret: null
  certificate_path: null
  federated_token_file: null
  use_managed_identity: false
  all_subscriptions: true       # every Enabled subscription when subscription_ids is empty
  map_management_groups: true   # management groups, subscriptions, Azure Policy
  regions: [ALL]

gcp:
  project_ids: []
  organization_id: null         # set to collect org-wide and map folders / org policy
  collection_scope: auto        # auto | organization | project
  skip_asset_types: [...]       # high-churn types (Pods, ReplicaSets, Events, ...) by default
  iam_policies: true
  map_hierarchy: true
  org_policies: true
  vpc_service_controls: true
  credentials_file: null
  impersonate_service_account: null
  regions: [ALL]

scanners:
  enabled: [prowler, scoutsuite, checkov, trivy, iam]
  timeout_seconds: 3600
  iac_directories: []           # targets for Checkov and Trivy fs
  trivy_images: []
  checkov_frameworks: []        # empty = auto-detect
  prowler_extra_args: []        # passthrough args, per scanner

inventory:
  tagging_sweep: true           # AWS catch-all sweep (Resource Groups Tagging API)
  link_references: true         # generic cross-service REFERENCES edges
  services: [all]               # families or collector names (see cloudg map)
  exclude_services: []
  kubernetes: true              # workloads inside EKS clusters
  kubernetes_timeout: 10
  iam_resource_edges: true      # principal -> granted resource edges
  max_images_per_repository: 20
  stack_resources: true         # CloudFormation stack -> resource edges
  account_hierarchy: true       # account nodes containing top-level resources
  cloud_control: true           # AWS breadth sweep over every listable resource type
  cloud_control_types: []       # only these CloudFormation types / prefixes
  cloud_control_exclude: []
  cloud_control_concurrency: 6

graph:
  persist_graphml: true
  compute_attack_paths: true

ontology:
  enabled: true
  export_formats: [turtle, json-ld]

rag:
  enabled: true
  chunk_strategy: hybrid        # entity | community | relation_group | hybrid
  max_chunk_tokens: 2000

terraform:
  enabled: false
  output_dir: ./reports/terraform

report:
  formats: [html, json, svg]
  inline_js: true               # air-gapped friendly

rulesets:
  rules_dir: null               # defaults to the packaged cloudg/rules/

concurrency_limit: 5

In code, the same structure is CloudGConfig, a Pydantic model:

from cloudg import CloudGConfig, load_config

config = load_config("config.yaml")          # file, validated
config = CloudGConfig(providers=["aws"])     # or programmatic
config.scanners.enabled = ["prowler", "iam"]
config.aws.regions = ["eu-west-1", "eu-central-1"]

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). The package root re-exports the essentials: CloudGEngine, CloudGConfig, load_config, PipelineResult, CollectionResult, AnalysisResult, InventoryMapper, InventoryResult.

The CLI is a thin layer; nearly everything cloudg does is public, importable API. Beyond the engine itself, the inventory mapping API and the deeper toolkit below are designed to be used directly from code.

Construction and hooks

from cloudg import CloudGConfig, CloudGEngine

engine = CloudGEngine(CloudGConfig(providers=["aws"]))

engine.on_phase_start = lambda phase: print(f"phase: {phase}")
engine.on_finding = lambda finding: siem.send(finding.model_dump(mode="json"))
engine.on_error = lambda phase, exc: alerting.notify(phase, exc)

Hook signatures:

Exceptions raised inside a hook are swallowed, so a broken callback cannot take the pipeline down.

Methods

run_pipeline(output_dir="./reports") -> PipelineResult and its sync twin run_pipeline_sync() run everything: collect, scan, analyse, normalise, report.

collect() -> CollectionResult runs multi-provider collection alone. The result carries assets (list[CloudAsset]), edges (list[NetworkEdge]), per-service coverage records, providers_scanned, regions_scanned and duration_ms.

scan(assets, edges, iac_dir=None, images=None, profile=None, output_dir="./reports") -> list[Finding] runs the scanners from config.scanners.enabled in parallel, plus graph reachability analysis and the IAM linter. Missing binaries are skipped.

ingest_reports(reports: dict[str, list[str | Path]]) -> list[Finding] parses existing scanner outputs instead of running anything. Keys are tool names (prowler, scoutsuite, checkov, trivy), values are lists of report paths, file or directory. Unparseable paths are logged and skipped rather than raising.

normalise_findings(findings, assets=None) -> ScanResult applies the full dedupe, cross-scanner merge, CVSS rescoring and compliance mapping. This is the same code path run_pipeline uses.

analyze(assets, edges, findings, output_dir="./reports") -> AnalysisResult builds the graph and produces the ontology, RAG chunks, Terraform files, attack paths and reachability findings, gated by the corresponding config sections.

run_from_reports(reports, output_dir="./reports") -> PipelineResult and run_from_reports_sync(...) chain ingest, normalise and report generation in one call.

map_inventory(output_dir=None, findings=None, tagging_sweep=None) -> InventoryResult and map_inventory_sync(...) run the scanner-independent inventory mapping: deep collection across all configured providers, catch-all sweeps, and relationship linking. Passing output_dir exports the map files; passing findings as well exports the merged asset and compliance maps.

Inventory mapping API

cloudg.inventory is a standalone package; the engine is optional:

from cloudg import CloudGConfig
from cloudg.inventory import InventoryMapper, RelationshipLinker

config = CloudGConfig(providers=["aws", "azure", "gcp"])
mapper = InventoryMapper(config)

inventory = mapper.map_inventory_sync()      # InventoryResult
print(inventory.summary)                     # services, types, regions, exposure
inventory.export("./reports")                # inventory-map.json / .graphml / graph

InventoryResult carries assets, edges, coverage, providers, regions, organization (the discovered topology, when mapped) and unresolved_references, plus a computed summary (totals, per-service/type/region/account breakdowns, edge types and relationships, cross-account edges, external accounts, security service gaps, internet-exposed and unlinked counts). export(dir) writes inventory-map.json, inventory-map.graphml, inventory-graph.json, inventory-dependencies.json and, for an organization, inventory-organization.json; InventoryResult.load(dir_or_file) reads a map back.

Interdependency questions go through DependencyGraph:

graph = inventory.dependency_graph()
role = graph.find("arn:aws:iam::123456789012:role/app-role")
needs = graph.depends_on(role.id)           # upstream, breadth-first
needed_by = graph.dependents(role.id)       # downstream: blast radius
view = graph.tree(role.id, direction="both", max_depth=3)
graph.shared_dependencies(top=25)
graph.blast_radius(top=25)
inventory.analysis()                        # everything above + coverage

Organizations and Control Tower discovery is usable on its own:

import boto3
from cloudg.inventory import discover_organization

topology = discover_organization(boto3.Session())
topology.governed_regions, topology.shared_accounts      # Control Tower
topology.target_accounts(include_ous=["Workloads"])     # nested OUs included
assets = topology.to_assets()                            # org, OUs, accounts, SCPs, controls

Merging scanner findings (from scan(), ingest_reports(), or a saved raw-findings.json) happens after the fact, so the mapper never touches the scanners:

asset_map = mapper.build_asset_map(inventory, findings)         # asset -> findings/severity
compliance_map = mapper.build_compliance_map(inventory, findings)  # framework -> assets
mapper.export_merged(inventory, findings, "./reports")          # asset-map.json + compliance-map.json

RelationshipLinker works on any list[CloudAsset], whoever collected them:

linker = RelationshipLinker(assets)
linker.seed_existing(existing_edges)   # don't duplicate edges you already have
edges = linker.link()                  # typed edges, see Relationships
linker.external_assets                 # placeholder nodes for unmapped accounts
linker.unresolved                      # declared references that resolved to nothing

Reference data lives in YAML catalogs under cloudg/inventory/catalogs/ rather than in code: aws_cloudcontrol.yaml (types the deep collectors already cover, account-wide types, CloudFormation type → asset type, per-type secret fields), aws_arn_types.yaml (ARN → asset type for sweep hits), azure_arm_types.yaml (ARM resource type → asset type) and gcp_asset_types.yaml (Cloud Asset Inventory type → asset type). Point $CLOUDG_CATALOG_DIR at a directory holding files of the same names to extend or override them without touching cloudg: mappings merge key by key, lists are extended. Unknown asset type names are rejected at load time.

Your own collectors can declare relationships the same way the built-in ones do, by listing them in metadata["relations"]: {"target": <any identifier>, "edge": "INVOKES", "relationship": "TRIGGERED_BY", "reverse": false}.

The deep collectors are public too, when you want single-region, single-account control: AWSDeepInventoryCollector(session, region="us-east-1", account_id=None, tagging_sweep=True, **options) (keyword options is_primary_region, services, exclude_services, kubernetes, kubernetes_timeout, iam_resource_edges, max_images_per_repository, stack_resources, cloud_control, cloud_control_types, cloud_control_exclude, cloud_control_concurrency; an unknown option raises TypeError), AzureDeepInventoryCollector(credential, subscription_id, graph_client_factory=None, use_resource_graph=True) and GCPDeepInventoryCollector(project_id, credentials=None, **options) (keyword options organization_id, scope, project_filter, skip_asset_types, page_size, include_iam, client, coverage, timeout, link_locally) all implement the standard collect() / collect_edges() / run() collector interface. Their assets carry declared relations; pass them through RelationshipLinker for the typed edges.

Every structure above, from summary to the dependency tree, is specified field by field in the inventory reference.

The deeper toolkit

Modules the pipeline uses internally that are equally useful standalone:

PipelineResult

Integration recipes

Working examples against the real API. Each is self-contained.

Aggregate existing scan results, get reports

from cloudg import CloudGConfig, CloudGEngine

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

print(result.total_findings, result.severity_breakdown)
print(result.report_paths["html"])   # ./reports/report.html
for f in result.findings:
    if f.severity.value in ("CRITICAL", "HIGH"):
        print(f.risk_score, f.title, f.resource_arn)

Parse first, decide later

from cloudg import CloudGConfig, CloudGEngine
from cloudg.ingest import parse_report, ingest_reports

findings = parse_report("prowler", "./prowler-output/")
findings += parse_report("checkov", "./results_json.json")
# or in one call:
findings = ingest_reports({"prowler": ["./out/"], "checkov": ["results_json.json"]})

engine = CloudGEngine(CloudGConfig())
scan_result = engine.normalise_findings(findings)

print(scan_result.summary)
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, if you want a single tool with no engine:

from cloudg.scanners.trivy import TrivyScanner

findings = TrivyScanner.parse_report("./trivy-image.json")

Stream findings into a SIEM as they arrive

import asyncio
from cloudg import CloudGConfig, CloudGEngine

engine = CloudGEngine(CloudGConfig(providers=["aws"]))
engine.on_finding = lambda f: queue.put(f.model_dump(mode="json"))
engine.on_error = lambda phase, exc: log.warning("cloudg %s failed: %s", phase, exc)

result = asyncio.run(engine.run_pipeline(output_dir="/var/lib/cloudg/reports"))

Run the phases yourself

import asyncio
from cloudg import CloudGConfig, CloudGEngine

async def main():
    engine = CloudGEngine(CloudGConfig(providers=["aws"]))

    collection = await engine.collect()
    findings = await engine.scan(collection.assets, collection.edges)
    analysis = await engine.analyze(collection.assets, collection.edges, findings)
    scan_result = engine.normalise_findings(findings, assets=collection.assets)

    print(analysis.graph_nodes, "nodes,", len(analysis.attack_paths), "attack paths")
    return scan_result

asyncio.run(main())

Combine live collection with ingested reports

Collection needs credentials; ingest does not. Together they give you the graph outputs and the aggregated findings in one result:

import asyncio
from cloudg import CloudGConfig, CloudGEngine

async def main():
    engine = CloudGEngine(CloudGConfig(providers=["aws"]))

    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)

asyncio.run(main())

Map the inventory, overlay findings later

Inventory mapping needs read credentials and nothing else. Scanners can run elsewhere (CI, a schedule, another host) and merge in whenever their output arrives:

from cloudg import CloudGConfig, CloudGEngine
from cloudg.inventory import InventoryMapper

config = CloudGConfig(providers=["aws"])
config.aws.regions = ["ALL"]

engine = CloudGEngine(config)
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
print(inventory.summary["unlinked_assets"])     # what nothing points at

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

Link relationships into someone else's inventory

The linker is pure post-processing. Feed it assets from any source that produces CloudAsset objects (a plugin collector, a CMDB import, a previous run):

from cloudg.inventory import RelationshipLinker
from cloudg.graph.builder import GraphBuilder

edges = RelationshipLinker(assets).link()
graph = GraphBuilder().build(assets, edges)
print(graph.number_of_nodes(), graph.number_of_edges())

Gate a CI job on severity

import sys
from cloudg import CloudGConfig, CloudGEngine

result = CloudGEngine(CloudGConfig()).run_from_reports_sync(
    {"checkov": ["results_json.json"], "trivy": ["trivy.json"]}
)
critical = result.severity_breakdown.get("CRITICAL", 0)
if critical:
    print(f"{critical} critical findings, failing the build")
    sys.exit(1)

Plugin system

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

A minimal scanner plugin:

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

class MyScanner:
    def __init__(self, **kwargs):
        ...

    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",
            )
        ]

Registered in the plugin's own pyproject.toml:

[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 is available in scanners.enabled and --scanners. Collectors follow the base interface in cloudg/collectors/base.py. A collector plugin's assets join the inventory map like the built-in ones: declare what each asset talks to in metadata["relations"] and extra identifiers in metadata["aliases"], and the relationship linker resolves them (see Inventory internals). If your scanner emits a stable check ID in source_finding_id, you can 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 a fixed priority order, so the same config works on a laptop, in CI and on cloud compute.

AWS, in priority order: direct keys (--aws-key, --aws-secret, --aws-session-token or the standard env vars); OIDC web identity federation (--aws-role-arn with --aws-web-identity-token-file, the GitHub Actions, GitLab CI and EKS pattern with no long-lived keys); a named CLI profile via --profile, SSO included; or nothing, letting the default chain pick up env vars, cached SSO or the instance role. On top of any of these, STS role assumption with --aws-role-arn and, for the third-party auditor pattern, --aws-external-id. Multi-account fan-out uses accounts plus role_name in the config, assuming that role in each account before collecting; the caller's own account is collected with the base credentials.

AWS Organizations and Control Tower: run cloudg map --org with credentials for the management account (Organizations APIs also work from a delegated administrator; the Control Tower APIs need the management account). cloudg lists the org read-only (organizations:Describe*/List*, controltower:List*/Get*) and then assumes the member role in each selected account. AWSControlTowerExecution exists in every account Control Tower enrolled, but it carries administrator permissions; the aws-controltower-ReadOnlyExecutionRole cannot be used directly because it only trusts a Lambda-only role in the audit account. For production, deploy a dedicated read-only role (for example cloudg-readonly with the SecurityAudit and ViewOnlyAccess managed policies, trusting the management account) to all accounts with a service-managed CloudFormation StackSet targeting your OUs, and pass it with --org-role cloudg-readonly. Mapping Kubernetes workloads additionally needs an EKS access entry for that role with AmazonEKSViewPolicy.

Azure, in priority order: workload identity federation (--azure-tenant-id, --azure-client-id, --azure-federated-token-file); a service principal with a client secret (--azure-client-secret) or a certificate (--azure-cert-path); managed identity (--azure-managed-identity, with managed_identity_client_id in config for user-assigned identities); or the DefaultAzureCredential chain, which covers az login.

GCP: a credentials file via --gcp-credentials-file (service account key JSON or a workload identity federation external_account config), or application default credentials. --gcp-impersonate-sa layers service account impersonation on either.

Docker

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

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

Or build locally with docker build -t cloudg:latest .

Found a gap in this reference? Open an issue.