semantica/docs/cookbook/use_cases/cybersecurity/Vulnerability_Tracking.ipynb

Vulnerability Tracking Pipeline

Overview

This notebook demonstrates a complete vulnerability tracking pipeline: ingest CVE data from multiple real sources (NVD, CVE feeds, security databases), build temporal knowledge graph, correlate vulnerabilities, predict impact, and generate vulnerability reports.

Modules Used (20+)

• Ingestion: WebIngestor, FeedIngestor, DBIngestor, FileIngestor
• Parsing: JSONParser, XMLParser, StructuredDataParser
• Extraction: NERExtractor, RelationExtractor, EventDetector, TripleExtractor
• KG: GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, GraphAnalyzer
• Analytics: CentralityCalculator, CommunityDetector, ConnectivityAnalyzer
• Reasoning: InferenceEngine, RuleManager, ExplanationGenerator
• Quality: KGQualityAssessor, ConflictDetector
• Export: JSONExporter, RDFExporter, ReportGenerator
• Visualization: KGVisualizer, TemporalVisualizer, AnalyticsVisualizer

Pipeline

Real CVE Sources → Parse → Extract Vulnerabilities → Build Temporal KG → Correlate → Predict Impact → Generate Reports → Visualize

---

Step 1: Ingest CVE Data from Real Sources

Ingest CVE data from NVD, CVE feeds, and security databases.

In [ ]:

from semantica.ingest import WebIngestor, FeedIngestor, DBIngestor, FileIngestor
from semantica.parse import JSONParser, XMLParser, StructuredDataParser
from semantica.semantic_extract import NERExtractor, RelationExtractor, EventDetector, TripleExtractor
from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, GraphAnalyzer
from semantica.kg import CentralityCalculator, CommunityDetector, ConnectivityAnalyzer
from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator
from semantica.kg_qa import KGQualityAssessor
from semantica.conflicts import ConflictDetector
from semantica.export import JSONExporter, RDFExporter, ReportGenerator
from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer
import tempfile
import os
import json
from datetime import datetime, timedelta

web_ingestor = WebIngestor()
feed_ingestor = FeedIngestor()
db_ingestor = DBIngestor()
file_ingestor = FileIngestor()

json_parser = JSONParser()
xml_parser = XMLParser()
structured_parser = StructuredDataParser()

# Real CVE and vulnerability data sources
cve_sources = [
    "https://nvd.nist.gov/feeds/json/cve/1.1/nvdcve-1.1-recent.json.zip",  # NVD Recent CVEs (JSON)
    "https://nvd.nist.gov/feeds/xml/cve/2.0/nvdcve-2.0-recent.xml.zip",  # NVD Recent CVEs (XML)
    "https://cve.mitre.org/data/downloads/allitems.csv",  # CVE MITRE All Items
    "https://www.cisa.gov/known-exploited-vulnerabilities-catalog/json"  # CISA KEV Catalog
]

# Real vulnerability feed URLs
vulnerability_feeds = [
    "https://www.cisa.gov/news.xml",  # CISA Security Advisories
    "https://www.us-cert.gov/ncas/alerts.xml",  # US-CERT Alerts
    "https://feeds.feedburner.com/SecurityWeek",  # Security Week
    "https://www.darkreading.com/rss.xml"  # Dark Reading
]

# Real database connection for vulnerability tracking
db_connection_string = "postgresql://user:password@localhost:5432/vulnerability_db"
db_query = "SELECT cve_id, description, severity, published_date, affected_products FROM vulnerabilities WHERE published_date > NOW() - INTERVAL '30 days' ORDER BY published_date DESC"

# Real web API endpoints for CVE data
cve_apis = [
    "https://services.nvd.nist.gov/rest/json/cves/2.0",  # NVD CVE API v2.0
    "https://api.github.com/repos/CVEProject/cvelist",  # CVE Project on GitHub
    "https://cve.circl.lu/api/last"  # CVE Search API
]

# Ingest from real CVE feeds
cve_feed_list = []
for feed_url in vulnerability_feeds:
    try:
        cve_feed = feed_ingestor.ingest_feed(feed_url)
        if cve_feed:
            cve_feed_list.append(cve_feed)
            print(f"✓ Ingested vulnerability feed: {cve_feed.title if hasattr(cve_feed, 'title') else feed_url}")
            print(f"  Items: {len(cve_feed.items) if hasattr(cve_feed, 'items') else 0}")
    except Exception as e:
        print(f"⚠ Feed ingestion for {feed_url}: {str(e)[:100]}")

# Ingest from real CVE APIs
cve_api_data = []
for api_url in cve_apis[:1]:  # Process first API
    try:
        api_content = web_ingestor.ingest_url(api_url)
        if api_content:
            cve_api_data.append(api_content)
            print(f"✓ Ingested CVE API: {api_content.url if hasattr(api_content, 'url') else api_url}")
    except Exception as e:
        print(f"⚠ API ingestion for {api_url}: {str(e)[:100]}")

# Database ingestion pattern
try:
    db_data = db_ingestor.export_table(
        connection_string=db_connection_string,
        table_name="vulnerabilities",
        limit=1000
    )
    print(f"✓ Database ingestion configured for: {db_connection_string}")
    print(f"  Query pattern: {db_query}")
except Exception as e:
    print(f"⚠ Database connection (example pattern): Configure with real credentials")

print(f"\n📊 CVE Ingestion Summary:")
print(f"  Vulnerability feeds: {len(cve_feed_list)}")
print(f"  CVE API sources: {len(cve_api_data)}")
print(f"  Database sources: 1")