init commit

.env.example

@@ -0,0 +1,6 @@
+# Tanium API Gateway URL
+# e.g. https://<customer>-api.cloud.tanium.com/plugin/products/gateway/graphql
+TANIUM_API_URL=
+
+# Tanium API Token
+TANIUM_API_TOKEN=

.gitignore

@@ -0,0 +1,26 @@
+# Environment
+.env
+
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+*.egg-info/
+dist/
+build/
+*.egg
+
+# Virtual environments
+venv/
+.venv/
+env/
+
+# IDE
+.vscode/
+.idea/
+*.swp
+*.swo
+
+# Output
+output/

client.py

@@ -0,0 +1,143 @@
+import time
+import requests
+
+import config
+from models import ComplianceFinding, CveFinding, EndpointCompliance
+
+ENDPOINTS_QUERY = """
+query getEndpoints($first: Int!, $after: String) {
+  endpoints(first: $first, after: $after) {
+    pageInfo {
+      endCursor
+      hasNextPage
+    }
+    edges {
+      node {
+        id
+        name
+        computerID
+        ipAddress
+        os {
+          name
+        }
+        compliance {
+          complianceFindings {
+            id
+            standard
+            standardVersion
+            state
+            ruleId
+          }
+          cveFindings {
+            cveId
+            cveYear
+            summary
+            severity
+            cvssScore
+          }
+        }
+      }
+    }
+  }
+}
+"""
+
+MAX_RETRIES = 3
+RETRY_BACKOFF = 2  # seconds, doubles each retry
+
+
+def _make_request(query, variables):
+    headers = {
+        "Content-Type": "application/json",
+        "session": config.TANIUM_API_TOKEN,
+    }
+    payload = {"query": query, "variables": variables}
+
+    for attempt in range(MAX_RETRIES):
+        resp = requests.post(config.TANIUM_API_URL, json=payload, headers=headers, timeout=60)
+
+        if resp.status_code == 429 or resp.status_code >= 500:
+            if attempt < MAX_RETRIES - 1:
+                wait = RETRY_BACKOFF * (2 ** attempt)
+                print(f"  Retrying in {wait}s (HTTP {resp.status_code})...")
+                time.sleep(wait)
+                continue
+            resp.raise_for_status()
+
+        resp.raise_for_status()
+        data = resp.json()
+
+        if "errors" in data:
+            error_messages = [e.get("message", str(e)) for e in data["errors"]]
+            raise RuntimeError(f"GraphQL errors: {'; '.join(error_messages)}")
+
+        return data["data"]
+
+    raise RuntimeError("Max retries exceeded")
+
+
+def _parse_endpoint(node):
+    os_info = node.get("os") or {}
+    compliance = node.get("compliance") or {}
+
+    compliance_findings = [
+        ComplianceFinding(
+            id=f["id"],
+            standard=f["standard"],
+            standard_version=f["standardVersion"],
+            state=f["state"],
+            rule_id=f["ruleId"],
+        )
+        for f in (compliance.get("complianceFindings") or [])
+    ]
+
+    cve_findings = [
+        CveFinding(
+            cve_id=f["cveId"],
+            cve_year=f["cveYear"],
+            summary=f["summary"],
+            severity=f["severity"],
+            cvss_score=f.get("cvssScore"),
+        )
+        for f in (compliance.get("cveFindings") or [])
+    ]
+
+    return EndpointCompliance(
+        endpoint_id=node["id"],
+        name=node.get("name", ""),
+        computer_id=node.get("computerID", ""),
+        ip_address=node.get("ipAddress", ""),
+        os_name=os_info.get("name", ""),
+        compliance_findings=compliance_findings,
+        cve_findings=cve_findings,
+    )
+
+
+def fetch_all_endpoints(page_size=500):
+    endpoints = []
+    cursor = None
+    page = 0
+
+    while True:
+        page += 1
+        variables = {"first": page_size}
+        if cursor:
+            variables["after"] = cursor
+
+        print(f"  Fetching page {page} (cursor: {cursor or 'start'})...")
+        data = _make_request(ENDPOINTS_QUERY, variables)
+
+        edges = data["endpoints"]["edges"]
+        page_info = data["endpoints"]["pageInfo"]
+
+        for edge in edges:
+            endpoints.append(_parse_endpoint(edge["node"]))
+
+        print(f"  Got {len(edges)} endpoints (total: {len(endpoints)})")
+
+        if not page_info["hasNextPage"]:
+            break
+
+        cursor = page_info["endCursor"]
+
+    return endpoints

config.py

@@ -0,0 +1,20 @@
+import sys
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+TANIUM_API_URL = os.getenv("TANIUM_API_URL")
+TANIUM_API_TOKEN = os.getenv("TANIUM_API_TOKEN")
+
+
+def validate():
+    missing = []
+    if not TANIUM_API_URL:
+        missing.append("TANIUM_API_URL")
+    if not TANIUM_API_TOKEN:
+        missing.append("TANIUM_API_TOKEN")
+    if missing:
+        print(f"Error: Missing required environment variables: {', '.join(missing)}")
+        print("Copy .env.example to .env and fill in your values.")
+        sys.exit(1)

export.py

@@ -0,0 +1,50 @@
+import csv
+import json
+import os
+from dataclasses import asdict
+
+
+def export_json(endpoints, filepath):
+    os.makedirs(os.path.dirname(filepath), exist_ok=True)
+    with open(filepath, "w") as f:
+        json.dump([asdict(ep) for ep in endpoints], f, indent=2)
+    print(f"  JSON exported to {filepath}")
+
+
+def export_csv(endpoints, output_dir):
+    os.makedirs(output_dir, exist_ok=True)
+
+    compliance_path = os.path.join(output_dir, "compliance_findings.csv")
+    cve_path = os.path.join(output_dir, "cve_findings.csv")
+
+    # Compliance findings CSV
+    with open(compliance_path, "w", newline="") as f:
+        writer = csv.writer(f)
+        writer.writerow([
+            "endpoint_id", "endpoint_name", "computer_id", "ip_address", "os_name",
+            "finding_id", "standard", "standard_version", "state", "rule_id",
+        ])
+        for ep in endpoints:
+            for finding in ep.compliance_findings:
+                writer.writerow([
+                    ep.endpoint_id, ep.name, ep.computer_id, ep.ip_address, ep.os_name,
+                    finding.id, finding.standard, finding.standard_version,
+                    finding.state, finding.rule_id,
+                ])
+    print(f"  Compliance findings CSV exported to {compliance_path}")
+
+    # CVE findings CSV
+    with open(cve_path, "w", newline="") as f:
+        writer = csv.writer(f)
+        writer.writerow([
+            "endpoint_id", "endpoint_name", "computer_id", "ip_address", "os_name",
+            "cve_id", "cve_year", "summary", "severity", "cvss_score",
+        ])
+        for ep in endpoints:
+            for finding in ep.cve_findings:
+                writer.writerow([
+                    ep.endpoint_id, ep.name, ep.computer_id, ep.ip_address, ep.os_name,
+                    finding.cve_id, finding.cve_year, finding.summary,
+                    finding.severity, finding.cvss_score,
+                ])
+    print(f"  CVE findings CSV exported to {cve_path}")

main.py

@@ -0,0 +1,66 @@
+import argparse
+import os
+
+import config
+from client import fetch_all_endpoints
+from export import export_csv, export_json
+
+
+def display_summary(endpoints):
+    total_compliance = sum(len(ep.compliance_findings) for ep in endpoints)
+    total_cve = sum(len(ep.cve_findings) for ep in endpoints)
+
+    print("\n=== Summary ===")
+    print(f"  Endpoints:           {len(endpoints)}")
+    print(f"  Compliance findings: {total_compliance}")
+    print(f"  CVE findings:        {total_cve}")
+
+    if endpoints:
+        # Top 5 endpoints by CVE count
+        by_cve = sorted(endpoints, key=lambda ep: len(ep.cve_findings), reverse=True)
+        print("\n  Top endpoints by CVE count:")
+        for ep in by_cve[:5]:
+            if ep.cve_findings:
+                print(f"    {ep.name} ({ep.ip_address}): {len(ep.cve_findings)} CVEs")
+
+        # Severity breakdown
+        severity_counts = {}
+        for ep in endpoints:
+            for cve in ep.cve_findings:
+                severity_counts[cve.severity] = severity_counts.get(cve.severity, 0) + 1
+        if severity_counts:
+            print("\n  CVE severity breakdown:")
+            for severity, count in sorted(severity_counts.items()):
+                print(f"    {severity}: {count}")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Fetch Tanium compliance and CVE findings")
+    parser.add_argument("--output-dir", default="./output", help="Directory for export files (default: ./output)")
+    parser.add_argument("--format", choices=["json", "csv", "both"], default="both", help="Export format (default: both)")
+    parser.add_argument("--page-size", type=int, default=500, help="Endpoints per page (default: 500)")
+    parser.add_argument("--display", action="store_true", help="Print summary table to console")
+    args = parser.parse_args()
+
+    config.validate()
+
+    print("Fetching endpoints with compliance data...")
+    endpoints = fetch_all_endpoints(page_size=args.page_size)
+    print(f"Fetched {len(endpoints)} endpoints.")
+
+    if args.display:
+        display_summary(endpoints)
+
+    if args.format in ("json", "both"):
+        print("\nExporting JSON...")
+        export_json(endpoints, os.path.join(args.output_dir, "endpoints.json"))
+
+    if args.format in ("csv", "both"):
+        print("\nExporting CSV...")
+        export_csv(endpoints, args.output_dir)
+
+    print("\nDone.")
+
+
+if __name__ == "__main__":
+    main()

models.py

@@ -0,0 +1,31 @@
+from dataclasses import dataclass, field
+from typing import Optional
+
+
+@dataclass
+class ComplianceFinding:
+    id: str
+    standard: str
+    standard_version: str
+    state: str
+    rule_id: str
+
+
+@dataclass
+class CveFinding:
+    cve_id: str
+    cve_year: int
+    summary: str
+    severity: str
+    cvss_score: Optional[float]
+
+
+@dataclass
+class EndpointCompliance:
+    endpoint_id: str
+    name: str
+    computer_id: str
+    ip_address: str
+    os_name: str
+    compliance_findings: list[ComplianceFinding] = field(default_factory=list)
+    cve_findings: list[CveFinding] = field(default_factory=list)

requirements.txt

@@ -0,0 +1,2 @@
+requests
+python-dotenv