#!/usr/bin/env python3
"""Vulnerability dashboard builder using DefectDojo API.

Queries DefectDojo REST API v2 for findings, products, and engagements
to build vulnerability management dashboards and metrics.
"""

import json
import datetime
import os
import collections

try:
    import requests
    HAS_REQUESTS = True
except ImportError:
    HAS_REQUESTS = False


class DefectDojoClient:
    """Client for DefectDojo REST API v2."""

    def __init__(self, url=None, api_key=None):
        self.url = (url or os.environ.get("DEFECTDOJO_URL", "http://localhost:8080")).rstrip("/")
        self.api_key = api_key or os.environ.get("DEFECTDOJO_API_KEY", "")
        self.headers = {
            "Authorization": "Token " + self.api_key,
            "Content-Type": "application/json",
        }

    def _get(self, endpoint, params=None):
        if not HAS_REQUESTS or not self.api_key:
            return {"error": "requests not available or no API key"}
        try:
            resp = requests.get(
                self.url + "/api/v2/" + endpoint,
                headers=self.headers, params=params, timeout=15
            )
            if resp.status_code == 200:
                return resp.json()
            return {"error": "HTTP {}".format(resp.status_code)}
        except Exception as e:
            return {"error": str(e)}

    def get_findings(self, severity=None, active=True, limit=100):
        params = {"active": active, "limit": limit}
        if severity:
            params["severity"] = severity
        return self._get("findings/", params)

    def get_products(self, limit=100):
        return self._get("products/", {"limit": limit})

    def get_engagements(self, product_id=None, limit=100):
        params = {"limit": limit}
        if product_id:
            params["product"] = product_id
        return self._get("engagements/", params)

    def get_finding_count_by_severity(self):
        result = {}
        for sev in ["Critical", "High", "Medium", "Low", "Info"]:
            data = self._get("findings/", {"severity": sev, "active": True, "limit": 1})
            if isinstance(data, dict) and "count" in data:
                result[sev] = data["count"]
        return result

    def import_scan(self, product_id, engagement_id, scan_type, file_path):
        if not HAS_REQUESTS or not self.api_key:
            return {"error": "requests not available or no API key"}
        try:
            with open(file_path, "rb") as f:
                resp = requests.post(
                    self.url + "/api/v2/import-scan/",
                    headers={"Authorization": "Token " + self.api_key},
                    data={
                        "product": product_id,
                        "engagement": engagement_id,
                        "scan_type": scan_type,
                        "active": True,
                        "verified": False,
                    },
                    files={"file": f},
                    timeout=60,
                )
            if resp.status_code in (200, 201):
                return resp.json()
            return {"error": "HTTP {}".format(resp.status_code)}
        except Exception as e:
            return {"error": str(e)}


def build_dashboard_data(findings):
    """Build dashboard metrics from findings list."""
    if not isinstance(findings, dict) or "results" not in findings:
        return {"error": "Invalid findings data"}

    results = findings["results"]
    severity_counts = collections.Counter()
    product_counts = collections.Counter()
    age_sum = 0
    overdue_count = 0
    now = datetime.datetime.now(datetime.timezone.utc)

    for f in results:
        severity_counts[f.get("severity", "Unknown")] += 1
        product_counts[f.get("test", {}).get("engagement", {}).get("product", {}).get("name", "Unknown")] += 1
        if f.get("date"):
            try:
                created = datetime.datetime.fromisoformat(f["date"])
                if created.tzinfo is None:
                    created = created.replace(tzinfo=datetime.timezone.utc)
                age = (now - created).days
                age_sum += age
                sla = {"Critical": 7, "High": 30, "Medium": 90, "Low": 180}.get(f.get("severity", ""), 999)
                if age > sla:
                    overdue_count += 1
            except ValueError:
                pass

    total = len(results)
    return {
        "total_active_findings": total,
        "by_severity": dict(severity_counts),
        "by_product": dict(product_counts.most_common(10)),
        "avg_age_days": round(age_sum / max(total, 1), 1),
        "overdue_count": overdue_count,
        "sla_compliance_pct": round((total - overdue_count) / max(total, 1) * 100, 1),
    }


SUPPORTED_SCAN_TYPES = [
    "Nessus Scan", "Qualys Scan", "Burp REST API",
    "ZAP Scan", "Trivy Scan", "Snyk Scan",
    "Semgrep JSON Report", "SARIF", "Generic Findings Import",
    "Anchore Grype", "Nuclei Scan", "Checkov Scan",
]


if __name__ == "__main__":
    print("=" * 60)
    print("Vulnerability Dashboard with DefectDojo")
    print("REST API v2 queries, severity metrics, SLA tracking")
    print("=" * 60)
    print("  requests available: {}".format(HAS_REQUESTS))

    client = DefectDojoClient()

    print("\n--- Supported Scan Types ---")
    for st in SUPPORTED_SCAN_TYPES:
        print("  - {}".format(st))

    print("\n--- API Endpoints ---")
    endpoints = [
        ("GET", "/api/v2/findings/", "List findings"),
        ("GET", "/api/v2/products/", "List products"),
        ("GET", "/api/v2/engagements/", "List engagements"),
        ("POST", "/api/v2/import-scan/", "Import scan results"),
        ("POST", "/api/v2/reimport-scan/", "Re-import scan results"),
    ]
    for method, path, desc in endpoints:
        print("  {} {:30s} {}".format(method, path, desc))

    demo_findings = {
        "count": 5,
        "results": [
            {"severity": "Critical", "title": "SQL Injection", "date": "2025-01-10", "test": {"engagement": {"product": {"name": "WebApp"}}}},
            {"severity": "High", "title": "XSS", "date": "2025-01-15", "test": {"engagement": {"product": {"name": "WebApp"}}}},
            {"severity": "Medium", "title": "Missing Headers", "date": "2024-12-01", "test": {"engagement": {"product": {"name": "API"}}}},
            {"severity": "Low", "title": "Cookie flag", "date": "2025-02-01", "test": {"engagement": {"product": {"name": "API"}}}},
            {"severity": "Critical", "title": "RCE", "date": "2025-02-20", "test": {"engagement": {"product": {"name": "WebApp"}}}},
        ],
    }

    dashboard = build_dashboard_data(demo_findings)
    print("\n--- Dashboard ---")
    for k, v in dashboard.items():
        print("  {}: {}".format(k, v))

    print("\n" + json.dumps({"findings_analyzed": demo_findings["count"]}, indent=2))

#!/usr/bin/env python3
"""DefectDojo Vulnerability Dashboard Automation.

Manages products, engagements, scan imports, and metrics via the
DefectDojo REST API v2.
"""

import argparse
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

import requests

DD_URL = os.environ.get("DD_URL", "http://localhost:8080/api/v2")
DD_API_KEY = os.environ.get("DD_API_KEY", "")


def get_headers():
    return {
        "Authorization": f"Token {DD_API_KEY}",
        "Content-Type": "application/json",
    }


def create_product_type(name, description=""):
    resp = requests.post(
        f"{DD_URL}/product_types/",
        headers=get_headers(),
        json={"name": name, "description": description},
        timeout=30,
    )
    resp.raise_for_status()
    pt = resp.json()
    print(f"[+] Created product type: {name} (ID: {pt['id']})")
    return pt["id"]


def create_product(name, product_type_id, description=""):
    resp = requests.post(
        f"{DD_URL}/products/",
        headers=get_headers(),
        json={
            "name": name,
            "description": description,
            "prod_type": product_type_id,
        },
        timeout=30,
    )
    resp.raise_for_status()
    product = resp.json()
    print(f"[+] Created product: {name} (ID: {product['id']})")
    return product["id"]


def create_engagement(name, product_id, start_date=None, end_date=None):
    if not start_date:
        start_date = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    if not end_date:
        end_date = "2025-12-31"
    resp = requests.post(
        f"{DD_URL}/engagements/",
        headers=get_headers(),
        json={
            "name": name,
            "product": product_id,
            "target_start": start_date,
            "target_end": end_date,
            "engagement_type": "CI/CD",
            "status": "In Progress",
        },
        timeout=30,
    )
    resp.raise_for_status()
    eng = resp.json()
    print(f"[+] Created engagement: {name} (ID: {eng['id']})")
    return eng["id"]


def import_scan(scan_file, scan_type, product_name, engagement_name=None):
    """Import or reimport scan results into DefectDojo."""
    data = {
        "scan_type": scan_type,
        "product_name": product_name,
        "auto_create_context": "true",
        "deduplication_on_engagement": "true",
        "close_old_findings": "true",
    }
    if engagement_name:
        data["engagement_name"] = engagement_name

    with open(scan_file, "rb") as f:
        resp = requests.post(
            f"{DD_URL}/reimport-scan/",
            headers={"Authorization": f"Token {DD_API_KEY}"},
            data=data,
            files={"file": f},
            timeout=120,
        )

    if resp.status_code in (200, 201):
        result = resp.json()
        test_id = result.get("test", 0)
        print(f"[+] Scan imported successfully (Test ID: {test_id})")
        print(f"    New findings: {result.get('statistics', {}).get('created', 0)}")
        print(f"    Closed findings: {result.get('statistics', {}).get('closed', 0)}")
        print(f"    Reactivated: {result.get('statistics', {}).get('reactivated', 0)}")
        return result
    else:
        print(f"[-] Import failed: {resp.status_code} {resp.text}")
        return None


def get_findings(product_id=None, severity=None, active=True, limit=100):
    """Query findings with filters."""
    params = {"limit": limit, "active": str(active).lower()}
    if product_id:
        params["test__engagement__product"] = product_id
    if severity:
        params["severity"] = severity

    resp = requests.get(f"{DD_URL}/findings/", headers=get_headers(), params=params, timeout=30)
    resp.raise_for_status()
    return resp.json()


def get_metrics(product_id=None):
    """Get vulnerability metrics for dashboard."""
    params = {"limit": 0}
    if product_id:
        params["test__engagement__product"] = product_id

    metrics = {}
    for severity in ["Critical", "High", "Medium", "Low", "Info"]:
        resp = requests.get(
            f"{DD_URL}/findings/",
            headers=get_headers(),
            params={**params, "severity": severity, "active": "true"},
            timeout=30,
        )
        if resp.status_code == 200:
            metrics[severity] = resp.json().get("count", 0)

    # SLA breached findings
    resp = requests.get(
        f"{DD_URL}/findings/",
        headers=get_headers(),
        params={**params, "active": "true", "is_mitigated": "false"},
        timeout=30,
    )
    if resp.status_code == 200:
        metrics["total_active"] = resp.json().get("count", 0)

    return metrics


def generate_dashboard_report(output_path, product_id=None):
    """Generate dashboard metrics report."""
    metrics = get_metrics(product_id)
    report = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "active_findings": metrics,
        "total_active": metrics.get("total_active", 0),
    }
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2)
    print(f"\n[+] Dashboard Report: {output_path}")
    print(f"    Critical: {metrics.get('Critical', 0)}")
    print(f"    High: {metrics.get('High', 0)}")
    print(f"    Medium: {metrics.get('Medium', 0)}")
    print(f"    Low: {metrics.get('Low', 0)}")
    print(f"    Total Active: {metrics.get('total_active', 0)}")
    return report


def main():
    parser = argparse.ArgumentParser(description="DefectDojo Dashboard Automation")
    parser.add_argument("--url", default=DD_URL, help="DefectDojo API URL")
    parser.add_argument("--api-key", default=DD_API_KEY, help="API key")

    sub = parser.add_subparsers(dest="command")

    setup = sub.add_parser("setup", help="Create product type, product, engagement")
    setup.add_argument("--product-type", required=True)
    setup.add_argument("--product", required=True)
    setup.add_argument("--engagement", default="CI/CD")

    imp = sub.add_parser("import", help="Import scan results")
    imp.add_argument("--file", required=True)
    imp.add_argument("--scan-type", required=True)
    imp.add_argument("--product", required=True)
    imp.add_argument("--engagement")

    dash = sub.add_parser("dashboard", help="Generate dashboard report")
    dash.add_argument("--product-id", type=int)
    dash.add_argument("--output", default="defectdojo_dashboard.json")

    findings = sub.add_parser("findings", help="List findings")
    findings.add_argument("--product-id", type=int)
    findings.add_argument("--severity")
    findings.add_argument("--limit", type=int, default=20)

    args = parser.parse_args()

    global DD_URL, DD_API_KEY
    DD_URL = args.url
    if args.api_key:
        DD_API_KEY = args.api_key

    if args.command == "setup":
        pt_id = create_product_type(args.product_type)
        prod_id = create_product(args.product, pt_id)
        create_engagement(args.engagement, prod_id)
    elif args.command == "import":
        import_scan(args.file, args.scan_type, args.product, args.engagement)
    elif args.command == "dashboard":
        generate_dashboard_report(args.output, args.product_id)
    elif args.command == "findings":
        result = get_findings(args.product_id, args.severity, limit=args.limit)
        for f in result.get("results", []):
            print(f"  [{f['severity']}] {f['title']} (ID: {f['id']})")
    else:
        parser.print_help()


if __name__ == "__main__":
    main()