InventoryMapper

Maps what is deployed and how it connects, with no scanner involved, and overlays scanner findings on that map whenever they turn up.

What a mapping run does

InventoryMapper(config, tagging_sweep=None) stores the config and reads the inventory section from it. tagging_sweep overrides config.inventory.tagging_sweep for this mapper only. Nothing touches the cloud until you call map_inventory().

flowchart TD
  A["deep copy of the config"] --> B{"aws.organization.enabled?"}
  B -->|yes| C["discover the org, pick accounts and member role"]
  B -->|no| D
  C --> D["Azure management groups, GCP org and folders"]
  D --> E["deep collectors per account and region"]
  E --> F["deduplicate assets seen twice"]
  F --> G["RelationshipLinker: typed edges"]
  G --> H["account hierarchy nodes"]
  H --> I["InventoryResult"]

InventoryMapper.map_inventory(), step by step

In more detail:

1. The config is deep-copied. Organization discovery rewrites aws.accounts, aws.role_name and, with aws.regions: [ALL] and Control Tower, aws.regions. Those changes stay inside the run; the object you passed in is never modified.
2. With aws.organization.enabled, the mapper reads the OU tree, the member accounts, SCPs and, if present, the Control Tower landing zone. The selected accounts become the collection targets, reached by assuming aws.organization.role_name (falling back to aws.role_name, then AWSControlTowerExecution). The topology is kept on mapper.organization after the run. When discovery fails, the error is logged, an organizations coverage record is marked FAILED, and only the caller's account is mapped.
3. With Azure in providers and azure.map_management_groups on, the management group and Azure Policy hierarchy is read. With GCP, gcp.organization_id set and gcp.map_hierarchy on, the organization, folders, org policies and VPC Service Controls perimeters are read.
4. The deep collectors run per account and region (AWS), per subscription (Azure) and per project or organization (GCP), with the inventory options applied: service families, exclusions, Kubernetes, the tagging and Cloud Control sweeps, IAM resource edges and the rest.
5. Assets that share an ARN (the same resource seen from two regions or from a collector and a sweep) collapse into one. The detailed copy wins over a sweep or placeholder copy, and their declared relations are merged.
6. The RelationshipLinker resolves declared relations and applies its provider rules, with inventory.link_references deciding whether the generic reference scan runs. Placeholder CLOUD_ACCOUNT nodes are added for accounts that were referenced but not mapped.
7. With inventory.account_hierarchy on (the default), account nodes are added and linked to the top-level resources they contain.

Cloud API throttling does not fail a map. Calls are rate limited and retried according to config.ratelimit; a service that is still throttled after its retries is recorded as FAILED or PARTIAL in coverage, and the run's throttling telemetry lands in result.throttling, with each message also logged as a warning. If the code that calls map_inventory() already runs inside a cloudg.resilience.stats_scope(), the mapper reports into that scope.

map_inventory() returns an InventoryResult. Calling the engine's map_inventory() instead gets you the same result plus an on_phase_start("inventory_mapping") hook and optional export.

Merging findings into a map

The last three methods never call a cloud API or a scanner. They take a finished InventoryResult and a list of findings from anywhere: CloudGEngine.scan(), ingest_reports(), a saved findings file.

build_asset_map(result, findings) returns one entry per asset in the map: id, arn, name, type, provider, region, account_id, internet_exposed, finding_count, severity_breakdown and finding_ids, sorted with the most findings first, plus total_assets and assets_with_findings at the top level. Findings are matched to assets through AssetIndex.resolve_finding(), the matcher the ontology and RAG exports use: the asset's internal id, then an ARN taken from resource_id or resource_arn, then a unique name, then a unique ARN tail. A name that several assets share matches none of them, so a finding for a table named orders does not land on a queue also named orders.

build_compliance_map(result, findings) groups by compliance framework. Each framework gets findings, severity_breakdown, affected_assets (the ARN of the asset the finding matched in the same way, or the finding's own resource_arn when no asset matched) and affected_asset_count. It reads finding.compliance_frameworks as it is, so run the findings through FindingsNormaliser first if you want the framework mapping cloudg adds on top of what the scanner reported.

export_merged(result, findings, output_dir) writes both: asset-map.json and compliance-map.json, and returns their paths under the keys asset_map and compliance_map.

Examples

Map an AWS organization

Run this from the management account or a delegated administrator. The member role must exist in every target account; a read-only role deployed with a StackSet is the usual choice.

map_org.py

from cloudg import CloudGConfig
from cloudg.inventory import InventoryMapper

config = CloudGConfig(providers=["aws"])
config.aws.regions = ["ALL"]                           # Control Tower governed regions, if any
config.aws.organization.enabled = True
config.aws.organization.role_name = "cloudg-readonly"
config.aws.organization.include_ous = ["Workloads"]    # nested OUs included
config.aws.organization.exclude_accounts = ["111111111111"]

mapper = InventoryMapper(config)
inventory = mapper.map_inventory_sync()

topology = mapper.organization                         # OrganizationTopology, or None
if topology is not None:
    print("governed regions:", topology.governed_regions)

summary = inventory.summary
print(summary["accounts"], "accounts,", summary["total_assets"], "assets")
print("cross-account edges:", summary["cross_account_edges"])
print("security service gaps:", summary["security_service_gaps"])

for cov in inventory.coverage:
    if cov.failed_services:
        print("failed:", cov.to_summary()["failures"])

inventory.export("./inventory")

AWS Organizations covers the role, the account selection and Control Tower in detail.

Overlay findings that arrive later

Mapping and scanning often run on different schedules or different hosts. This script runs offline: it loads a map exported earlier, ingests Prowler and Trivy output, normalises it, and writes the merged maps next to the inventory.

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

$ python overlay_findings.py
{'asset_map': 'inventory/asset-map.json', 'compliance_map': 'inventory/compliance-map.json'}
CIS 1 ['arn:aws:s3:::my-bucket']
CVE 1 ['myrepo/app:latest']
GDPR-AWS 1 ['arn:aws:s3:::my-bucket']
6 assets, 0 with findings

The map in this run holds six assets of an orders service, and the findings are about a bucket and a container image that are not in it. asset-map.json therefore shows no asset with findings, while compliance-map.json still lists both findings under their own identifiers. A finding about something the map contains is counted against that asset.

The mapper needs a config only to exist; the merge methods ignore it. InventoryMapper(CloudGConfig()) is enough here.

Notes

map_inventory() and the sync wrapper keep no state between calls except mapper.organization, which is overwritten by every run. One mapper can map several times.

map_inventory_sync() uses asyncio.run(). In a notebook or any code that already runs an event loop, await mapper.map_inventory() instead.

InventoryResult.coverage is saved in inventory-map.json under coverage, and InventoryResult.load() reads it back. The CLI also prints the failed collectors after a cloudg map run.

Reference

InventoryMapper

InventoryMapper(config: CloudGConfig, tagging_sweep: bool | None = None)

Maps complete cloud infrastructure without running any scanner.

Parameters:

Name	Type	Description	Default
config	CloudGConfig	cloudg configuration (not mutated).	required
tagging_sweep	bool | None	Override config.inventory.tagging_sweep.	None

Source code in cloudg/inventory/mapper.py

def __init__(self, config: CloudGConfig, tagging_sweep: bool | None = None) -> None:
    self._config = config
    inv_cfg = getattr(config, "inventory", None)
    self._inventory = inv_cfg
    self._tagging_sweep = (
        tagging_sweep
        if tagging_sweep is not None
        else (inv_cfg.tagging_sweep if inv_cfg else True)
    )
    self._link_references = inv_cfg.link_references if inv_cfg else True
    self.organization: Any = None

map_inventory async

map_inventory() -> InventoryResult

Deep-collect every provider and link the assets into a map.

Cloud API throttling never fails the map: calls are rate limited and retried (config.ratelimit), a service still throttled after its retries is recorded FAILED / PARTIAL with reason "throttled", and the run's throttling telemetry is attached as result.throttling.

Source code in cloudg/inventory/mapper.py

async def map_inventory(self) -> InventoryResult:
    """Deep-collect every provider and link the assets into a map.

    Cloud API throttling never fails the map: calls are rate limited and
    retried (``config.ratelimit``), a service still throttled after its
    retries is recorded FAILED / PARTIAL with reason "throttled", and the
    run's throttling telemetry is attached as ``result.throttling``.
    """
    from cloudg.resilience import configure, current_stats, get_governor, stats_scope

    configure(self._config)
    # Reuse a run already bound by the caller (e.g. an MCP live tool) so
    # its stats see this map's throttling too
    with stats_scope(current_stats()) as stats:
        result = await self._map_inventory()
    if stats.eventful:
        result.throttling = get_governor().summary(stats)
        for line in result.throttling.get("messages", []):
            logger.warning("Throttling: %s", line)
    return result

map_inventory_sync

map_inventory_sync() -> InventoryResult

Synchronous wrapper for map_inventory.

Source code in cloudg/inventory/mapper.py

def map_inventory_sync(self) -> InventoryResult:
    """Synchronous wrapper for :meth:`map_inventory`."""
    return asyncio.run(self.map_inventory())

build_asset_map

build_asset_map(result: InventoryResult, findings: list[Finding]) -> dict[str, Any]

Overlay scanner findings onto the inventory: asset to risk view.

Source code in cloudg/inventory/mapper.py

def build_asset_map(self, result: InventoryResult, findings: list[Finding]) -> dict[str, Any]:
    """Overlay scanner findings onto the inventory: asset to risk view."""
    by_asset = self._findings_by_asset(result, findings)
    entries = []
    for asset in result.assets:
        matched = by_asset.get(asset.id, [])
        severity: dict[str, int] = {}
        for f in matched:
            severity[f.severity.value] = severity.get(f.severity.value, 0) + 1
        entries.append(
            {
                "id": asset.id,
                "arn": asset.arn,
                "name": asset.name,
                "type": asset.asset_type.value,
                "provider": asset.provider.value,
                "region": asset.region,
                "account_id": asset.account_id,
                "internet_exposed": asset.is_internet_exposed,
                "finding_count": len(matched),
                "severity_breakdown": severity,
                "finding_ids": [f.id for f in matched],
            }
        )
    entries.sort(key=lambda e: -e["finding_count"])
    return {
        "total_assets": len(entries),
        "assets_with_findings": sum(1 for e in entries if e["finding_count"]),
        "assets": entries,
    }

build_compliance_map

build_compliance_map(
    result: InventoryResult, findings: list[Finding]
) -> dict[str, Any]

Compliance framework to affected assets, from finding mappings.

Findings are matched to assets the way build_asset_map matches them (AssetIndex.resolve_finding); a finding that matches no asset counts its resource_arn as the affected asset.

Source code in cloudg/inventory/mapper.py

def build_compliance_map(
    self, result: InventoryResult, findings: list[Finding]
) -> dict[str, Any]:
    """Compliance framework to affected assets, from finding mappings.

    Findings are matched to assets the way :meth:`build_asset_map`
    matches them (:meth:`AssetIndex.resolve_finding`); a finding that
    matches no asset counts its ``resource_arn`` as the affected asset.
    """
    index = AssetIndex(result.assets)
    id_index = {a.id: a for a in result.assets}

    frameworks: dict[str, dict[str, Any]] = {}
    for f in findings:
        asset = id_index.get(index.resolve_finding(f) or "")
        for fw in f.compliance_frameworks:
            entry = frameworks.setdefault(
                fw,
                {"findings": 0, "severity_breakdown": {}, "affected_assets": set()},
            )
            entry["findings"] += 1
            sev = f.severity.value
            entry["severity_breakdown"][sev] = entry["severity_breakdown"].get(sev, 0) + 1
            if asset:
                entry["affected_assets"].add(asset.arn or asset.id)
            elif f.resource_arn:
                entry["affected_assets"].add(f.resource_arn)

    serialised = {
        fw: {
            "findings": e["findings"],
            "severity_breakdown": e["severity_breakdown"],
            "affected_assets": sorted(e["affected_assets"]),
            "affected_asset_count": len(e["affected_assets"]),
        }
        for fw, e in sorted(frameworks.items())
    }
    return {
        "frameworks": serialised,
        "total_frameworks": len(serialised),
        "total_inventory_assets": len(result.assets),
    }

export_merged

export_merged(
    result: InventoryResult, findings: list[Finding], output_dir: str | Path
) -> dict[str, Path]

Export asset and compliance maps that merge inventory + findings.

Source code in cloudg/inventory/mapper.py

def export_merged(
    self,
    result: InventoryResult,
    findings: list[Finding],
    output_dir: str | Path,
) -> dict[str, Path]:
    """Export asset and compliance maps that merge inventory + findings."""
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    paths: dict[str, Path] = {}

    asset_map_path = out / "asset-map.json"
    with open(asset_map_path, "w") as f:
        json.dump(self.build_asset_map(result, findings), f, indent=2, default=str)
    paths["asset_map"] = asset_map_path

    compliance_path = out / "compliance-map.json"
    with open(compliance_path, "w") as f:
        json.dump(self.build_compliance_map(result, findings), f, indent=2, default=str)
    paths["compliance_map"] = compliance_path

    return paths

Related

• InventoryResult

  What map_inventory() returns.

• RelationshipLinker

  The linking step on its own.

• DependencyGraph

  Blast radius and dependency walks over the map.

• Inventory mapping

  The guide, with the CLI side.

• cloudg map

  The command that wraps this class.