s0u9ata/security-kg

from datasets import load_dataset

ds = load_dataset("s0u9ata/security-kg", "enterprise")
print(ds["train"][0])
# {'subject': 'T1059.001', 'predicate': 'rdf:type', 'object': 'Technique', 'source': 'attack', 'object_type': 'enum', 'meta': ''}

     Group  Campaign
        \      /
          uses
           |
           v
      TECHNIQUE -----> Tactic
        ^  ^  ^
        |  |  |
        |  |  +-- D3FEND (counters)
        |  |  +-- CAR (detects)
        |  |  +-- Sigma (detects)
        |  |  +-- ENGAGE (engages)
        |  |  +-- F3 (fraud techniques)
        |  |  +-- ATLAS (related)
        |  |  +-- MISP Galaxies (cross-refs)
        |  |  +-- LOLBAS (maps-to)
        |  |  +-- LOLDrivers (maps-to)
        |  |  +-- Atomic Red Team (tests)
        |  |  +-- NIST 800-53 (mitigates)
        |  |
        |  +-- Mitigation (mitigates)
        |  +-- DataComponent (detects)
        |
        +-- maps-to -- CAPEC
                         |
                  related-weakness
                         |
                         v
                        CWE
                         ^
                         |
                  related-weakness
                         |
                        CVE ----> CPE
                         ^
                         |
                   EPSS (score)
                   KEV (exploited)
                   GHSA (advisory)
                   Vulnrichment (SSVC)
                   ExploitDB (exploit)
                   Nuclei (detection template)
                   EUVD (EU advisory)
                   OSV (open-source vuln)

git clone https://github.com/S0UGATA/security-kg.git
cd security-kg
pip install -r requirements.txt
python src/convert.py

neighborhoods/
  T1059.json              # array of Triple objects (depth=2, limit=500)
  T1059.001.json
  CVE-2024-1234.json
  CAPEC-100.json
  ...
  index.json              # { source, fingerprint, depth, limit, entities: [...] }

import pandas as pd
from datasets import load_dataset

# Load combined graph for cross-source queries
ds = load_dataset("s0u9ata/security-kg", "combined")
df = ds["train"].to_pandas()

# Build SSVC triage matrix: exploitation status × automatable × EPSS score
ssvc = df[df.predicate == "ssvc-exploitation"][["subject", "object"]].rename(columns={"object": "exploitation"})
auto = df[df.predicate == "ssvc-automatable"][["subject", "object"]].rename(columns={"object": "automatable"})
epss = df[df.predicate == "epss-score"][["subject", "object"]].copy()
epss["epss"] = epss.object.astype(float)

triage = ssvc.merge(auto, on="subject").merge(epss[["subject", "epss"]], on="subject")

# Highest priority: actively exploited + automatable + high EPSS
critical = triage[(triage.exploitation == "active") & (triage.automatable == "yes") & (triage.epss > 0.9)]
print(f"Immediate action: {len(critical)} CVEs")

# Find ATT&CK techniques heavily used by APT groups but poorly covered by detections
uses = df[(df.predicate == "uses") & df.subject.str.startswith("G")]
group_usage = uses.groupby("object").subject.nunique().rename("groups_using")

# Count detection sources per technique (Sigma + CAR + D3FEND + ENGAGE)
sigma = df[df.predicate == "detects-technique"].groupby("object").subject.nunique().rename("detections")
d3fend = df[df.predicate == "restricts"].groupby("object").subject.nunique().rename("defenses")

coverage = pd.DataFrame(group_usage).join(sigma).join(d3fend).fillna(0)
gaps = coverage[(coverage.groups_using > 10) & (coverage.detections < 5)]
print(f"High-usage, low-detection techniques: {len(gaps)}")

# Score open-source packages by aggregating risk from linked CVEs
ghsa_cve = df[df.predicate == "related-cve"][["subject", "object"]].rename(columns={"subject": "ghsa", "object": "cve"})
packages = df[df.predicate == "affects-package"][["subject", "object"]].rename(columns={"subject": "ghsa", "object": "pkg"})

epss_scores = df[df.predicate == "epss-score"][["subject", "object"]].copy()
epss_scores["epss"] = epss_scores.object.astype(float)

kev_cves = set(df[(df.predicate == "rdf:type") & (df.object == "KnownExploitedVulnerability")].subject)
exploit_cves = set(df[df.predicate == "exploits-cve"].object)

# Join package → GHSA → CVE → enrichment
risk = packages.merge(ghsa_cve, on="ghsa").merge(epss_scores[["subject", "epss"]], left_on="cve", right_on="subject")
risk["in_kev"] = risk.cve.isin(kev_cves)
risk["has_exploit"] = risk.cve.isin(exploit_cves)
risk["ecosystem"] = risk.pkg.str.split("/").str[0]

# Top ecosystems by high-risk CVE count
high_risk = risk[(risk.epss > 0.5) | risk.in_kev | risk.has_exploit]
print(high_risk.groupby("ecosystem").cve.nunique().sort_values(ascending=False).head(10))

capec = load_dataset("s0u9ata/security-kg", "capec")["train"].to_pandas()
cve = load_dataset("s0u9ata/security-kg", "cve")["train"].to_pandas()

# Find CWEs related to SQL Injection (CAPEC-66)
cwe_ids = capec[(capec.subject == "CAPEC-66") & (capec.predicate == "related-weakness")].object.tolist()

# Find CVEs with those CWEs
for cwe_id in cwe_ids:
    related_cves = cve[(cve.predicate == "related-weakness") & (cve.object == cwe_id)].subject.unique()
    print(f"{cwe_id}: {len(related_cves)} CVEs")

ds = load_dataset("s0u9ata/security-kg", "d3fend")
df = ds["train"].to_pandas()

# All 497 defensive techniques in the D3FEND taxonomy
defenses = df[(df.predicate == "rdf:type") & (df.object == "DefensiveTechnique")]
print(f"Defensive techniques: {len(defenses)}")

# Find children of a category (e.g., all techniques under Network Traffic Analysis)
children = df[(df.predicate == "child-of") & (df.object == "NetworkTrafficAnalysis")].subject.tolist()

# Get their names
names = df[df.predicate == "name"][["subject", "object"]]
print(names[names.subject.isin(children)].to_string(index=False))