Python Integration Guide
Official Python SDK
The official SDK wraps the API with typed results, pagination, automatic retry on 429, and helpers for asset correlation and Windows patch level (fetch every CVE page, send Windows hosts in batches of 100):
pip install "git+https://github.com/Citedrelevance/[email protected]"
import breachspider
bs = breachspider.Client("bs_live_...")
resp = bs.correlate.correlate([{"asset_id": "EXAMPLE-SWITCH-01", "vendor": "Moxa",
"product": "EDS-518A", "version": "V3.5"}])
for cve in resp.results[0].cves:
print(cve.priority_rank, cve.cve_id, cve.priority_reason)
Source, guides, examples and an OpenAPI document: https://github.com/Citedrelevance/breachspider-python.
Without the SDK
Complete Python examples for common BreachSpider workflows using requests directly.
Installation
pip install requests
Client Setup
import requests
from typing import Optional, Iterator
class BreachSpiderClient:
BASE_URL = "https://breachspider.com/api/v1"
def __init__(self, api_key: str):
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"Accept": "application/json",
})
def get_cve(self, cve_id: str) -> dict:
response = self.session.get(f"{self.BASE_URL}/cves/{cve_id}")
response.raise_for_status()
return response.json()["data"]
def search_cves(
self,
severity: Optional[str] = None,
kev_only: bool = False,
vendor: Optional[str] = None,
limit: int = 20,
) -> Iterator[dict]:
page = 1
while True:
params = {
"page": page,
"per_page": limit,
}
if kev_only:
params["kev"] = True
if severity:
params["severity"] = severity
if vendor:
params["vendor"] = vendor
response = self.session.get(f"{self.BASE_URL}/cves", params=params)
response.raise_for_status()
data = response.json()
yield from data["data"]
if not data["pagination"]["has_next"]:
break
page += 1
def get_kev(self, limit: int = 100) -> list:
response = self.session.get(
f"{self.BASE_URL}/cves/kev",
params={"per_page": limit}
)
response.raise_for_status()
return response.json()["data"]
def get_dashboard(self) -> dict:
response = self.session.get(f"{self.BASE_URL}/dashboard")
response.raise_for_status()
return response.json()["data"]
def get_environments(self) -> list:
response = self.session.get(f"{self.BASE_URL}/environments")
response.raise_for_status()
return response.json()["data"]
def create_environment(self, name: str, description: str = "") -> dict:
response = self.session.post(
f"{self.BASE_URL}/environments",
json={"name": name, "description": description}
)
response.raise_for_status()
return response.json()["data"]
def add_asset(self, env_id: int, asset: dict) -> dict:
response = self.session.post(
f"{self.BASE_URL}/environments/{env_id}/assets",
json=asset
)
response.raise_for_status()
return response.json()["data"]
Common Workflows
Fetch All KEV Entries for Your Stack
client = BreachSpiderClient("bs_live_your_key_here")
kev_entries = client.get_kev(limit=200)
critical_kev = [
cve for cve in kev_entries
if cve["cvss_score"] and cve["cvss_score"] >= 9.0
]
print(f"Total KEV: {len(kev_entries)}")
print(f"Critical KEV: {len(critical_kev)}")
for cve in critical_kev[:5]:
print(f"{cve['cve_id']} | CVSS {cve['cvss_score']} | {cve['primary_vendor']}")
Monitor Your Environment for New Critical CVEs
import time
from datetime import datetime
def monitor_environment(client, env_id: int, check_interval_seconds: int = 300):
while True:
response = client.session.get(
f"{client.BASE_URL}/environments/{env_id}/summary"
)
summary = response.json()["data"]
critical_count = summary.get("critical_count", 0)
kev_count = summary.get("kev_count", 0)
print(f"[{datetime.utcnow().isoformat()}] "
f"Critical: {critical_count} | KEV: {kev_count}")
time.sleep(check_interval_seconds)
Export CVE Report to CSV
import csv
def export_cves_to_csv(client, output_file: str, severity: str = "CRITICAL"):
with open(output_file, "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=[
"cve_id", "bsid", "severity", "cvss_score", "bcs_score",
"epss_percentile", "kev_flagged", "exploit_maturity",
"patch_status", "primary_vendor", "published_at"
])
writer.writeheader()
for cve in client.search_cves(severity=severity):
writer.writerow({
"cve_id": cve["cve_id"],
"bsid": cve.get("bsid", ""),
"severity": cve["severity"],
"cvss_score": cve["cvss_score"],
"bcs_score": cve.get("bcs_score", ""),
"epss_percentile": cve.get("epss_percentile", ""),
"kev_flagged": cve["kev_flagged"],
"exploit_maturity": cve.get("exploit_maturity", ""),
"patch_status": cve["patch_status"],
"primary_vendor": cve.get("primary_vendor", ""),
"published_at": cve["published_at"],
})
print(f"Exported to {output_file}")
Webhook Consumer Server
from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib
app = FastAPI()
WEBHOOK_SECRET = "your-webhook-secret"
def verify_signature(payload: bytes, signature: str) -> bool:
# The HMAC key is the SHA-256 hex digest of your webhook secret.
key = hashlib.sha256(WEBHOOK_SECRET.encode()).hexdigest().encode()
expected = "sha256=" + hmac.new(key, payload, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, signature)
@app.post("/breachspider-webhook")
async def handle_webhook(request: Request):
payload = await request.body()
signature = request.headers.get("X-BreachSpider-Signature", "")
if not verify_signature(payload, signature):
raise HTTPException(status_code=401, detail="Invalid signature")
event = await request.json()
event_type = event["event"]
cve_data = event["data"]
if event_type == "cve.alert": # the only event type
label = "Known-exploited" if cve_data["kev_flagged"] else cve_data["sage_priority"]
print(f"{label}: {cve_data['cve_id']} | CVSS {cve_data['cvss_score']}")
return {"status": "received"}