CloudOntology

Writes assets, edges and findings into an rdflib graph with 64 typed relations, so you can export it as RDF and query it with SPARQL.

Last changed v0.6.0

The ontology is the semantic view of an inventory. Where GraphBuilder keeps the collector's edge types (SECURITY_GROUP_RULE, ATTACHED_TO), CloudOntology reads each edge together with its endpoint types, ports and CIDR and turns it into relations that say what the edge means: INTERNET_REACHABLE, ONLY_SSH, PROTECTED_BY_SG, VPC_CONTAINS_SUBNET. cloudg run builds it after the scanners finish, so findings are part of the graph, and writes ontology.ttl and ontology.jsonld by default.

What build() writes

flowchart LR
  S["Schema: classes and properties"] --> G["rdflib Graph"]
  A["Assets"] --> I["Individuals with data properties"] --> G
  E["Edges"] --> R["infer_relations()"] --> G
  E --> M["EdgeMetadata blank nodes"] --> G
  A --> AR["Asset-level relations"] --> G
  F["Findings"] --> X["AssetIndex resolves the asset"] --> G

Where the triples come from

Three namespaces are bound on the graph, so they appear as prefixes in every serialisation and in your SPARQL:

Prefix	IRI	Holds
cm:	https://cloudg.io/ontology#	classes such as cm:RelationalDatabase and cm:SecurityFinding
cmr:	https://cloudg.io/resource/	individuals, keyed by asset id (cmr:orders-db), plus cmr:finding_<id>, cmr:compliance_<framework> and tag nodes
cmp:	https://cloudg.io/property/	the 64 relations and the data properties (cmp:hasName, cmp:hasARN)

In order, build() adds:

1. The schema. An OWL class for every asset type (EC2, VIRTUAL_MACHINE and GCE_INSTANCE all map to cm:ComputeInstance; types without a hand-picked name get the CamelCase of the enum name), each a subclass of cm:CloudResource. Every RelationType becomes an owl:ObjectProperty with an rdfs:label and a cm:relationGroup literal. This part alone is 636 triples, so an empty inventory still produces a sizeable file.
2. One individual per asset: rdf:type, cmp:hasName, cmp:hasProvider, cmp:hasRegion, cmp:isInternetExposed (an xsd:boolean), and cmp:hasARN and cmp:hasAccountId when set. Each tag adds a cmp:TAGGED_WITH link to a cm:TagValue node named tag_<key>_<value>. With CloudOntology(include_raw_metadata=True) (the ontology.include_raw_metadata key, read by cloudg run and CloudGEngine.analyze()), each asset's metadata dict is added too, as one JSON string literal under cmp:hasRawMetadata.
3. For each edge, the relations from infer_relations(), as source relation target. Endpoints that are not assets (0.0.0.0/0) become cm:CloudResource individuals named after the id. An edge with a CIDR, port or protocol also gets a blank node of type cm:EdgeMetadata with cm:fromNode, cm:toNode, cmp:hasCIDR, cmp:hasPort and cmp:hasProtocol.
4. Asset-level relations from metadata: OWNED_BY, COST_ALLOCATED_TO and MONITORED_BY from tags such as owner, team, costcenter and monitoring; VPC_CONTAINS_SUBNET and CONTAINS from a VPC to the assets that share its vpc_id; ENCRYPTED_BY_KMS from an encrypted asset to its key; ROTATES_SECRET; and PROTECTED_BY_SG from an EC2 instance's security_groups metadata.
5. Each finding as a cm:SecurityFinding with cmp:hasName (the title), cmp:hasSeverity and cmp:hasRiskScore, linked by cmp:FINDING_AFFECTS to the asset it resolves to. Each compliance framework on the finding becomes cmr:compliance_<name> with a cmp:COMPLIANCE_GOVERNS link to the same asset.

Findings are matched to assets through AssetIndex: by asset id, then ARN (from resource_id or resource_arn), then a unique name, then a unique ARN tail. Scanner findings that name a resource by ARN therefore attach to the real asset instead of a dangling cmr:arn:aws:... node. Only when nothing matches does the raw resource_id become the object.

Edge types and the relations they produce

Edge	Relations
SECURITY_GROUP_RULE, NACL_RULE	INGRESS_ALLOWED or EGRESS_ALLOWED; INTERNET_REACHABLE when the source is 0.0.0.0/0, ::/0 or an Azure Internet, Any or * tag; CIDR_RESTRICTED for any other CIDR; then ALL_TRAFFIC for an all-ports rule, else ONLY_SSH, ONLY_HTTP, ONLY_HTTPS, ONLY_RDP for each of 22, 80, 443, 3389 inside the ranges, else PORT_RESTRICTED
ATTACHED_TO	PROTECTED_BY_SG when the target is a security group or NSG, PROTECTED_BY_NACL for a NACL, DEPENDS_ON otherwise
CONTAINS	VPC_CONTAINS_SUBNET, SUBNET_CONTAINS_INSTANCE, CLUSTER_CONTAINS_SERVICE or ORG_CONTAINS_ACCOUNT when both endpoint types fit, plain CONTAINS otherwise
IAM_TRUST	ROLE_ASSUMES_ROLE, plus CROSS_ACCOUNT_TRUST when the two accounts differ
IAM_POLICY_ATTACHMENT	USER_HAS_POLICY, ROLE_HAS_POLICY or GROUP_HAS_POLICY by target type
LOAD_BALANCER_TARGET	LB_TARGETS_INSTANCE and LOAD_BALANCED_BY
PEERING, ROUTE, INTERNET_EXPOSED	VPC_PEERED, TRANSIT_ROUTED, INTERNET_REACHABLE
INVOKES, LOGS_TO	the relation of the same name
USES_IMAGE, ASSUMES_ROLE	RUNS_ON
GRANTS_ACCESS, PROTECTS, MONITORS, MANAGES, GOVERNS, REFERENCES	POLICY_ALLOWS_ACTION, PROTECTED_BY_WAF, MONITORED_BY, OWNED_BY, COMPLIANCE_GOVERNS, DEPENDS_ON

A relationship declared on the edge (any RelationType name, as the linker writes them) changes this. On CONTAINS, ATTACHED_TO and the typed inventory edges it replaces the inferred relation. On the network and IAM edges it comes first and the inferred ones follow.

The 64 relation types

Group	Relations
NETWORK (16)	INGRESS_ALLOWED, INGRESS_DENIED, EGRESS_ALLOWED, EGRESS_DENIED, ONLY_HTTP, ONLY_HTTPS, ONLY_SSH, ONLY_RDP, ALL_TRAFFIC, PORT_RESTRICTED, CIDR_RESTRICTED, INTERNET_REACHABLE, VPC_PEERED, TRANSIT_ROUTED, NAT_TRANSLATED, DNS_RESOLVED
CONTAINMENT (8)	CONTAINS, VPC_CONTAINS_SUBNET, SUBNET_CONTAINS_INSTANCE, REGION_CONTAINS_VPC, ACCOUNT_CONTAINS_REGION, ORG_CONTAINS_ACCOUNT, CLUSTER_CONTAINS_SERVICE, LB_TARGETS_INSTANCE
IAM (10)	ROLE_ASSUMES_ROLE, USER_HAS_POLICY, ROLE_HAS_POLICY, GROUP_HAS_POLICY, POLICY_ALLOWS_ACTION, POLICY_DENIES_ACTION, CROSS_ACCOUNT_TRUST, SERVICE_LINKED_ROLE, PERMISSION_BOUNDARY_LIMITS, SCP_RESTRICTS
DATA_FLOW (9)	READS_FROM, WRITES_TO, ENCRYPTS_WITH, DECRYPTS_WITH, LOGS_TO, STREAMS_TO, REPLICATES_TO, BACKUP_TO, CACHE_FOR
SECURITY (9)	PROTECTED_BY_SG, PROTECTED_BY_NACL, PROTECTED_BY_WAF, ENCRYPTED_BY_KMS, ROTATES_SECRET, CERTIFICATE_SECURES, FINDING_AFFECTS, VULNERABILITY_EXPLOITS, COMPLIANCE_GOVERNS
COMPUTE (8)	RUNS_ON, TRIGGERED_BY, INVOKES, SCALES_WITH, LOAD_BALANCED_BY, SCHEDULED_BY, DEPENDS_ON, SERVES_TRAFFIC_TO
GOVERNANCE (4)	TAGGED_WITH, COST_ALLOCATED_TO, OWNED_BY, MONITORED_BY

Several of these, TRIGGERED_BY and READS_FROM among them, are never inferred from edges and appear only when a collector declares them as an edge's relationship. RelationType, RelationGroup, get_relation_group() and get_relations_for_group() are importable from cloudg.graph.ontology.

Serialisation formats

fmt passed to save()	rdflib format	File cloudg run and CloudGEngine.analyze() write
turtle, ttl	Turtle	ontology.ttl
json-ld, jsonld	JSON-LD	ontology.jsonld
xml, rdfxml, rdf	RDF/XML	ontology.rdf
nt, ntriples, n-triples	N-Triples	ontology.nt

Names are matched without regard to case. save() writes Turtle, with a warning, for any other value instead of raising, and creates missing parent directories. cloudg run and analyze() write one file per entry in ontology.export_formats (default ["turtle", "json-ld"]), and --no-ontology or ontology.enabled: false skips the phase. The table is ONTOLOGY_FORMATS in cloudg.graph.ontology; ontology_format(name) gives the rdflib format and extension of a name, and ontology_extension(name) the extension alone.

Examples

Build, query and save

Four assets, two edges and one scanner finding that names the database by ARN. The query lists every relation in the graph with readable names on both ends, skipping the data properties.

ontology_demo.py

"""Build the RDF ontology for a small estate, save it and query it with SPARQL."""

from pathlib import Path

from rdflib import Graph

from cloudg.graph.ontology import CloudOntology, RelationGroup
from cloudg.schema.models import (
    AssetType,
    CloudAsset,
    CloudProvider,
    EdgeType,
    Finding,
    NetworkEdge,
    Severity,
)

ACCOUNT = "123456789012"
REGION = "eu-west-1"
KEY_ARN = f"arn:aws:kms:{REGION}:{ACCOUNT}:key/1111aaaa-22bb-33cc-44dd-555555eeeeee"

def asset(asset_id: str, asset_type: AssetType, arn: str, **extra) -> CloudAsset:
    return CloudAsset(
        id=asset_id,
        name=asset_id,
        asset_type=asset_type,
        provider=CloudProvider.AWS,
        region=REGION,
        account_id=ACCOUNT,
        arn=arn,
        **extra,
    )

assets = [
    asset("sg-web", AssetType.SECURITY_GROUP, f"arn:aws:ec2:{REGION}:{ACCOUNT}:security-group/sg-web"),
    asset("web-1", AssetType.EC2, f"arn:aws:ec2:{REGION}:{ACCOUNT}:instance/i-0web1",
          tags={"owner": "payments"}),
    asset("orders-key", AssetType.KMS_KEY, KEY_ARN),
    asset("orders-db", AssetType.RDS_INSTANCE, f"arn:aws:rds:{REGION}:{ACCOUNT}:db:orders-db",
          metadata={"storage_encrypted": True, "kms_key_id": KEY_ARN}),
]

edges = [
    NetworkEdge(source_id="0.0.0.0/0", target_id="sg-web", edge_type=EdgeType.SECURITY_GROUP_RULE,
                port_range="22", protocol="TCP", cidr="0.0.0.0/0", direction="ingress"),
    NetworkEdge(source_id="web-1", target_id="sg-web", edge_type=EdgeType.ATTACHED_TO),
]

# A scanner finding that names the database by ARN, not by asset id
findings = [
    Finding(
        resource_id=f"arn:aws:rds:{REGION}:{ACCOUNT}:db:orders-db",
        severity=Severity.HIGH,
        title="RDS instance has no deletion protection",
        description="Deletion protection is disabled.",
        source_tool="prowler",
        compliance_frameworks=["CIS"],
    )
]

onto = CloudOntology()
onto.build(assets, edges, findings)
stats = onto.stats()
print(stats["total_triples"], "triples,", stats["classes_used"], "classes,",
      stats["individuals"], "individuals")
print("groups:", dict(sorted(stats["relation_group_counts"].items())))

# The graph binds the cm:, cmr: and cmp: prefixes, so queries can use them
rows = onto.query("""
    SELECT ?subject ?relation ?object WHERE {
        ?s ?p ?o .
        FILTER(STRSTARTS(STR(?p), STR(cmp:)))
        FILTER(!STRSTARTS(STR(?p), STR(cmp:has)))
        FILTER(?p != cmp:isInternetExposed)
        ?s cmp:hasName ?subject .
        OPTIONAL { ?o cmp:hasName ?oname }
        BIND(STRAFTER(STR(?p), STR(cmp:)) AS ?relation)
        BIND(COALESCE(?oname, STR(?o)) AS ?object)
    } ORDER BY ?subject ?relation
""")
for row in rows:
    print(f"  {row['subject']} {row['relation']} {row['object']}")

print("internet reachable:", [r["name"] for r in onto.query_internet_exposed()])
print("security group:", len(onto.query_by_relation_group(RelationGroup.SECURITY)), "triples")
print("compliance gaps:", [(r["severity"], r["resourceName"]) for r in onto.query_compliance_gaps()])

out = Path("ontology-out")
for fmt, name in [("turtle", "ontology.ttl"), ("json-ld", "ontology.jsonld"),
                  ("xml", "ontology.rdf"), ("nt", "ontology.nt")]:
    onto.save(out / name, fmt=fmt)

# Any RDF tool can read the files back; here rdflib itself
reloaded = Graph().parse(out / "ontology.ttl", format="turtle")
print("reloaded", len(reloaded), "triples from", sorted(p.name for p in out.iterdir()))

$ python ontology_demo.py
689 triples, 9 classes, 9 individuals
groups: {'GOVERNANCE': 2, 'NETWORK': 3, 'SECURITY': 4}
  0.0.0.0/0 INGRESS_ALLOWED sg-web
  0.0.0.0/0 INTERNET_REACHABLE sg-web
  0.0.0.0/0 ONLY_SSH sg-web
  RDS instance has no deletion protection FINDING_AFFECTS orders-db
  orders-db ENCRYPTED_BY_KMS orders-key
  web-1 OWNED_BY https://cloudg.io/resource/tag_payments
  web-1 PROTECTED_BY_SG sg-web
  web-1 TAGGED_WITH https://cloudg.io/resource/tag_owner_payments
internet reachable: ['sg-web']
security group: 4 triples
compliance gaps: [('HIGH', 'orders-db')]
reloaded 689 triples from ['ontology.jsonld', 'ontology.nt', 'ontology.rdf', 'ontology.ttl']

Things to notice:

• 636 of the 689 triples are schema. The stats() counts cover classes and individuals actually used.
• ENCRYPTED_BY_KMS points at orders-key because the database's kms_key_id matched the key asset's ARN. With no collected key, the object would be the raw key reference.
• The finding attached to orders-db although its resource_id was an ARN.
• query_internet_exposed() lists the objects of INTERNET_REACHABLE triples, which here is the security group. It reads the rule edges, not the reachability walk; for "which workloads can the internet reach" use ReachabilityAnalyzer.
• The OWNED_BY object is a tag node (tag_payments) with no cmp:hasName, so the query printed its IRI.

Saving N-Triples prints a harmless UserWarning: NTSerializer always uses UTF-8 encoding from rdflib.

Query a saved file without cloudg

The exported files are plain RDF. This needs only rdflib and the ontology.ttl from the previous example (or from cloudg run).

query_saved.py

from rdflib import Graph, Namespace

CMP = Namespace("https://cloudg.io/property/")

g = Graph().parse("ontology-out/ontology.ttl")
rows = g.query(
    """
    SELECT ?resource ?finding ?severity WHERE {
        ?f cmp:FINDING_AFFECTS ?r ;
           cmp:hasName ?finding ;
           cmp:hasSeverity ?severity .
        ?r cmp:hasName ?resource .
    }
    """,
    initNs={"cmp": CMP},
)
for resource, finding, severity in rows:
    print(f"{severity}: {resource}: {finding}")

$ python query_saved.py
HIGH: orders-db: RDS instance has no deletion protection

The same query works against ontology.jsonld with Graph().parse("ontology.jsonld", format="json-ld"), and the Turtle file loads unchanged into a triple store such as GraphDB or Jena Fuseki.

Notes

• query() returns every value as a string, including literals such as "false" and "9.5". An OPTIONAL variable that stays unbound comes back as Python None. query(sparql, bindings={"asset": URIRef(...)}) passes initial variable bindings to rdflib, which is the safe way to put an IRI or a value into a query.
• query_asset_neighbourhood(asset_id) binds the asset's IRI as a value instead of writing it into the query text, so ids holding :, / or spaces (0.0.0.0/0, an ARN) work. The one-hop result has predicate, neighbour and neighbourName per triple, and includes data properties (hasName, hasRegion) next to the relations.
• With hops greater than 1, query_asset_neighbourhood() walks hop by hop in Python, forwards over the network and containment relations in NEIGHBOURHOOD_RELATIONS (INGRESS_ALLOWED, EGRESS_ALLOWED, CONTAINS, VPC_CONTAINS_SUBNET, SUBNET_CONTAINS_INSTANCE, INTERNET_REACHABLE, PROTECTED_BY_SG, LB_TARGETS_INSTANCE). It returns each resource reached once as {"neighbour", "neighbourName", "hops"}, nearest first, with hops the shortest distance. In the example above, query_asset_neighbourhood("web-1", hops=2) gives [{'neighbour': 'https://cloudg.io/resource/sg-web', 'neighbourName': 'sg-web', 'hops': 1}].
• The tag node name is tag_<key>_<value> with spaces and / replaced by _. Other characters pass through into the IRI as they are.
• Each build() call adds to the same graph. Create a new CloudOntology per inventory.
• The JSON-LD output uses rdflib's serialiser without a framing context, so it is a flat list of nodes.

Reference

CloudOntology

CloudOntology(include_raw_metadata: bool = False)

Builds and queries a semantic RDF/OWL graph for cloud infrastructure.

Constructs OWL class hierarchy, typed object/data properties, and individual instances from CloudG's Pydantic models.

Parameters:

Name	Type	Description	Default
include_raw_metadata	bool	Also store each asset's provider metadata (CloudAsset.metadata) as a JSON-encoded string literal on cmp:hasRawMetadata (ontology.include_raw_metadata). Off by default: the metadata can be large and may hold configuration details you do not want in the ontology files.	False

Source code in cloudg/graph/ontology.py

def __init__(self, include_raw_metadata: bool = False) -> None:
    self._include_raw_metadata = include_raw_metadata
    self._graph = Graph()
    self._graph.bind("cm", CM)
    self._graph.bind("cmr", CMR)
    self._graph.bind("cmp", CMP)
    self._graph.bind("owl", OWL)
    self._triple_count = 0

graph property

graph: Graph

Access the underlying rdflib Graph.

triple_count property

triple_count: int

build

build(
    assets: list[CloudAsset],
    edges: list[NetworkEdge],
    findings: list[Finding] | None = None,
) -> Graph

Build the full ontology graph from CloudG data.

Parameters:

Name	Type	Description	Default
assets	list[CloudAsset]	Collected cloud assets.	required
edges	list[NetworkEdge]	Network/relationship edges.	required
findings	list[Finding] | None	Optional security findings.	None

Returns:

Type	Description
Graph	Populated rdflib Graph.

Source code in cloudg/graph/ontology.py

def build(
    self,
    assets: list[CloudAsset],
    edges: list[NetworkEdge],
    findings: list[Finding] | None = None,
) -> Graph:
    """Build the full ontology graph from CloudG data.

    Args:
        assets: Collected cloud assets.
        edges: Network/relationship edges.
        findings: Optional security findings.

    Returns:
        Populated rdflib Graph.
    """
    logger.info("Building ontology from %d assets, %d edges", len(assets), len(edges))

    self._define_schema()
    assets_by_id = {a.id: a for a in assets}

    # Add asset individuals
    for asset in assets:
        self._add_asset(asset)

    # Add edges with inferred semantic relations
    for edge in edges:
        self._add_edge(edge, assets_by_id)

    # Add inferred asset-level relations (KMS key index built once)
    for asset, rel_type, target_id in iter_asset_relations(assets):
        src_uri = CMR[asset.id]
        tgt_uri = (
            CMR[target_id]
            if not target_id.startswith("tag:")
            else CMR[target_id.replace(":", "_")]
        )
        self._graph.add((src_uri, CMP[rel_type.value], tgt_uri))

    # Add findings, attached to the asset they resolve to
    if findings:
        index = AssetIndex(assets_by_id.values())
        for finding in findings:
            self._add_finding(finding, index)

    logger.info("Ontology built: %d triples", len(self._graph))
    return self._graph

query

query(sparql: str, bindings: dict[str, Any] | None = None) -> list[dict[str, Any]]

Execute a SPARQL query and return results as list of dicts.

Values are strings; a variable the row leaves unbound (an OPTIONAL that did not match) is None, not the string "None".

Parameters:

Name	Type	Description	Default
sparql	str	The query. The cm, cmr, cmp, owl, rdf and rdfs prefixes are predefined.	required
bindings	dict[str, Any] | None	Initial variable bindings (name without ? to an rdflib term), passed to rdflib as initBindings. Use them for values such as resource IRIs instead of formatting them into the query text.	None

Source code in cloudg/graph/ontology.py

def query(self, sparql: str, bindings: dict[str, Any] | None = None) -> list[dict[str, Any]]:
    """Execute a SPARQL query and return results as list of dicts.

    Values are strings; a variable the row leaves unbound (an
    ``OPTIONAL`` that did not match) is None, not the string "None".

    Args:
        sparql: The query. The ``cm``, ``cmr``, ``cmp``, ``owl``,
            ``rdf`` and ``rdfs`` prefixes are predefined.
        bindings: Initial variable bindings (name without ``?`` to an
            rdflib term), passed to rdflib as ``initBindings``. Use
            them for values such as resource IRIs instead of formatting
            them into the query text.
    """
    results = []
    for row in self._graph.query(sparql, initBindings=bindings or {}):
        results.append(
            {str(var): (None if val is None else str(val)) for var, val in zip(row.labels, row)}
        )
    return results

query_internet_exposed

query_internet_exposed() -> list[dict[str, str]]

Find all internet-reachable resources.

Source code in cloudg/graph/ontology.py

def query_internet_exposed(self) -> list[dict[str, str]]:
    """Find all internet-reachable resources."""
    sparql = """
    SELECT ?resource ?name ?type WHERE {
        ?source cmp:INTERNET_REACHABLE ?resource .
        ?resource cmp:hasName ?name .
        ?resource a ?type .
        FILTER(?type != owl:NamedIndividual)
    }
    """
    return self.query(sparql)

query_by_relation_group

query_by_relation_group(group: RelationGroup) -> list[dict[str, str]]

Find all triples belonging to a specific relation group.

Source code in cloudg/graph/ontology.py

def query_by_relation_group(self, group: RelationGroup) -> list[dict[str, str]]:
    """Find all triples belonging to a specific relation group."""
    relation_types = get_relations_for_group(group)
    if not relation_types:
        return []

    # Build UNION of all relation types in the group
    patterns = []
    for rt in relation_types:
        patterns.append(f"{{ ?src cmp:{rt.value} ?tgt }}")
    union = " UNION ".join(patterns)

    sparql = f"""
    SELECT ?src ?tgt ?srcName ?tgtName WHERE {{
        {union}
        OPTIONAL {{ ?src cmp:hasName ?srcName }}
        OPTIONAL {{ ?tgt cmp:hasName ?tgtName }}
    }}
    """
    return self.query(sparql)

query_asset_neighbourhood

query_asset_neighbourhood(asset_id: str, hops: int = 1) -> list[dict[str, Any]]

Find all resources within N hops of a given asset.

asset_id is the asset's id as given to build (the local part of its cmr: IRI); ids holding :, / or other characters that are not valid in a prefixed name work, since the IRI is bound as a value rather than written into the query.

With hops 1 (or less) every cmp: triple that has the asset as subject or object is returned as {"predicate", "neighbour", "neighbourName"}. With more hops the walk follows the network relations in NEIGHBOURHOOD_RELATIONS forwards, up to hops steps, and returns each resource reached once as {"neighbour", "neighbourName", "hops"} (hops is the shortest distance), nearest first. neighbourName is None for a node with no cmp:hasName.

Source code in cloudg/graph/ontology.py

def query_asset_neighbourhood(self, asset_id: str, hops: int = 1) -> list[dict[str, Any]]:
    """Find all resources within N hops of a given asset.

    ``asset_id`` is the asset's id as given to :meth:`build` (the local
    part of its ``cmr:`` IRI); ids holding ``:``, ``/`` or other
    characters that are not valid in a prefixed name work, since the
    IRI is bound as a value rather than written into the query.

    With ``hops`` 1 (or less) every ``cmp:`` triple that has the asset
    as subject or object is returned as ``{"predicate", "neighbour",
    "neighbourName"}``. With more hops the walk follows the network
    relations in :data:`NEIGHBOURHOOD_RELATIONS` forwards, up to
    ``hops`` steps, and returns each resource reached once as
    ``{"neighbour", "neighbourName", "hops"}`` (``hops`` is the
    shortest distance), nearest first. ``neighbourName`` is None for a
    node with no ``cmp:hasName``.
    """
    start = CMR[asset_id]
    if hops <= 1:
        sparql = """
        SELECT ?predicate ?neighbour ?neighbourName WHERE {
            {
                ?asset ?predicate ?neighbour .
                FILTER(STRSTARTS(STR(?predicate), STR(cmp:)))
            } UNION {
                ?neighbour ?predicate ?asset .
                FILTER(STRSTARTS(STR(?predicate), STR(cmp:)))
            }
            OPTIONAL { ?neighbour cmp:hasName ?neighbourName }
        }
        """
        return self.query(sparql, bindings={"asset": start})

    predicates = [CMP[rt.value] for rt in NEIGHBOURHOOD_RELATIONS]
    name_p = CMP["hasName"]
    distance: dict[Any, int] = {start: 0}
    queue: deque[Any] = deque([start])
    rows: list[dict[str, Any]] = []
    while queue:
        node = queue.popleft()
        depth = distance[node]
        if depth >= hops:
            continue
        for pred in predicates:
            for nxt in self._graph.objects(node, pred):
                if nxt in distance:
                    continue
                distance[nxt] = depth + 1
                queue.append(nxt)
                name = self._graph.value(nxt, name_p)
                rows.append(
                    {
                        "neighbour": str(nxt),
                        "neighbourName": None if name is None else str(name),
                        "hops": depth + 1,
                    }
                )
    return rows

query_compliance_gaps

query_compliance_gaps() -> list[dict[str, str]]

Find resources affected by findings with compliance frameworks.

Source code in cloudg/graph/ontology.py

def query_compliance_gaps(self) -> list[dict[str, str]]:
    """Find resources affected by findings with compliance frameworks."""
    sparql = """
    SELECT ?finding ?severity ?resource ?resourceName ?framework WHERE {
        ?finding a cm:SecurityFinding .
        ?finding cmp:hasSeverity ?severity .
        ?finding cmp:FINDING_AFFECTS ?resource .
        ?framework cmp:COMPLIANCE_GOVERNS ?resource .
        ?framework rdfs:label ?frameworkName .
        OPTIONAL { ?resource cmp:hasName ?resourceName }
    }
    """
    return self.query(sparql)

save

save(path: str | Path, fmt: str = 'turtle') -> Path

Save ontology to file.

Parameters:

Name	Type	Description	Default
path	str | Path	Output file path, written as given; use ontology_extension to pick a matching extension.	required
fmt	str	Serialization format, a key of ONTOLOGY_FORMATS (turtle, json-ld, xml, nt and their aliases); an unknown name writes Turtle.	'turtle'

Returns:

Type	Description
Path	Path to saved file.

Source code in cloudg/graph/ontology.py

def save(self, path: str | Path, fmt: str = "turtle") -> Path:
    """Save ontology to file.

    Args:
        path: Output file path, written as given; use
            :func:`ontology_extension` to pick a matching extension.
        fmt: Serialization format, a key of :data:`ONTOLOGY_FORMATS`
            (turtle, json-ld, xml, nt and their aliases); an unknown
            name writes Turtle.

    Returns:
        Path to saved file.
    """
    output = Path(path)
    output.parent.mkdir(parents=True, exist_ok=True)

    rdflib_fmt = ontology_format(fmt)[0]

    self._graph.serialize(destination=str(output), format=rdflib_fmt)
    logger.info("Saved ontology (%d triples) to %s as %s", len(self._graph), output, rdflib_fmt)
    return output

to_turtle

to_turtle() -> str

Serialize graph to Turtle format.

Source code in cloudg/graph/ontology.py

def to_turtle(self) -> str:
    """Serialize graph to Turtle format."""
    return self._graph.serialize(format="turtle")

to_jsonld

to_jsonld() -> str

Serialize graph to JSON-LD format.

Source code in cloudg/graph/ontology.py

def to_jsonld(self) -> str:
    """Serialize graph to JSON-LD format."""
    return self._graph.serialize(format="json-ld")

to_rdfxml

to_rdfxml() -> str

Serialize graph to RDF/XML format.

Source code in cloudg/graph/ontology.py

def to_rdfxml(self) -> str:
    """Serialize graph to RDF/XML format."""
    return self._graph.serialize(format="xml")

to_ntriples

to_ntriples() -> str

Serialize graph to N-Triples format.

Source code in cloudg/graph/ontology.py

def to_ntriples(self) -> str:
    """Serialize graph to N-Triples format."""
    return self._graph.serialize(format="nt")

stats

stats() -> dict[str, Any]

Return ontology statistics.

Source code in cloudg/graph/ontology.py

def stats(self) -> dict[str, Any]:
    """Return ontology statistics."""
    classes = set()
    individuals = set()
    relation_counts: dict[str, int] = {}

    for s, p, o in self._graph:
        if p == RDF.type and isinstance(o, URIRef) and str(o).startswith(str(CM)):
            cls_name = str(o).replace(str(CM), "")
            classes.add(cls_name)
            individuals.add(str(s))

        if isinstance(p, URIRef) and str(p).startswith(str(CMP)):
            prop_name = str(p).replace(str(CMP), "")
            relation_counts[prop_name] = relation_counts.get(prop_name, 0) + 1

    # Group relation counts
    group_counts: dict[str, int] = {}
    for rt_name, count in relation_counts.items():
        try:
            rt = RelationType(rt_name)
            grp = get_relation_group(rt).value
            group_counts[grp] = group_counts.get(grp, 0) + count
        except ValueError:
            pass  # Data properties, not relations

    return {
        "total_triples": len(self._graph),
        "classes_used": len(classes),
        "individuals": len(individuals),
        "relation_type_counts": relation_counts,
        "relation_group_counts": group_counts,
    }

Related

• Graph, ontology and RAG

  When to use which view of the inventory.

• RAGExporter

  Relation-group chunks built from the same inference rules.

• GraphBuilder

  The NetworkX view of the same edges.

• cloudg run

  The --ontology/--no-ontology switch and the files it writes.

• Output files

  Every file the pipeline writes, ontology included.