Recipes

Section 15 · From docs/INVENTORY_INTERNALS.md on GitHub

15.1 Add an AWS service collector

1. Pick the package by domain (application, containers, data_ml, governance, network_ext, platform, security, serverless) and add a module, or extend one that has room under the size limits (17).

2. Write a mixin with one _collect_<task> coroutine per task, built only from the helpers:

   from cloudg.inventory.aws_services._base import AWSServiceMixin, gather_limited, rel
   from cloudg.schema.models import AssetType, CloudAsset, EdgeType
   
   
   class WidgetCollectorsMixin(AWSServiceMixin):
       async def _collect_widgets(self) -> list[CloudAsset]:
           assets: list[CloudAsset] = []
           async with self._client("widgets") as client:
               async for w in self._paginate(client, "list_widgets", "Widgets"):
                   assets.append(
                       self._asset(
                           arn=w["WidgetArn"],
                           name=w["Name"],
                           asset_type=AssetType.OTHER,
                           tags=w.get("Tags"),
                           metadata={"status": w.get("Status")},
                           relations=[
                               rel(
                                   w.get("RoleArn"),
                                   EdgeType.ASSUMES_ROLE,
                                   "RUNS_ON",
                                   description="widget role",
                               ),
                               rel(w.get("KmsKeyId"), EdgeType.REFERENCES, "ENCRYPTED_BY_KMS"),
                               rel(
                                   w.get("SourceQueueArn"),
                                   EdgeType.INVOKES,
                                   "TRIGGERED_BY",
                                   reverse=True,
                               ),
                           ],
                       )
                   )
           return assets
   

   Use _pages() when the operation has no paginator, gather_limited() for per-item describe calls, _disabled_asset() for security services that can be off. 3. Add the mixin to the package's composite mixin and register the task in the package registry: "widgets": (self._collect_widgets, "<family>", <is_global>). 4. If Cloud Control lists the type, add it to covered_types in aws_cloudcontrol.yaml. 5. If the ARN shape is new, add it to aws_arn_types.yaml. 6. Test with moto when it supports the service, otherwise with a stubbed client (18). Assert on the asset and on the linked edge. 7. Add the task to the family table in DOCUMENTATION.md if it is user-visible.

15.2 Add a service family

Use the new family name in the registry entries; SERVICE_FAMILIES picks it up and --services <family> works immediately. Document it in the family table.

15.3 Add an Azure extractor

from cloudg.inventory.aws_services._base import rel
from cloudg.inventory.azure_graph.helpers import _get, _rid
from cloudg.inventory.azure_graph.registry import _Draft, _set_exposure, extractor
from cloudg.schema.models import EdgeType


@extractor("microsoft.widgets/widgets")
def _widget(d: _Draft, b) -> None:
    d.md["tier"] = _get(d.props, "tier")
    d.add(
        rel(
            _rid(_get(d.props, "subnet")),
            EdgeType.CONTAINS,
            "SUBNET_CONTAINS_INSTANCE",
            reverse=True,
        )
    )
    _set_exposure(d)

Put it in the module for its area (the module must be imported by azure_graph/__init__.py), add the ARM type to azure_arm_types.yaml, and test it with a recorded Resource Graph row passed through build_assets(rows, subscription_id).

Add an Azure child resource kind

When the type is not in Resource Graph (a proxy resource under another one), add a ChildSpec to azure_graph/children.py and give it a task name in families.AZURE_TASK_FAMILIES:

ChildSpec("widgets", ("microsoft.widgets/factories",), "widgets", "2024-01-01")

The rows come back ARM-shaped, so the @extractor("microsoft.widgets/factories/widgets") above applies to them unchanged. If the rows only link other assets and should not become assets themselves, register a handler instead:

from cloudg.inventory.azure_graph.registry import relation_row


@relation_row("microsoft.widgets/factories/links")
def _widget_link(row, b) -> None:
    b.defer(parent_resource_id(row["id"]), rel(_get(row["properties"], "targetId"), EdgeType.REFERENCES, "DEPENDS_ON"))

Test it with tests/azure_fakes.py: collect_linked(tables, arm_routes) runs the collector against a TypedFakeGraph and a FakeArm keyed by request path, and returns the linked graph and the coverage records.

15.4 Add a GCP extractor

from cloudg.inventory.gcp_relations.context import Extracted, GCPContext, _extractor
from cloudg.inventory.gcp_relations.names import full_name
from cloudg.schema.models import EdgeType


@_extractor("widgets.googleapis.com/Widget")
def _widget(ctx: GCPContext, out: Extracted) -> None:
    out.sa(ctx.data.get("serviceAccount"), ctx)
    out.add(full_name(ctx.data.get("network"), "compute"), EdgeType.ATTACHED_TO)
    if ctx.data.get("publicAccess"):
        out.expose("public widget endpoint", protocol="tcp", ports=["443"])

Add the type to gcp_asset_types.yaml (deep), and test with a recorded list_assets record.

Registering a second extractor for a type that already has one is fine: both run, in import order, on the same Extracted, and a failing one does not stop the other. A type that only a service API returns gets a source in gcp_sources/ (9.6) that hands its records to ctx.build(), plus an entry marked (API) in the catalog. Put the type's family in families or type_families so --services can select it.

15.5 Add an asset type

1. Add the value to AssetType in cloudg/schema/models.py, in the right section, with a comment naming the native resources it covers.
2. Map native types to it in the catalogs.
3. If it is a security service, give its assets security_service / enabled.
4. If it should be scanned or protected, review security_coverage().
5. Regenerate INVENTORY_CATALOG.md entries by hand or with the capture procedure in 18.4.

15.6 Declare relations from your own collector or plugin

Any asset that reaches the linker can declare relations; no cloudg code change is needed:

asset.metadata["relations"] = [
    {
        "target": "arn:aws:iam::123456789012:role/app",
        "edge": "ASSUMES_ROLE",
        "relationship": "RUNS_ON",
    },
    {"target": "sg-0123456789abcdef0", "edge": "ATTACHED_TO", "relationship": "PROTECTED_BY_SG"},
]
asset.metadata["aliases"] = ["my-internal-id-42"]
edges = RelationshipLinker(all_assets).link()

15.7 Extend the catalogs without code

export CLOUDG_CATALOG_DIR=/etc/cloudg/catalogs
cat > $CLOUDG_CATALOG_DIR/aws_cloudcontrol.yaml <<'EOF'
asset_types:
  AWS::MyCorp::Widget: OTHER
covered_types:
  internal:
    - AWS::MyCorp::Gadget
EOF
cloudg map -p aws --regions all