| """Load bounded, high-value public-world graph sources. |
| |
| These loaders deliberately preserve source semantics. ATT&CK and NVD edges are |
| security knowledge, not causal claims; GeoNames edges are geographic structure; |
| and Exoplanet Archive rows are astronomical observations. Keeping the source |
| kind explicit prevents a mixed corpus from teaching the model that every edge |
| means cause and effect. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import csv |
| import gzip |
| import json |
| import re |
| import zipfile |
| from collections.abc import Mapping |
| from pathlib import Path |
| from urllib.parse import unquote |
|
|
| from graph.graph import EdgeRecord |
| from graph.real.causal import _make_graph, _node_id |
| from graph.real.corpus import RealGraph, SourceMetadata, relation_phrase |
|
|
|
|
| def _short(value: object, limit: int = 4000) -> str: |
| return " ".join(str(value or "").split())[:limit] |
|
|
|
|
| def _stix_identifier(value: str) -> str: |
| return "stix:" + re.sub(r"[^A-Za-z0-9_.:-]+", "_", value) |
|
|
|
|
| def load_mitre_attack_stix( |
| path: str | Path, |
| source: SourceMetadata, |
| *, |
| max_objects: int | None = None, |
| ) -> RealGraph: |
| """Load named ATT&CK STIX objects and their relationships.""" |
| payload = json.loads(Path(path).read_text(encoding="utf-8")) |
| objects = payload.get("objects", []) |
| if not isinstance(objects, list): |
| raise TypeError("ATT&CK STIX bundle has no object list") |
|
|
| selected: dict[str, Mapping[str, object]] = {} |
| for item in objects: |
| if not isinstance(item, dict): |
| continue |
| object_id = str(item.get("id", "")) |
| object_type = str(item.get("type", "")) |
| if not object_id or object_type in {"relationship", "marking-definition"}: |
| continue |
| if not item.get("name") and object_type not in {"identity"}: |
| continue |
| selected[object_id] = item |
| if max_objects is not None and len(selected) >= max_objects: |
| break |
|
|
| nodes: dict[str, str] = {} |
| labels: dict[str, str] = {} |
| attributes: dict[str, dict[str, str]] = {} |
| for object_id, item in selected.items(): |
| node_id = _stix_identifier(object_id) |
| object_type = str(item.get("type", "stix_object")) |
| nodes[node_id] = object_type |
| labels[node_id] = _short(item.get("name") or object_id, 240) |
| external_ids = [] |
| references = item.get("external_references", ()) |
| if not isinstance(references, list): |
| references = [] |
| for reference in references: |
| if isinstance(reference, dict) and reference.get("external_id"): |
| external_ids.append(str(reference["external_id"])) |
| attributes[node_id] = { |
| "stix_type": object_type, |
| "description": _short(item.get("description")), |
| "external_ids": ", ".join(external_ids[:16]), |
| "evidence_kind": "mitre_attack_stix_object", |
| } |
|
|
| edges: list[EdgeRecord] = [] |
| relation_ids: set[str] = set() |
| for item in objects: |
| if not isinstance(item, dict) or item.get("type") != "relationship": |
| continue |
| left = str(item.get("source_ref", "")) |
| right = str(item.get("target_ref", "")) |
| if left not in selected or right not in selected or left == right: |
| continue |
| relation = ( |
| re.sub( |
| r"[^a-z0-9_]+", |
| "_", |
| str(item.get("relationship_type", "related_to")).lower(), |
| ).strip("_") |
| or "related_to" |
| ) |
| left_id = _stix_identifier(left) |
| right_id = _stix_identifier(right) |
| relation_ids.add(relation) |
| edges.append( |
| EdgeRecord( |
| left_id, |
| relation, |
| right_id, |
| nodes[left_id], |
| nodes[right_id], |
| relation_phrase(relation), |
| { |
| "evidence_kind": "mitre_attack_stix_relationship", |
| "relationship_id": str(item.get("id", "")), |
| "description": _short(item.get("description"), 1000), |
| }, |
| ) |
| ) |
| return _make_graph( |
| source, |
| nodes, |
| edges, |
| labels=labels, |
| node_attributes=attributes, |
| relation_ids=relation_ids or {"related_to"}, |
| ) |
|
|
|
|
| def _read_json(path: str | Path) -> Mapping[str, object]: |
| file_path = Path(path) |
| opener = gzip.open if file_path.name.endswith(".gz") else open |
| with opener(file_path, "rt", encoding="utf-8") as handle: |
| payload = json.load(handle) |
| if not isinstance(payload, dict): |
| raise TypeError(f"expected JSON object in {file_path}") |
| return payload |
|
|
|
|
| def _cpe_label(value: str) -> str: |
| decoded = unquote(value) |
| parts = decoded.split(":") |
| if len(parts) >= 5: |
| return ":".join(parts[:5]) |
| return decoded[:240] |
|
|
|
|
| def load_nvd_json( |
| path: str | Path, |
| source: SourceMetadata, |
| *, |
| max_records: int | None = None, |
| ) -> RealGraph: |
| """Load NVD 2.0 CVE records, weaknesses, products, scores, and references.""" |
| payload = _read_json(path) |
| raw_records = payload.get("vulnerabilities", payload.get("CVE_Items", ())) |
| if not isinstance(raw_records, list): |
| raise TypeError("NVD feed has no vulnerabilities list") |
| records = raw_records if max_records is None else raw_records[:max_records] |
|
|
| nodes: dict[str, str] = {} |
| labels: dict[str, str] = {} |
| attributes: dict[str, dict[str, str]] = {} |
| edges: list[EdgeRecord] = [] |
| relation_ids = {"has_weakness", "affects_product", "has_reference", "has_severity"} |
|
|
| def add_node(node_id: str, node_type: str, label: str, **attrs: str) -> None: |
| nodes[node_id] = node_type |
| labels[node_id] = _short(label, 240) |
| if attrs: |
| attributes.setdefault(node_id, {}).update(attrs) |
|
|
| def add_edge( |
| left: str, |
| relation: str, |
| right: str, |
| left_type: str, |
| right_type: str, |
| **attrs: str, |
| ) -> None: |
| edges.append( |
| EdgeRecord( |
| left, |
| relation, |
| right, |
| left_type, |
| right_type, |
| relation_phrase(relation), |
| {"evidence_kind": "nvd_cve_record", **attrs}, |
| ) |
| ) |
|
|
| for index, item in enumerate(records): |
| if not isinstance(item, dict): |
| continue |
| cve = item.get("cve", item) |
| if not isinstance(cve, dict): |
| continue |
| cve_id = str( |
| cve.get("id") |
| or cve.get("CVE_data_meta", {}).get("ID", f"CVE-unknown-{index}") |
| ) |
| cve_node = f"nvd:cve:{cve_id}" |
| descriptions = cve.get( |
| "descriptions", cve.get("description", {}).get("description_data", ()) |
| ) |
| description = "" |
| if isinstance(descriptions, list): |
| for entry in descriptions: |
| if isinstance(entry, dict) and ( |
| not description or entry.get("lang") == "en" |
| ): |
| description = _short(entry.get("value")) |
| if entry.get("lang") == "en": |
| break |
| add_node( |
| cve_node, |
| "cve", |
| cve_id, |
| description=description, |
| published=_short(cve.get("published")), |
| last_modified=_short(cve.get("lastModified")), |
| ) |
|
|
| weaknesses = cve.get("weaknesses", ()) |
| for weakness in weaknesses if isinstance(weaknesses, list) else (): |
| if not isinstance(weakness, dict): |
| continue |
| descriptions = weakness.get("description", ()) |
| for entry in descriptions if isinstance(descriptions, list) else (): |
| if not isinstance(entry, dict): |
| continue |
| value = str(entry.get("value", "")).strip() |
| if not value: |
| continue |
| weakness_node = _node_id("nvd_cwe", value) |
| add_node(weakness_node, "weakness", value) |
| add_edge(cve_node, "has_weakness", weakness_node, "cve", "weakness") |
|
|
| configurations = cve.get("configurations", ()) |
| stack: list[object] = ( |
| list(configurations) if isinstance(configurations, list) else [] |
| ) |
| while stack: |
| current = stack.pop() |
| if not isinstance(current, dict): |
| continue |
| stack.extend( |
| current.get("children", ()) |
| if isinstance(current.get("children"), list) |
| else () |
| ) |
| stack.extend( |
| current.get("nodes", ()) |
| if isinstance(current.get("nodes"), list) |
| else () |
| ) |
| matches = current.get("cpeMatch", ()) |
| for match in matches if isinstance(matches, list) else (): |
| if not isinstance(match, dict): |
| continue |
| cpe = str(match.get("criteria") or match.get("cpe23Uri") or "").strip() |
| if not cpe: |
| continue |
| product_node = _node_id("nvd_product", cpe) |
| add_node(product_node, "affected_product", _cpe_label(cpe), cpe=cpe) |
| add_edge( |
| cve_node, "affects_product", product_node, "cve", "affected_product" |
| ) |
|
|
| references = cve.get("references", ()) |
| if isinstance(references, dict): |
| references = references.get("reference_data", ()) |
| for reference in references if isinstance(references, list) else (): |
| if not isinstance(reference, dict): |
| continue |
| url = str(reference.get("url", "")).strip() |
| if not url: |
| continue |
| reference_node = _node_id("nvd_reference", url) |
| add_node( |
| reference_node, |
| "reference", |
| url, |
| source=str(reference.get("source", "")), |
| ) |
| add_edge(cve_node, "has_reference", reference_node, "cve", "reference") |
|
|
| metrics = cve.get("metrics", {}) |
| if not isinstance(metrics, dict): |
| metrics = {} |
| score = severity = "" |
| for metric_name in ( |
| "cvssMetricV40", |
| "cvssMetricV31", |
| "cvssMetricV30", |
| "cvssMetricV2", |
| ): |
| entries = metrics.get(metric_name, ()) |
| if not isinstance(entries, list): |
| continue |
| for entry in entries: |
| if not isinstance(entry, dict): |
| continue |
| cvss = entry.get("cvssData", entry) |
| if isinstance(cvss, dict): |
| score = str(cvss.get("baseScore", "")) |
| severity = str(cvss.get("baseSeverity", "")) |
| if score or severity: |
| break |
| if score or severity: |
| break |
| if score or severity: |
| severity_node = _node_id("nvd_severity", f"{severity}-{score}") |
| add_node( |
| severity_node, |
| "severity", |
| f"{severity} {score}".strip(), |
| score=score, |
| severity=severity, |
| ) |
| add_edge(cve_node, "has_severity", severity_node, "cve", "severity") |
|
|
| return _make_graph( |
| source, |
| nodes, |
| edges, |
| labels=labels, |
| node_attributes=attributes, |
| relation_ids=relation_ids, |
| ) |
|
|
|
|
| def load_nvd_cpe_dictionary( |
| path: str | Path, |
| source: SourceMetadata, |
| *, |
| max_records: int | None = None, |
| ) -> RealGraph: |
| """Load the NVD CPE dictionary as a product/vendor/version ontology. |
| |
| CPE names are software/product identifiers, not vulnerability assertions. |
| The adapter therefore emits product taxonomy edges and keeps the NVD CPE |
| identifier in node attributes for auditability. |
| """ |
| payload = _read_json(path) |
| raw_products = payload.get("products", ()) |
| if not isinstance(raw_products, list): |
| raise TypeError("NVD CPE feed has no products list") |
| records = raw_products if max_records is None else raw_products[:max_records] |
| nodes: dict[str, str] = {} |
| labels: dict[str, str] = {} |
| attributes: dict[str, dict[str, str]] = {} |
| edges: list[EdgeRecord] = [] |
| relation_ids = {"has_vendor", "has_product", "has_version"} |
|
|
| def add_node(node_id: str, node_type: str, label: str, **attrs: str) -> None: |
| nodes[node_id] = node_type |
| labels[node_id] = _short(label, 240) |
| attributes.setdefault(node_id, {}).update(attrs) |
|
|
| for item in records: |
| if not isinstance(item, dict): |
| continue |
| cpe = item.get("cpe", item) |
| if not isinstance(cpe, dict): |
| continue |
| uri = str(cpe.get("cpe23Uri") or cpe.get("cpeName") or "").strip() |
| if not uri: |
| continue |
| parts = uri.split(":") |
| if len(parts) < 6: |
| continue |
| vendor = unquote(parts[3]).replace("_", " ") or "unknown" |
| product = unquote(parts[4]).replace("_", " ") or "unknown" |
| version = unquote(parts[5]).replace("_", " ") or "unspecified" |
| cpe_id = _node_id("nvd_cpe", uri) |
| vendor_id = _node_id("nvd_vendor", vendor) |
| product_id = _node_id("nvd_product_family", f"{vendor} {product}") |
| version_id = _node_id("nvd_version", f"{vendor} {product} {version}") |
| title = "" |
| titles = cpe.get("titles", ()) |
| if isinstance(titles, list): |
| for value in titles: |
| if isinstance(value, dict) and value.get("title"): |
| title = str(value["title"]) |
| if value.get("lang") in {"en", "en-us"}: |
| break |
| add_node(cpe_id, "cpe_name", title or _cpe_label(uri), cpe_uri=uri) |
| add_node(vendor_id, "vendor", vendor) |
| add_node(product_id, "product_family", f"{vendor} {product}") |
| add_node(version_id, "product_version", f"{product} {version}") |
| edge_attrs = {"evidence_kind": "nvd_cpe_dictionary"} |
| for left, relation, right, left_type, right_type in ( |
| (cpe_id, "has_vendor", vendor_id, "cpe_name", "vendor"), |
| (cpe_id, "has_product", product_id, "cpe_name", "product_family"), |
| (cpe_id, "has_version", version_id, "cpe_name", "product_version"), |
| ): |
| edges.append( |
| EdgeRecord( |
| left, |
| relation, |
| right, |
| left_type, |
| right_type, |
| relation_phrase(relation), |
| edge_attrs, |
| ) |
| ) |
| return _make_graph( |
| source, |
| nodes, |
| edges, |
| labels=labels, |
| node_attributes=attributes, |
| relation_ids=relation_ids, |
| ) |
|
|
|
|
| def load_cisa_kev(path: str | Path, source: SourceMetadata) -> RealGraph: |
| """Load CISA's Known Exploited Vulnerabilities catalog.""" |
| payload = _read_json(path) |
| records = payload.get("vulnerabilities", ()) |
| if not isinstance(records, list): |
| raise TypeError("CISA KEV feed has no vulnerabilities list") |
| nodes: dict[str, str] = {} |
| labels: dict[str, str] = {} |
| attributes: dict[str, dict[str, str]] = {} |
| edges: list[EdgeRecord] = [] |
| relation_ids = {"affects_product", "involves_vendor", "has_remediation", "due_by"} |
| for item in records: |
| if not isinstance(item, dict): |
| continue |
| cve = str(item.get("cveID", "")).strip() |
| if not cve: |
| continue |
| cve_node = f"kev:cve:{cve}" |
| nodes[cve_node] = "exploited_vulnerability" |
| labels[cve_node] = cve |
| attributes[cve_node] = { |
| "vulnerability_name": _short(item.get("vulnerabilityName")), |
| "description": _short(item.get("shortDescription")), |
| "date_added": _short(item.get("dateAdded")), |
| } |
| for field, relation, node_type in ( |
| ("vendorProject", "involves_vendor", "vendor"), |
| ("product", "affects_product", "product"), |
| ("requiredAction", "has_remediation", "remediation"), |
| ("dueDate", "due_by", "date"), |
| ): |
| value = str(item.get(field, "")).strip() |
| if not value: |
| continue |
| target = _node_id("kev_" + node_type, value) |
| nodes[target] = node_type |
| labels[target] = _short(value, 240) |
| edges.append( |
| EdgeRecord( |
| cve_node, |
| relation, |
| target, |
| "exploited_vulnerability", |
| node_type, |
| relation_phrase(relation), |
| {"evidence_kind": "cisa_known_exploited_vulnerability"}, |
| ) |
| ) |
| return _make_graph( |
| source, |
| nodes, |
| edges, |
| labels=labels, |
| node_attributes=attributes, |
| relation_ids=relation_ids, |
| ) |
|
|
|
|
| def _country_info(path: str | Path | None) -> dict[str, tuple[str, str]]: |
| if path is None: |
| return {} |
| result: dict[str, tuple[str, str]] = {} |
| with Path(path).open(encoding="utf-8") as handle: |
| for line in handle: |
| if not line.strip() or line.startswith("#"): |
| continue |
| fields = line.rstrip("\n").split("\t") |
| if len(fields) > 8: |
| result[fields[0]] = (fields[4], fields[8]) |
| return result |
|
|
|
|
| def load_geonames_cities( |
| archive: str | Path, |
| source: SourceMetadata, |
| *, |
| country_info: str | Path | None = None, |
| max_rows: int | None = None, |
| ) -> RealGraph: |
| """Load a bounded GeoNames city/country/region/time-zone graph.""" |
| countries = _country_info(country_info) |
| nodes: dict[str, str] = {} |
| labels: dict[str, str] = {} |
| attributes: dict[str, dict[str, str]] = {} |
| edges: list[EdgeRecord] = [] |
| relation_ids = {"in_country", "in_region", "in_timezone", "within_continent"} |
| with zipfile.ZipFile(archive) as handle: |
| member = next(name for name in handle.namelist() if name.endswith(".txt")) |
| with handle.open(member) as binary: |
| reader = csv.reader( |
| (line.decode("utf-8", "replace") for line in binary), delimiter="\t" |
| ) |
| for index, fields in enumerate(reader): |
| if max_rows is not None and index >= max_rows: |
| break |
| if len(fields) < 18: |
| continue |
| geoname_id, name, ascii_name = fields[:3] |
| country_code, admin1, timezone = fields[8], fields[10], fields[17] |
| if not geoname_id or not country_code: |
| continue |
| city = f"geonames:city:{geoname_id}" |
| country = f"geonames:country:{country_code}" |
| region = _node_id("geonames_region", f"{country_code}:{admin1}") |
| tz = _node_id("geonames_timezone", timezone) if timezone else "" |
| nodes[city] = "city" |
| labels[city] = ascii_name or name |
| attributes[city] = { |
| "name": _short(name, 240), |
| "latitude": fields[4], |
| "longitude": fields[5], |
| "population": fields[14], |
| "feature_code": fields[7], |
| } |
| nodes[country] = "country" |
| labels[country] = ( |
| countries.get(country_code, (country_code, ""))[0] or country_code |
| ) |
| edges.append( |
| EdgeRecord( |
| city, |
| "in_country", |
| country, |
| "city", |
| "country", |
| relation_phrase("in_country"), |
| {"evidence_kind": "geonames_city"}, |
| ) |
| ) |
| if admin1: |
| nodes[region] = "admin_region" |
| labels[region] = f"{country_code} {admin1}" |
| edges.append( |
| EdgeRecord( |
| city, |
| "in_region", |
| region, |
| "city", |
| "admin_region", |
| relation_phrase("in_region"), |
| {"evidence_kind": "geonames_city"}, |
| ) |
| ) |
| if tz: |
| nodes[tz] = "timezone" |
| labels[tz] = timezone |
| edges.append( |
| EdgeRecord( |
| city, |
| "in_timezone", |
| tz, |
| "city", |
| "timezone", |
| relation_phrase("in_timezone"), |
| {"evidence_kind": "geonames_city"}, |
| ) |
| ) |
| continent = countries.get(country_code, ("", ""))[1] |
| if continent: |
| continent_node = _node_id("geonames_continent", continent) |
| nodes[continent_node] = "continent" |
| labels[continent_node] = continent |
| edges.append( |
| EdgeRecord( |
| country, |
| "within_continent", |
| continent_node, |
| "country", |
| "continent", |
| relation_phrase("within_continent"), |
| {"evidence_kind": "geonames_country_info"}, |
| ) |
| ) |
| return _make_graph( |
| source, |
| nodes, |
| edges, |
| labels=labels, |
| node_attributes=attributes, |
| relation_ids=relation_ids, |
| ) |
|
|
|
|
| def load_exoplanet_csv( |
| path: str | Path, |
| source: SourceMetadata, |
| *, |
| max_rows: int | None = None, |
| ) -> RealGraph: |
| """Load confirmed-planet rows from NASA's Exoplanet Archive.""" |
| nodes: dict[str, str] = {} |
| labels: dict[str, str] = {} |
| attributes: dict[str, dict[str, str]] = {} |
| edges: list[EdgeRecord] = [] |
| relation_ids = {"orbits", "discovered_by", "discovered_in", "has_measurement"} |
| with Path(path).open(encoding="utf-8", newline="") as handle: |
| reader = csv.DictReader(handle) |
| for index, row in enumerate(reader): |
| if max_rows is not None and index >= max_rows: |
| break |
| planet_name = str(row.get("pl_name", "")).strip() |
| host_name = str(row.get("hostname", "")).strip() |
| if not planet_name or not host_name: |
| continue |
| planet = _node_id("exoplanet", planet_name) |
| host = _node_id("exostar", host_name) |
| nodes[planet] = "exoplanet" |
| labels[planet] = planet_name |
| nodes[host] = "host_star" |
| labels[host] = host_name |
| attributes[planet] = { |
| "discovery_method": _short(row.get("discoverymethod")), |
| "discovery_year": _short(row.get("disc_year")), |
| "orbital_period_days": _short(row.get("pl_orbper")), |
| "radius_earth": _short(row.get("pl_rade")), |
| "mass_earth": _short(row.get("pl_bmasse")), |
| } |
| edges.append( |
| EdgeRecord( |
| planet, |
| "orbits", |
| host, |
| "exoplanet", |
| "host_star", |
| relation_phrase("orbits"), |
| {"evidence_kind": "nasa_exoplanet_archive"}, |
| ) |
| ) |
| method = str(row.get("discoverymethod", "")).strip() |
| if method: |
| method_node = _node_id("exoplanet_method", method) |
| nodes[method_node] = "discovery_method" |
| labels[method_node] = method |
| edges.append( |
| EdgeRecord( |
| planet, |
| "discovered_by", |
| method_node, |
| "exoplanet", |
| "discovery_method", |
| relation_phrase("discovered_by"), |
| {"evidence_kind": "nasa_exoplanet_archive"}, |
| ) |
| ) |
| year = str(row.get("disc_year", "")).strip() |
| if year: |
| year_node = f"exoplanet:year:{year}" |
| nodes[year_node] = "year" |
| labels[year_node] = year |
| edges.append( |
| EdgeRecord( |
| planet, |
| "discovered_in", |
| year_node, |
| "exoplanet", |
| "year", |
| relation_phrase("discovered_in"), |
| {"evidence_kind": "nasa_exoplanet_archive"}, |
| ) |
| ) |
| for field, value in ( |
| ("pl_rade", row.get("pl_rade")), |
| ("pl_bmasse", row.get("pl_bmasse")), |
| ("st_teff", row.get("st_teff")), |
| ): |
| text = str(value or "").strip() |
| if not text: |
| continue |
| measure_node = _node_id("exoplanet_measure", f"{field}:{text}") |
| nodes[measure_node] = "measurement" |
| labels[measure_node] = f"{field} {text}" |
| edges.append( |
| EdgeRecord( |
| planet, |
| "has_measurement", |
| measure_node, |
| "exoplanet", |
| "measurement", |
| relation_phrase("has_measurement"), |
| {"evidence_kind": "nasa_exoplanet_archive", "field": field}, |
| ) |
| ) |
| return _make_graph( |
| source, |
| nodes, |
| edges, |
| labels=labels, |
| node_attributes=attributes, |
| relation_ids=relation_ids, |
| ) |
|
|
