From c2f1c498052730106adb5b0c4117eabf30dd6714 Mon Sep 17 00:00:00 2001 From: garretttolmanj Date: Thu, 16 Jul 2026 12:06:37 -0600 Subject: [PATCH] Add 13 Datadog evidence fetchers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports all Datadog fetchers from evidence-fetchers into the prmfy-fetchers framework. Creates fetchers/datadog/ with 13 single-target Python fetchers covering: SIEM detection rules, signals, and configuration; log pipelines, indexes, and archives; monitors; agent hosts, containers, and infra agent checks; APM services; incidents list and timelines. Each fetcher was ported to the v0.x contract: removed parse_fetcher_args in favor of load_dotenv + EVIDENCE_DIR, switched status output to logger.info/logger.warning, and added fetcher.yaml with secrets, evidence_set block, and EVD-DD-* reference IDs. Fixed two fetchers (agent_hosts, infra_agent_checks) to re-raise HTTPError so auth failures surface as non-zero exits. Added not_available to the exit-0 set in the three SIEM fetchers. Renamed analysis → summary in all output payloads for consistency. Adds fetchers/_categories/datadog.yaml and fetchers/datadog/README.md with setup, permissions, manifest wiring, and credential rotation instructions. --- fetchers/_categories/datadog.yaml | 5 + fetchers/datadog/README.md | 210 ++++++++++++++++ fetchers/datadog/agent_hosts/fetcher.py | 235 ++++++++++++++++++ fetchers/datadog/agent_hosts/fetcher.yaml | 27 ++ fetchers/datadog/apm_services/fetcher.py | 150 +++++++++++ fetchers/datadog/apm_services/fetcher.yaml | 27 ++ fetchers/datadog/containers/fetcher.py | 205 +++++++++++++++ fetchers/datadog/containers/fetcher.yaml | 27 ++ .../datadog/incident_timelines/fetcher.py | 200 +++++++++++++++ .../datadog/incident_timelines/fetcher.yaml | 28 +++ fetchers/datadog/incidents_list/fetcher.py | 211 ++++++++++++++++ fetchers/datadog/incidents_list/fetcher.yaml | 28 +++ .../datadog/infra_agent_checks/fetcher.py | 232 +++++++++++++++++ .../datadog/infra_agent_checks/fetcher.yaml | 27 ++ fetchers/datadog/log_archives/fetcher.py | 142 +++++++++++ fetchers/datadog/log_archives/fetcher.yaml | 27 ++ fetchers/datadog/log_indexes/fetcher.py | 156 ++++++++++++ fetchers/datadog/log_indexes/fetcher.yaml | 27 ++ fetchers/datadog/log_pipelines/fetcher.py | 133 ++++++++++ fetchers/datadog/log_pipelines/fetcher.yaml | 27 ++ fetchers/datadog/monitors_list/fetcher.py | 147 +++++++++++ fetchers/datadog/monitors_list/fetcher.yaml | 27 ++ .../datadog/siem_configuration/fetcher.py | 164 ++++++++++++ .../datadog/siem_configuration/fetcher.yaml | 27 ++ .../datadog/siem_detection_rules/fetcher.py | 182 ++++++++++++++ .../datadog/siem_detection_rules/fetcher.yaml | 27 ++ fetchers/datadog/siem_signals/fetcher.py | 182 ++++++++++++++ fetchers/datadog/siem_signals/fetcher.yaml | 28 +++ manifest.yaml | 70 +++++- 29 files changed, 2975 insertions(+), 3 deletions(-) create mode 100644 fetchers/_categories/datadog.yaml create mode 100644 fetchers/datadog/README.md create mode 100644 fetchers/datadog/agent_hosts/fetcher.py create mode 100644 fetchers/datadog/agent_hosts/fetcher.yaml create mode 100644 fetchers/datadog/apm_services/fetcher.py create mode 100644 fetchers/datadog/apm_services/fetcher.yaml create mode 100644 fetchers/datadog/containers/fetcher.py create mode 100644 fetchers/datadog/containers/fetcher.yaml create mode 100644 fetchers/datadog/incident_timelines/fetcher.py create mode 100644 fetchers/datadog/incident_timelines/fetcher.yaml create mode 100644 fetchers/datadog/incidents_list/fetcher.py create mode 100644 fetchers/datadog/incidents_list/fetcher.yaml create mode 100644 fetchers/datadog/infra_agent_checks/fetcher.py create mode 100644 fetchers/datadog/infra_agent_checks/fetcher.yaml create mode 100644 fetchers/datadog/log_archives/fetcher.py create mode 100644 fetchers/datadog/log_archives/fetcher.yaml create mode 100644 fetchers/datadog/log_indexes/fetcher.py create mode 100644 fetchers/datadog/log_indexes/fetcher.yaml create mode 100644 fetchers/datadog/log_pipelines/fetcher.py create mode 100644 fetchers/datadog/log_pipelines/fetcher.yaml create mode 100644 fetchers/datadog/monitors_list/fetcher.py create mode 100644 fetchers/datadog/monitors_list/fetcher.yaml create mode 100644 fetchers/datadog/siem_configuration/fetcher.py create mode 100644 fetchers/datadog/siem_configuration/fetcher.yaml create mode 100644 fetchers/datadog/siem_detection_rules/fetcher.py create mode 100644 fetchers/datadog/siem_detection_rules/fetcher.yaml create mode 100644 fetchers/datadog/siem_signals/fetcher.py create mode 100644 fetchers/datadog/siem_signals/fetcher.yaml diff --git a/fetchers/_categories/datadog.yaml b/fetchers/_categories/datadog.yaml new file mode 100644 index 0000000..7ae1757 --- /dev/null +++ b/fetchers/_categories/datadog.yaml @@ -0,0 +1,5 @@ +--- +description: > + Datadog monitoring platform. Collects host inventory, container inventory, + APM service catalog, monitors, log archives/indexes/pipelines, SIEM detection + rules, SIEM signals, and incident records. diff --git a/fetchers/datadog/README.md b/fetchers/datadog/README.md new file mode 100644 index 0000000..f49884c --- /dev/null +++ b/fetchers/datadog/README.md @@ -0,0 +1,210 @@ +# Datadog + +Datadog fetchers pull security monitoring, logging, infrastructure, APM, and incident evidence from the Datadog API. The 13 fetchers in this category cover Cloud SIEM detection rules and signals, SIEM operational configuration, log pipelines, indexes, and archives, infrastructure host and container inventory, agent check results, APM service catalog, and incident records with timelines — providing broad coverage across KSIs for monitoring/logging, infrastructure, incident response, and availability. + +## Environment variables + +| Variable | Required | Description | +|---|---|---| +| `DATADOG_API_KEY` | Yes | Datadog API key (identifies the organization) | +| `DATADOG_APP_KEY` | Yes | Datadog application key (identifies the service account) | +| `DATADOG_BASE_URL` | No | API base URL (default: `https://api.ddog-gov.com`; use `https://api.datadoghq.com` for commercial) | +| `DATADOG_SIGNALS_LOOKBACK_DAYS` | No | Lookback window for SIEM signals (default: `30`) | +| `DATADOG_INCIDENTS_LOOKBACK_DAYS` | No | Lookback window for incident records (default: `90`) | + +## Creating a service account and API credentials + +Datadog requires two separate credentials: an **API Key** (identifies the organization) and an **Application Key** (identifies the user or service account making the request). Use a dedicated service account — not a personal account — so credentials survive personnel changes. + +### 1. Create a service account + +1. Log in to the Datadog console. +2. Navigate to **Organization Settings → Service Accounts**. +3. Click **New Service Account**. +4. Name it `paramify-evidence-fetchers`. +5. Assign the **Datadog Read Only** role — all fetchers are read-only. +6. Click **Create**. + +### 2. Generate an application key + +1. On the service account page, click **New Key**. +2. Name it `paramify-evidence-fetchers-appkey`. +3. Copy the application key value immediately — it is only shown once. +4. Store it in your secrets manager as `DATADOG_APP_KEY`. + +### 3. Generate an API key + +1. Navigate to **Organization Settings → API Keys**. +2. Click **New Key**. +3. Name it `paramify-evidence-fetchers-apikey`. +4. Copy the API key value. +5. Store it in your secrets manager as `DATADOG_API_KEY`. + +## Required permissions + +The service account role must include read access to: + +| Datadog Product | Required Permission | +|---|---| +| Cloud SIEM | `security_monitoring_rules_read`, `security_monitoring_signals_read` | +| Log Management | `logs_read_data`, `logs_read_index_data`, `logs_read_archives` | +| Infrastructure | `metrics_read`, `hosts_read` | +| APM | `apm_read` | +| Monitors | `monitors_read` | +| Incidents | `incident_read` *(only if using Datadog Incident Management)* | + +The built-in **Datadog Read Only** role covers all of these. Verify in **Organization Settings → Roles** that the role includes these scopes before running fetchers. + +## Wiring into a manifest + +All Datadog fetchers share the same three secrets (`api_key`, `app_key`, `base_url`). Two fetchers also take a lookback config secret. + +**SIEM** + +```bash +paramify manifest add datadog_siem_detection_rules +paramify manifest set-secret datadog_siem_detection_rules api_key DATADOG_API_KEY +paramify manifest set-secret datadog_siem_detection_rules app_key DATADOG_APP_KEY +paramify manifest set-secret datadog_siem_detection_rules base_url DATADOG_BASE_URL + +paramify manifest add datadog_siem_signals +paramify manifest set-secret datadog_siem_signals api_key DATADOG_API_KEY +paramify manifest set-secret datadog_siem_signals app_key DATADOG_APP_KEY +paramify manifest set-secret datadog_siem_signals base_url DATADOG_BASE_URL +paramify manifest set-secret datadog_siem_signals signals_lookback_days DATADOG_SIGNALS_LOOKBACK_DAYS + +paramify manifest add datadog_siem_configuration +paramify manifest set-secret datadog_siem_configuration api_key DATADOG_API_KEY +paramify manifest set-secret datadog_siem_configuration app_key DATADOG_APP_KEY +paramify manifest set-secret datadog_siem_configuration base_url DATADOG_BASE_URL +``` + +**Logging** + +```bash +paramify manifest add datadog_log_pipelines +paramify manifest set-secret datadog_log_pipelines api_key DATADOG_API_KEY +paramify manifest set-secret datadog_log_pipelines app_key DATADOG_APP_KEY +paramify manifest set-secret datadog_log_pipelines base_url DATADOG_BASE_URL + +paramify manifest add datadog_log_indexes +paramify manifest set-secret datadog_log_indexes api_key DATADOG_API_KEY +paramify manifest set-secret datadog_log_indexes app_key DATADOG_APP_KEY +paramify manifest set-secret datadog_log_indexes base_url DATADOG_BASE_URL + +paramify manifest add datadog_log_archives +paramify manifest set-secret datadog_log_archives api_key DATADOG_API_KEY +paramify manifest set-secret datadog_log_archives app_key DATADOG_APP_KEY +paramify manifest set-secret datadog_log_archives base_url DATADOG_BASE_URL +``` + +**Infrastructure** + +```bash +paramify manifest add datadog_monitors_list +paramify manifest set-secret datadog_monitors_list api_key DATADOG_API_KEY +paramify manifest set-secret datadog_monitors_list app_key DATADOG_APP_KEY +paramify manifest set-secret datadog_monitors_list base_url DATADOG_BASE_URL + +paramify manifest add datadog_agent_hosts +paramify manifest set-secret datadog_agent_hosts api_key DATADOG_API_KEY +paramify manifest set-secret datadog_agent_hosts app_key DATADOG_APP_KEY +paramify manifest set-secret datadog_agent_hosts base_url DATADOG_BASE_URL + +paramify manifest add datadog_containers +paramify manifest set-secret datadog_containers api_key DATADOG_API_KEY +paramify manifest set-secret datadog_containers app_key DATADOG_APP_KEY +paramify manifest set-secret datadog_containers base_url DATADOG_BASE_URL + +paramify manifest add datadog_infra_agent_checks +paramify manifest set-secret datadog_infra_agent_checks api_key DATADOG_API_KEY +paramify manifest set-secret datadog_infra_agent_checks app_key DATADOG_APP_KEY +paramify manifest set-secret datadog_infra_agent_checks base_url DATADOG_BASE_URL +``` + +**APM** + +```bash +paramify manifest add datadog_apm_services +paramify manifest set-secret datadog_apm_services api_key DATADOG_API_KEY +paramify manifest set-secret datadog_apm_services app_key DATADOG_APP_KEY +paramify manifest set-secret datadog_apm_services base_url DATADOG_BASE_URL +``` + +**Incidents** + +```bash +paramify manifest add datadog_incidents_list +paramify manifest set-secret datadog_incidents_list api_key DATADOG_API_KEY +paramify manifest set-secret datadog_incidents_list app_key DATADOG_APP_KEY +paramify manifest set-secret datadog_incidents_list base_url DATADOG_BASE_URL +paramify manifest set-secret datadog_incidents_list incidents_lookback_days DATADOG_INCIDENTS_LOOKBACK_DAYS + +paramify manifest add datadog_incident_timelines +paramify manifest set-secret datadog_incident_timelines api_key DATADOG_API_KEY +paramify manifest set-secret datadog_incident_timelines app_key DATADOG_APP_KEY +paramify manifest set-secret datadog_incident_timelines base_url DATADOG_BASE_URL +paramify manifest set-secret datadog_incident_timelines incidents_lookback_days DATADOG_INCIDENTS_LOOKBACK_DAYS +``` + +Use `paramify catalog` to see all available Datadog fetchers. + +## Smoke test + +```bash +curl -s \ + -H "DD-API-KEY: $DATADOG_API_KEY" \ + -H "DD-APPLICATION-KEY: $DATADOG_APP_KEY" \ + "$DATADOG_BASE_URL/api/v1/validate" | python3 -m json.tool +``` + +Expected response: + +```json +{"valid": true} +``` + +If `valid` is `false` or you receive a 403, the credentials are invalid or the service account role is missing required permissions. + +## Rotating credentials + +### Application key rotation + +1. Navigate to **Organization Settings → Service Accounts → paramify-evidence-fetchers**. +2. Click the existing key and select **Revoke**. +3. Click **New Key** and name it with today's date for traceability. +4. Copy the new value and update `DATADOG_APP_KEY` in your secrets manager. +5. Run the smoke test to confirm. + +### API key rotation + +1. Navigate to **Organization Settings → API Keys**. +2. Locate `paramify-evidence-fetchers-apikey` and click **Revoke**. +3. Click **New Key** and name it with today's date. +4. Copy the new value and update `DATADOG_API_KEY` in your secrets manager. +5. Run the smoke test to confirm. + +## Fetcher reference + +| Fetcher | What it collects | Evidence set ID | +|---|---|---| +| `datadog_siem_detection_rules` | Custom SIEM detection rules — enabled state, type, severity, last updated | `EVD-DD-SIEM-RULES` | +| `datadog_siem_signals` | Security signals generated by detection rules within the lookback window | `EVD-DD-SIEM-SIGNALS` | +| `datadog_siem_configuration` | Suppression rules and notification integrations (webhook domains only) | `EVD-DD-SIEM-CONFIG` | +| `datadog_log_pipelines` | Log processing pipeline configurations and filter queries | `EVD-DD-LOG-PIPELINES` | +| `datadog_log_indexes` | Log index retention settings and per-index filter queries | `EVD-DD-LOG-INDEXES` | +| `datadog_log_archives` | Long-term log archive destinations (S3, Azure, GCS) and state | `EVD-DD-LOG-ARCHIVES` | +| `datadog_monitors_list` | Monitor configurations — type, query, status, notification targets | `EVD-DD-MONITORS` | +| `datadog_agent_hosts` | Real-time host inventory — platform, cloud provider, agent version, mute state | `EVD-DD-AGENT-HOSTS` | +| `datadog_containers` | Real-time container inventory — image digest, state, namespace, environment | `EVD-DD-CONTAINERS` | +| `datadog_infra_agent_checks` | Agent check results across the host fleet — status per check per host | `EVD-DD-INFRA-CHECKS` | +| `datadog_apm_services` | APM service catalog — language, team, ownership, contact types | `EVD-DD-APM-SERVICES` | +| `datadog_incidents_list` | Incident records within the lookback window — severity, status, postmortem presence | `EVD-DD-INCIDENTS-LIST` | +| `datadog_incident_timelines` | Per-incident timeline cell metadata — cell types, timestamps, author count | `EVD-DD-INCIDENT-TIMELINES` | + +## Notes + +- Datadog GovCloud (`ddog-gov.com`) and commercial (`datadoghq.com`) are separate tenants with separate credentials. The fetchers default to GovCloud. Set `DATADOG_BASE_URL=https://api.datadoghq.com` for commercial tenants. +- Application keys are scoped to the creating service account. If the service account is deleted, the application key is invalidated immediately. Always use a dedicated service account, not a personal account. +- Datadog does not enforce API or application key expiration by default. Establish a rotation schedule (90 days is a common baseline) and track it outside the platform. +- The `incident_timelines` and `incidents_list` fetchers require Datadog Incident Management to be enabled on the tenant. If the feature is not active, both fetchers will return zero records without error. diff --git a/fetchers/datadog/agent_hosts/fetcher.py b/fetchers/datadog/agent_hosts/fetcher.py new file mode 100644 index 0000000..4e134b6 --- /dev/null +++ b/fetchers/datadog/agent_hosts/fetcher.py @@ -0,0 +1,235 @@ +#!/usr/bin/env python3 +""" +DataDog Agent Hosts Retrieval + +Purpose: Retrieve real-time inventory of all hosts reporting via DataDog agent. +""" + +import json +import logging +import os +import sys +from collections import Counter +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +import requests +from dotenv import load_dotenv + +logger = logging.getLogger("datadog_agent_hosts") + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def get_env(name: str, default: Optional[str] = None) -> str: + value = os.environ.get(name, default) + if value is None or value == "": + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def get_dd_headers() -> Dict[str, str]: + return { + "Content-Type": "application/json", + "DD-API-KEY": get_env("DATADOG_API_KEY"), + "DD-APPLICATION-KEY": get_env("DATADOG_APP_KEY"), + } + + +def get_base_url() -> str: + return os.environ.get("DATADOG_BASE_URL", "https://api.ddog-gov.com").rstrip("/") + + +def get_total_hosts(base_url: str, headers: Dict[str, str]) -> int: + try: + response = requests.get( + f"{base_url}/api/v1/hosts", + headers=headers, + params={"count": 1}, + timeout=30, + ) + response.raise_for_status() + return response.json().get("total_matching", 0) + except requests.exceptions.HTTPError: + raise + except Exception: + return 0 + + +def fetch_all_hosts(base_url: str, headers: Dict[str, str]) -> List[Dict[str, Any]]: + all_hosts: List[Dict[str, Any]] = [] + count = 1000 + start = 0 + endpoint = f"{base_url}/api/v1/hosts" + + total_matching = get_total_hosts(base_url, headers) + if total_matching == 0: + return [] + + while start < total_matching: + params = {"start": start, "count": count} + try: + response = requests.get(endpoint, headers=headers, params=params, timeout=60) + response.raise_for_status() + payload = response.json() + hosts = payload.get("host_list", []) + if not hosts: + break + all_hosts.extend(hosts) + start += count + except requests.exceptions.RequestException as e: + logger.warning("Pagination interrupted at start=%s: %s", start, e) + break + + return all_hosts + + +def parse_gohai(gohai_str: Optional[str]) -> Dict[str, Any]: + if not gohai_str: + return {} + try: + return json.loads(gohai_str) if isinstance(gohai_str, str) else gohai_str + except (json.JSONDecodeError, TypeError): + return {} + + +def detect_cloud_provider(host: Dict[str, Any], gohai: Dict[str, Any]) -> str: + # Prefer explicit gohai.cloud.provider when present (some agent configs set this) + provider = gohai.get("cloud", {}).get("provider", "") + if provider: + return provider + # EC2 instance IDs start with "i-" followed by hex chars + aliases = host.get("aliases", []) + if any(a.startswith("i-") and len(a) >= 10 for a in aliases): + return "aws" + # AWS internal DNS suffixes + name = host.get("name", "") + if ".compute.internal" in name or ".ec2.internal" in name: + return "aws" + # VPC ID in meta.network + meta_network = host.get("meta", {}).get("network", {}) + if isinstance(meta_network, dict) and str(meta_network.get("network-id", "")).startswith("vpc-"): + return "aws" + return "" + + +def extract_host_fields(host: Dict[str, Any]) -> Dict[str, Any]: + meta = host.get("meta", {}) + gohai = parse_gohai(meta.get("gohai")) + cloud_provider = detect_cloud_provider(host, gohai) + + last_reported = host.get("last_reported_time") + if isinstance(last_reported, (int, float)): + last_reported = datetime.fromtimestamp(last_reported, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + return { + "id": host.get("id"), + "name": host.get("name"), + "aliases": host.get("aliases", []), + "platform": meta.get("platform"), + "agent_version": meta.get("agent_version"), + "cloud_provider": cloud_provider, + "last_reported_time": last_reported, + "up": host.get("up"), + "is_muted": host.get("is_muted"), + "tags_by_source": host.get("tags_by_source", {}), + } + + +def get_agent_hosts() -> Dict[str, Any]: + base_url = get_base_url() + headers = get_dd_headers() + endpoint = f"{base_url}/api/v1/hosts" + + try: + hosts = fetch_all_hosts(base_url, headers) + + if not hosts: + return { + "status": "partial_or_empty", + "message": "No hosts found", + "api_endpoint": endpoint, + "record_count": 0, + "data": [], + "summary": { + "total_hosts": 0, + "active_hosts": 0, + "muted_hosts": 0, + "hosts_by_platform": {}, + "hosts_by_cloud_provider": {}, + "agent_version_distribution": {}, + }, + "retrieved_at": current_timestamp(), + } + + extracted = [extract_host_fields(h) for h in hosts] + + active_hosts = [h for h in extracted if h.get("up")] + muted_hosts = [h for h in extracted if h.get("is_muted")] + + platform_counts: Dict[str, int] = dict(Counter( + h.get("platform", "unknown") or "unknown" for h in extracted + )) + cloud_counts: Dict[str, int] = dict(Counter( + h.get("cloud_provider", "unknown") or "unknown" for h in extracted + if h.get("cloud_provider") + )) + version_counts: Dict[str, int] = dict(Counter( + h.get("agent_version", "unknown") or "unknown" for h in extracted + )) + + current_version = max(version_counts, key=lambda v: version_counts[v]) if version_counts else None + hosts_outdated = sum( + 1 for h in extracted + if (h.get("agent_version") or "unknown") != current_version + ) if current_version else 0 + + return { + "status": "success", + "api_endpoint": endpoint, + "record_count": len(extracted), + "data": extracted, + "summary": { + "total_hosts": len(extracted), + "active_hosts": len(active_hosts), + "muted_hosts": len(muted_hosts), + "hosts_by_platform": platform_counts, + "hosts_by_cloud_provider": cloud_counts, + "agent_version_distribution": version_counts, + "hosts_with_outdated_agent": hosts_outdated, + }, + "retrieved_at": current_timestamp(), + } + + except RuntimeError as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except Exception as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + result = get_agent_hosts() + + output_json = output_dir / "datadog_agent_hosts.json" + with open(output_json, "w") as f: + json.dump(result, f, indent=2, default=str) + + logger.info("Evidence saved to %s", output_json) + + return 0 if result.get("status") in {"success", "partial_or_empty"} else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/datadog/agent_hosts/fetcher.yaml b/fetchers/datadog/agent_hosts/fetcher.yaml new file mode 100644 index 0000000..b0fb952 --- /dev/null +++ b/fetchers/datadog/agent_hosts/fetcher.yaml @@ -0,0 +1,27 @@ +name: datadog_agent_hosts +version: 0.1.0 +description: Retrieves real-time inventory of all hosts reporting via the Datadog agent, including platform, cloud provider, agent version, and up/muted status. +category: datadog + +supports_targets: false + +runtime: + type: python + entry: fetcher.py + +output: + type: json + path: datadog_agent_hosts.json + +secrets: +- name: api_key + env: DATADOG_API_KEY +- name: app_key + env: DATADOG_APP_KEY +- name: base_url + env: DATADOG_BASE_URL + +evidence_set: + reference_id: EVD-DD-AGENT-HOSTS + name: Datadog Agent Hosts + instructions: 'Calls GET /api/v1/hosts with pagination (start/count). Extracts host name, platform, agent version, cloud provider, aliases, and up/muted state. Writes JSON inventory.' diff --git a/fetchers/datadog/apm_services/fetcher.py b/fetchers/datadog/apm_services/fetcher.py new file mode 100644 index 0000000..a352c0d --- /dev/null +++ b/fetchers/datadog/apm_services/fetcher.py @@ -0,0 +1,150 @@ +#!/usr/bin/env python3 +""" +DataDog APM Services Retrieval + +Purpose: Retrieve the APM service catalog to prove application services are registered and monitored. +""" + +import json +import logging +import os +import sys +from collections import Counter +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +import requests +from dotenv import load_dotenv + +logger = logging.getLogger("datadog_apm_services") + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def get_env(name: str, default: Optional[str] = None) -> str: + value = os.environ.get(name, default) + if value is None or value == "": + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def get_dd_headers() -> Dict[str, str]: + return { + "Content-Type": "application/json", + "DD-API-KEY": get_env("DATADOG_API_KEY"), + "DD-APPLICATION-KEY": get_env("DATADOG_APP_KEY"), + } + + +def get_base_url() -> str: + return os.environ.get("DATADOG_BASE_URL", "https://api.ddog-gov.com").rstrip("/") + + +def fetch_all_services(base_url: str, headers: Dict[str, str]) -> List[Dict[str, Any]]: + all_services: List[Dict[str, Any]] = [] + page_number = 0 + page_size = 100 + endpoint = f"{base_url}/api/v2/services/definitions" + + while True: + params = {"page[size]": page_size, "page[number]": page_number} + try: + response = requests.get(endpoint, headers=headers, params=params, timeout=30) + response.raise_for_status() + payload = response.json() + data = payload.get("data", []) + if not data: + break + all_services.extend(data) + if len(data) < page_size: + break + page_number += 1 + except requests.exceptions.RequestException as e: + logger.warning("Pagination interrupted at page %s: %s", page_number, e) + break + + return all_services + + +def extract_service_fields(service: Dict[str, Any]) -> Dict[str, Any]: + schema = service.get("attributes", {}).get("schema", {}) + contacts = schema.get("contacts", []) or [] + return { + "service_name": schema.get("dd-service"), + "team": schema.get("team"), + "tier": schema.get("tier"), + "languages": schema.get("languages", []), + "has_contact": len(contacts) > 0, + "contact_types": sorted({c.get("type") for c in contacts if c.get("type")}), + } + + +def get_apm_services() -> Dict[str, Any]: + base_url = get_base_url() + headers = get_dd_headers() + endpoint = f"{base_url}/api/v2/services/definitions" + + try: + services = fetch_all_services(base_url, headers) + extracted = [extract_service_fields(s) for s in services] + + language_counts: Dict[str, int] = Counter() + for svc in extracted: + for lang in (svc.get("languages") or []): + if lang: + language_counts[lang] += 1 + + team_counts: Dict[str, int] = dict(Counter( + svc.get("team", "unassigned") or "unassigned" for svc in extracted + )) + + services_with_owner = sum(1 for svc in extracted if svc.get("team")) + services_without_owner = len(extracted) - services_with_owner + + return { + "status": "success" if extracted else "partial_or_empty", + "api_endpoint": endpoint, + "record_count": len(extracted), + "data": extracted, + "summary": { + "total_services": len(extracted), + "services_by_language": dict(language_counts), + "services_by_team": team_counts, + "services_with_owner": services_with_owner, + "services_without_owner": services_without_owner, + }, + "retrieved_at": current_timestamp(), + } + + except RuntimeError as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except Exception as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + result = get_apm_services() + + output_json = output_dir / "datadog_apm_services.json" + with open(output_json, "w") as f: + json.dump(result, f, indent=2, default=str) + + logger.info("Evidence saved to %s", output_json) + + return 0 if result.get("status") in {"success", "partial_or_empty"} else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/datadog/apm_services/fetcher.yaml b/fetchers/datadog/apm_services/fetcher.yaml new file mode 100644 index 0000000..e809c68 --- /dev/null +++ b/fetchers/datadog/apm_services/fetcher.yaml @@ -0,0 +1,27 @@ +name: datadog_apm_services +version: 0.1.0 +description: Retrieves the Datadog APM service catalog to prove application services are registered and monitored with assigned owners. +category: datadog + +supports_targets: false + +runtime: + type: python + entry: fetcher.py + +output: + type: json + path: datadog_apm_services.json + +secrets: +- name: api_key + env: DATADOG_API_KEY +- name: app_key + env: DATADOG_APP_KEY +- name: base_url + env: DATADOG_BASE_URL + +evidence_set: + reference_id: EVD-DD-APM-SERVICES + name: Datadog APM Services + instructions: 'Calls GET /api/v2/services/definitions with pagination. Extracts service name, team, tier, languages, and contact presence. Writes JSON catalog.' diff --git a/fetchers/datadog/containers/fetcher.py b/fetchers/datadog/containers/fetcher.py new file mode 100644 index 0000000..f09363a --- /dev/null +++ b/fetchers/datadog/containers/fetcher.py @@ -0,0 +1,205 @@ +#!/usr/bin/env python3 +""" +DataDog Containers Retrieval + +Purpose: Retrieve real-time inventory of all running containers across the organization, +including image, state, host/node, and Kubernetes metadata. Pairs with datadog_agent_hosts +(nodes) to provide a complete two-layer inventory: nodes → containers. +""" + +import json +import logging +import os +import sys +from collections import Counter +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +import requests +from dotenv import load_dotenv + +logger = logging.getLogger("datadog_containers") + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def get_env(name: str, default: Optional[str] = None) -> str: + value = os.environ.get(name, default) + if value is None or value == "": + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def get_dd_headers() -> Dict[str, str]: + return { + "Content-Type": "application/json", + "DD-API-KEY": get_env("DATADOG_API_KEY"), + "DD-APPLICATION-KEY": get_env("DATADOG_APP_KEY"), + } + + +def get_base_url() -> str: + return os.environ.get("DATADOG_BASE_URL", "https://api.ddog-gov.com").rstrip("/") + + +def fetch_all_containers(base_url: str, headers: Dict[str, str]) -> List[Dict[str, Any]]: + all_containers: List[Dict[str, Any]] = [] + cursor: Optional[str] = None + endpoint = f"{base_url}/api/v2/containers" + + while True: + params: Dict[str, Any] = {"page[size]": 1000} + if cursor: + params["page[cursor]"] = cursor + try: + response = requests.get(endpoint, headers=headers, params=params, timeout=60) + response.raise_for_status() + payload = response.json() + data = payload.get("data", []) + all_containers.extend(data) + cursor = payload.get("meta", {}).get("pagination", {}).get("next_cursor") + if not cursor or not data: + break + except requests.exceptions.RequestException as e: + logger.warning("Pagination interrupted: %s", e) + break + + return all_containers + + +def parse_tag_value(tags: List[str], key: str) -> Optional[str]: + prefix = f"{key}:" + for tag in tags: + if tag.startswith(prefix): + return tag[len(prefix):] + return None + + +def safe_tags(tags: List[str]) -> List[str]: + # Strip any tag whose value portion contains '=' — guards against accidental env var leakage. + result = [] + for tag in tags: + value = tag.split(":", 1)[1] if ":" in tag else tag + if "=" not in value: + result.append(tag) + return result + + +def extract_container_fields(item: Dict[str, Any]) -> Dict[str, Any]: + attrs = item.get("attributes", {}) + tags = safe_tags(attrs.get("tags") or []) + return { + "container_id": attrs.get("container_id"), + "name": attrs.get("name"), + "host": attrs.get("host"), + "image_name": attrs.get("image_name"), + "image_tags": attrs.get("image_tags") or [], + "image_digest": attrs.get("image_digest") or None, + "state": attrs.get("state"), + "started_at": attrs.get("started_at"), + "tags": tags, + } + + +def get_containers() -> Dict[str, Any]: + base_url = get_base_url() + headers = get_dd_headers() + endpoint = f"{base_url}/api/v2/containers" + + try: + raw = fetch_all_containers(base_url, headers) + + if not raw: + return { + "status": "partial_or_empty", + "message": "No containers found", + "api_endpoint": endpoint, + "record_count": 0, + "data": [], + "summary": { + "total_containers": 0, + "containers_by_state": {}, + "containers_by_environment": {}, + "containers_by_namespace": {}, + "unique_images": 0, + "containers_without_image_digest": 0, + "non_running_containers_count": 0, + "prod_non_running_containers": 0, + }, + "retrieved_at": current_timestamp(), + } + + containers = [extract_container_fields(c) for c in raw] + + state_counts: Dict[str, int] = dict(Counter( + c.get("state") or "unknown" for c in containers + )) + + env_counts: Counter = Counter() + namespace_counts: Counter = Counter() + for c in containers: + tags = c.get("tags", []) + env = parse_tag_value(tags, "env") + env_counts[env if env else "untagged"] += 1 + ns = parse_tag_value(tags, "kube_namespace") + if ns: + namespace_counts[ns] += 1 + + unique_images = len(set(c.get("image_name") for c in containers if c.get("image_name"))) + containers_without_digest = sum(1 for c in containers if not c.get("image_digest")) + non_running = sum(1 for c in containers if c.get("state") != "running") + prod_non_running = sum( + 1 for c in containers + if c.get("state") != "running" and "env:prod" in c.get("tags", []) + ) + + return { + "status": "success", + "api_endpoint": endpoint, + "record_count": len(containers), + "data": containers, + "summary": { + "total_containers": len(containers), + "containers_by_state": state_counts, + "containers_by_environment": dict(env_counts), + "containers_by_namespace": dict(namespace_counts), + "unique_images": unique_images, + "containers_without_image_digest": containers_without_digest, + "non_running_containers_count": non_running, + "prod_non_running_containers": prod_non_running, + }, + "retrieved_at": current_timestamp(), + } + + except RuntimeError as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except Exception as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + result = get_containers() + + output_json = output_dir / "datadog_containers.json" + with open(output_json, "w") as f: + json.dump(result, f, indent=2, default=str) + + logger.info("Evidence saved to %s", output_json) + + return 0 if result.get("status") in {"success", "partial_or_empty"} else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/datadog/containers/fetcher.yaml b/fetchers/datadog/containers/fetcher.yaml new file mode 100644 index 0000000..aa9ad22 --- /dev/null +++ b/fetchers/datadog/containers/fetcher.yaml @@ -0,0 +1,27 @@ +name: datadog_containers +version: 0.1.0 +description: Retrieves real-time inventory of all running containers across the organization, including image, state, host, and Kubernetes metadata. +category: datadog + +supports_targets: false + +runtime: + type: python + entry: fetcher.py + +output: + type: json + path: datadog_containers.json + +secrets: +- name: api_key + env: DATADOG_API_KEY +- name: app_key + env: DATADOG_APP_KEY +- name: base_url + env: DATADOG_BASE_URL + +evidence_set: + reference_id: EVD-DD-CONTAINERS + name: Datadog Containers + instructions: 'Calls GET /api/v2/containers with cursor-based pagination. Extracts container name, image, state, host, tags. Writes JSON inventory.' diff --git a/fetchers/datadog/incident_timelines/fetcher.py b/fetchers/datadog/incident_timelines/fetcher.py new file mode 100644 index 0000000..31b6f49 --- /dev/null +++ b/fetchers/datadog/incident_timelines/fetcher.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python3 +""" +DataDog Incident Timelines Retrieval + +Purpose: Retrieve per-incident timeline entries to support after-action reports. +""" + +import json +import logging +import os +import sys +from collections import Counter +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +import requests +from dotenv import load_dotenv + +logger = logging.getLogger("datadog_incident_timelines") + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def get_env(name: str, default: Optional[str] = None) -> str: + value = os.environ.get(name, default) + if value is None or value == "": + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def get_dd_headers() -> Dict[str, str]: + return { + "Content-Type": "application/json", + "DD-API-KEY": get_env("DATADOG_API_KEY"), + "DD-APPLICATION-KEY": get_env("DATADOG_APP_KEY"), + } + + +def get_base_url() -> str: + return os.environ.get("DATADOG_BASE_URL", "https://api.ddog-gov.com").rstrip("/") + + +def fetch_incidents(base_url: str, headers: Dict[str, str], from_ts: str, to_ts: str) -> List[Dict[str, Any]]: + all_incidents: List[Dict[str, Any]] = [] + offset = 0 + page_size = 100 + endpoint = f"{base_url}/api/v2/incidents" + + while True: + params: Dict[str, Any] = { + "page[offset]": offset, + "page[size]": page_size, + "filter[created][start]": from_ts, + "filter[created][end]": to_ts, + } + try: + response = requests.get(endpoint, headers=headers, params=params, timeout=30) + response.raise_for_status() + payload = response.json() + data = payload.get("data", []) + all_incidents.extend(data) + total = payload.get("meta", {}).get("pagination", {}).get("total", 0) + offset += len(data) + if not data or offset >= total: + break + except requests.exceptions.RequestException as e: + logger.warning("Incident pagination interrupted: %s", e) + break + + return all_incidents + + +def fetch_timeline_cells( + base_url: str, + headers: Dict[str, str], + incident_id: str, +) -> List[Dict[str, Any]]: + endpoint = f"{base_url}/api/v2/incidents/{incident_id}/relationships/timeline_cells" + try: + response = requests.get(endpoint, headers=headers, timeout=30) + response.raise_for_status() + return response.json().get("data", []) + except requests.exceptions.RequestException as e: + logger.warning("Timeline fetch failed for incident %s: %s", incident_id, e) + return [] + + +def extract_cell_fields(cell: Dict[str, Any]) -> Dict[str, Any]: + attrs = cell.get("attributes", {}) + return { + "cell_type": attrs.get("cell_type"), + "created": attrs.get("created"), + "modified": attrs.get("modified"), + "created_by_id": cell.get("relationships", {}).get("created_by", {}).get("data", {}).get("id"), + } + + +def get_incident_timelines() -> Dict[str, Any]: + base_url = get_base_url() + headers = get_dd_headers() + + lookback_days = int(os.environ.get("DATADOG_INCIDENTS_LOOKBACK_DAYS", "90")) + now = datetime.now(timezone.utc) + from_ts = (now - timedelta(days=lookback_days)).strftime("%Y-%m-%dT%H:%M:%SZ") + to_ts = now.strftime("%Y-%m-%dT%H:%M:%SZ") + + try: + incidents = fetch_incidents(base_url, headers, from_ts, to_ts) + + if not incidents: + return { + "status": "partial_or_empty", + "message": "No incidents found in the lookback window", + "record_count": 0, + "data": [], + "summary": { + "incidents_with_timeline": 0, + "total_timeline_cells": 0, + "avg_cells_per_incident": 0, + "cell_type_distribution": {}, + }, + "retrieved_at": current_timestamp(), + } + + all_timelines = [] + cell_type_counter: Counter = Counter() + incidents_with_timeline = 0 + total_cells = 0 + + for incident in incidents: + incident_id = incident.get("id") + if not incident_id: + continue + + attrs = incident.get("attributes", {}) + cells_raw = fetch_timeline_cells(base_url, headers, incident_id) + cells = [extract_cell_fields(c) for c in cells_raw] + + for cell in cells: + cell_type_counter[cell.get("cell_type", "unknown")] += 1 + + total_cells += len(cells) + if cells: + incidents_with_timeline += 1 + + all_timelines.append({ + "incident_id": incident_id, + "incident_title": attrs.get("title"), + "cell_count": len(cells), + "timeline_cells": cells, + }) + + avg_cells = round(total_cells / incidents_with_timeline, 2) if incidents_with_timeline else 0 + + return { + "status": "success", + "api_endpoint": f"{base_url}/api/v2/incidents/{{id}}/relationships/timeline_cells", + "record_count": total_cells, + "data": all_timelines, + "summary": { + "incidents_with_timeline": incidents_with_timeline, + "total_timeline_cells": total_cells, + "avg_cells_per_incident": avg_cells, + "cell_type_distribution": dict(cell_type_counter), + }, + "retrieved_at": current_timestamp(), + } + + except RuntimeError as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except Exception as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + result = get_incident_timelines() + + output_json = output_dir / "datadog_incident_timelines.json" + with open(output_json, "w") as f: + json.dump(result, f, indent=2, default=str) + + logger.info("Evidence saved to %s", output_json) + + return 0 if result.get("status") in {"success", "partial_or_empty"} else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/datadog/incident_timelines/fetcher.yaml b/fetchers/datadog/incident_timelines/fetcher.yaml new file mode 100644 index 0000000..9631f36 --- /dev/null +++ b/fetchers/datadog/incident_timelines/fetcher.yaml @@ -0,0 +1,28 @@ +name: datadog_incident_timelines +version: 0.1.0 +description: Retrieves per-incident timeline entries from Datadog to support after-action reviews and incident response documentation. +category: datadog + +supports_targets: false + +runtime: + type: python + entry: fetcher.py + +output: + type: json + path: datadog_incident_timelines.json + +secrets: +- name: api_key + env: DATADOG_API_KEY +- name: app_key + env: DATADOG_APP_KEY +- name: base_url + env: DATADOG_BASE_URL +- name: incidents_lookback_days + env: DATADOG_INCIDENTS_LOOKBACK_DAYS +evidence_set: + reference_id: EVD-DD-INCIDENT-TIMELINES + name: Datadog Incident Timelines + instructions: 'Calls GET /api/v2/incidents then GET /api/v2/incidents/{id}/timeline for each. Lookback window controlled by DATADOG_INCIDENTS_LOOKBACK_DAYS (default 90). Writes JSON timeline data.' diff --git a/fetchers/datadog/incidents_list/fetcher.py b/fetchers/datadog/incidents_list/fetcher.py new file mode 100644 index 0000000..d8bb3b8 --- /dev/null +++ b/fetchers/datadog/incidents_list/fetcher.py @@ -0,0 +1,211 @@ +#!/usr/bin/env python3 +""" +DataDog Incidents Retrieval + +Purpose: Retrieve incident records for after-action review and pattern analysis. +""" + +import json +import logging +import os +import sys +from collections import Counter +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +import requests + +from dotenv import load_dotenv + +logger = logging.getLogger("datadog_incidents_list") + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def get_env(name: str, default: Optional[str] = None) -> str: + value = os.environ.get(name, default) + if value is None or value == "": + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def get_dd_headers() -> Dict[str, str]: + return { + "Content-Type": "application/json", + "DD-API-KEY": get_env("DATADOG_API_KEY"), + "DD-APPLICATION-KEY": get_env("DATADOG_APP_KEY"), + } + + +def get_base_url() -> str: + return os.environ.get("DATADOG_BASE_URL", "https://api.ddog-gov.com").rstrip("/") + + +def fetch_all_incidents( + base_url: str, + headers: Dict[str, str], + from_ts: str, + to_ts: str, +) -> List[Dict[str, Any]]: + all_incidents: List[Dict[str, Any]] = [] + offset = 0 + page_size = 100 + endpoint = f"{base_url}/api/v2/incidents" + + while True: + params: Dict[str, Any] = { + "page[offset]": offset, + "page[size]": page_size, + "filter[created][start]": from_ts, + "filter[created][end]": to_ts, + } + try: + response = requests.get(endpoint, headers=headers, params=params, timeout=30) + response.raise_for_status() + payload = response.json() + data = payload.get("data", []) + all_incidents.extend(data) + + total = payload.get("meta", {}).get("pagination", {}).get("total", 0) + offset += len(data) + if not data or offset >= total: + break + except requests.exceptions.RequestException as e: + logger.warning("Pagination interrupted at offset %s: %s", offset, e) + break + + return all_incidents + + +def parse_resolution_hours(detected: Optional[str], resolved: Optional[str]) -> Optional[float]: + if not detected or not resolved: + return None + try: + fmt = "%Y-%m-%dT%H:%M:%SZ" + d = datetime.strptime(detected, fmt) + r = datetime.strptime(resolved, fmt) + delta = r - d + return round(delta.total_seconds() / 3600, 2) + except Exception: + return None + + +def extract_incident_fields(incident: Dict[str, Any]) -> Dict[str, Any]: + attrs = incident.get("attributes", {}) + detected = attrs.get("detected") + resolved = attrs.get("resolved") + return { + "id": incident.get("id"), + "title": attrs.get("title"), + "severity": attrs.get("severity"), + "status": attrs.get("status"), + "detected": detected, + "resolved": resolved, + "resolution_hours": parse_resolution_hours(detected, resolved), + "customer_impact_scope": attrs.get("customer_impact_scope"), + "customer_impact_duration": attrs.get("customer_impact_duration"), + "commander_id": attrs.get("commander", {}).get("data", {}).get("id"), + "postmortem_id": attrs.get("postmortem_id"), + } + + +def get_incidents_list() -> Dict[str, Any]: + base_url = get_base_url() + headers = get_dd_headers() + endpoint = f"{base_url}/api/v2/incidents" + + lookback_days = int(os.environ.get("DATADOG_INCIDENTS_LOOKBACK_DAYS", "90")) + now = datetime.now(timezone.utc) + from_dt = now - timedelta(days=lookback_days) + from_ts = from_dt.strftime("%Y-%m-%dT%H:%M:%SZ") + to_ts = now.strftime("%Y-%m-%dT%H:%M:%SZ") + + try: + incidents = fetch_all_incidents(base_url, headers, from_ts, to_ts) + extracted = [extract_incident_fields(i) for i in incidents] + + if not extracted: + return { + "status": "partial_or_empty", + "message": "No incidents found in the lookback window", + "api_endpoint": endpoint, + "record_count": 0, + "data": [], + "summary": { + "total_incidents": 0, + "incidents_by_severity": {}, + "incidents_by_status": {}, + "incidents_with_postmortem": 0, + "avg_resolution_hours": None, + "lookback_days": lookback_days, + }, + "retrieved_at": current_timestamp(), + } + + severity_counts: Dict[str, int] = dict(Counter( + i.get("severity", "UNKNOWN") or "UNKNOWN" for i in extracted + )) + status_counts: Dict[str, int] = dict(Counter( + i.get("status", "unknown") or "unknown" for i in extracted + )) + incidents_with_postmortem = sum(1 for i in extracted if i.get("postmortem_id")) + + resolved_without_postmortem_ids = [ + i["id"] for i in extracted + if i.get("status") == "resolved" and not i.get("postmortem_id") + ] + + resolution_times = [ + i["resolution_hours"] for i in extracted if i.get("resolution_hours") is not None + ] + avg_resolution = round(sum(resolution_times) / len(resolution_times), 2) if resolution_times else None + + return { + "status": "success", + "api_endpoint": endpoint, + "record_count": len(extracted), + "data": extracted, + "summary": { + "total_incidents": len(extracted), + "incidents_by_severity": severity_counts, + "incidents_by_status": status_counts, + "incidents_with_postmortem": incidents_with_postmortem, + "resolved_without_postmortem_ids": resolved_without_postmortem_ids, + "avg_resolution_hours": avg_resolution, + "lookback_days": lookback_days, + }, + "retrieved_at": current_timestamp(), + } + + except RuntimeError as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except Exception as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + result = get_incidents_list() + + output_json = output_dir / "datadog_incidents_list.json" + with open(output_json, "w") as f: + json.dump(result, f, indent=2, default=str) + + logger.info("Evidence saved to %s", output_json) + + return 0 if result.get("status") in {"success", "partial_or_empty"} else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/datadog/incidents_list/fetcher.yaml b/fetchers/datadog/incidents_list/fetcher.yaml new file mode 100644 index 0000000..1c47cdc --- /dev/null +++ b/fetchers/datadog/incidents_list/fetcher.yaml @@ -0,0 +1,28 @@ +name: datadog_incidents_list +version: 0.1.0 +description: Retrieves incident records from Datadog for after-action review and pattern analysis within a configurable lookback window. +category: datadog + +supports_targets: false + +runtime: + type: python + entry: fetcher.py + +output: + type: json + path: datadog_incidents_list.json + +secrets: +- name: api_key + env: DATADOG_API_KEY +- name: app_key + env: DATADOG_APP_KEY +- name: base_url + env: DATADOG_BASE_URL +- name: incidents_lookback_days + env: DATADOG_INCIDENTS_LOOKBACK_DAYS +evidence_set: + reference_id: EVD-DD-INCIDENTS-LIST + name: Datadog Incidents List + instructions: 'Calls GET /api/v2/incidents with offset pagination and date filters. Lookback window controlled by DATADOG_INCIDENTS_LOOKBACK_DAYS (default 90). Writes JSON incident list.' diff --git a/fetchers/datadog/infra_agent_checks/fetcher.py b/fetchers/datadog/infra_agent_checks/fetcher.py new file mode 100644 index 0000000..fe0ac0d --- /dev/null +++ b/fetchers/datadog/infra_agent_checks/fetcher.py @@ -0,0 +1,232 @@ +#!/usr/bin/env python3 +""" +DataDog Infrastructure Agent Checks Retrieval + +Purpose: Retrieve per-host agent check results to prove configuration evaluation is actively +running and passing across the fleet. Check data is sourced from meta.agent_checks on each +host record returned by GET /api/v1/hosts — no separate check endpoint is required. + +Each agent_checks entry is a list: [check_name, module, instance_id, STATUS, message, _, tags] +Status values: "OK", "WARNING", "ERROR", "UNKNOWN" +""" + +import json +import logging +import os +import sys +from collections import Counter, defaultdict +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +import requests +from dotenv import load_dotenv + +logger = logging.getLogger("datadog_infra_agent_checks") + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def get_env(name: str, default: Optional[str] = None) -> str: + value = os.environ.get(name, default) + if value is None or value == "": + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def get_dd_headers() -> Dict[str, str]: + return { + "Content-Type": "application/json", + "DD-API-KEY": get_env("DATADOG_API_KEY"), + "DD-APPLICATION-KEY": get_env("DATADOG_APP_KEY"), + } + + +def get_base_url() -> str: + return os.environ.get("DATADOG_BASE_URL", "https://api.ddog-gov.com").rstrip("/") + + +# Map DataDog agent check status strings to lowercase compliance-friendly names. +# "ERROR" maps to "critical" since a failed check is a critical configuration finding. +STATUS_NORMALIZE = {"OK": "ok", "WARNING": "warning", "ERROR": "critical", "UNKNOWN": "unknown"} + + +def get_total_hosts(base_url: str, headers: Dict[str, str]) -> int: + try: + response = requests.get( + f"{base_url}/api/v1/hosts", + headers=headers, + params={"count": 1}, + timeout=30, + ) + response.raise_for_status() + return response.json().get("total_matching", 0) + except requests.exceptions.HTTPError: + raise + except Exception: + return 0 + + +def fetch_hosts_with_checks(base_url: str, headers: Dict[str, str]) -> List[Dict[str, Any]]: + all_hosts: List[Dict[str, Any]] = [] + count = 1000 + start = 0 + endpoint = f"{base_url}/api/v1/hosts" + + total_matching = get_total_hosts(base_url, headers) + if total_matching == 0: + return [] + + while start < total_matching: + params = {"start": start, "count": count} + try: + response = requests.get(endpoint, headers=headers, params=params, timeout=60) + response.raise_for_status() + payload = response.json() + hosts = payload.get("host_list", []) + if not hosts: + break + all_hosts.extend(hosts) + start += count + except requests.exceptions.RequestException as e: + logger.warning("Pagination interrupted at start=%s: %s", start, e) + break + + return all_hosts + + +def parse_agent_checks(host: Dict[str, Any]) -> List[Tuple[str, str, str]]: + """Extract (check_name, status, message) tuples from meta.agent_checks. + + Each entry in agent_checks is a list: + [check_name, module, instance_id, STATUS, message, extra, tags] + """ + checks = host.get("meta", {}).get("agent_checks", []) + results = [] + for entry in checks: + if not isinstance(entry, list) or len(entry) < 4: + continue + check_name = entry[0] or "unknown" + raw_status = (entry[3] or "UNKNOWN").upper() + message = entry[4] if len(entry) > 4 else "" + status = STATUS_NORMALIZE.get(raw_status, "unknown") + results.append((check_name, status, str(message) if message else "")) + return results + + +def get_infra_agent_checks() -> Dict[str, Any]: + base_url = get_base_url() + headers = get_dd_headers() + endpoint = f"{base_url}/api/v1/hosts" + + try: + hosts = fetch_hosts_with_checks(base_url, headers) + + if not hosts: + return { + "status": "partial_or_empty", + "message": "No hosts found", + "api_endpoint": endpoint, + "record_count": 0, + "data": [], + "summary": { + "total_check_runs": 0, + "checks_by_status": {}, + "unique_check_names": 0, + "hosts_with_critical_checks": 0, + "hosts_with_warning_checks": 0, + "checks_by_name": {}, + }, + "retrieved_at": current_timestamp(), + } + + all_check_records = [] + checks_by_name: Dict[str, Dict[str, int]] = defaultdict(lambda: defaultdict(int)) + critical_hosts: set = set() + warning_hosts: set = set() + status_counter: Counter = Counter() + + for host in hosts: + host_name = host.get("name", "unknown") + check_tuples = parse_agent_checks(host) + for check_name, status, message in check_tuples: + all_check_records.append({ + "check_name": check_name, + "host_name": host_name, + "status": status, + "message": message if message else None, + }) + checks_by_name[check_name][status] += 1 + status_counter[status] += 1 + if status == "critical": + critical_hosts.add(host_name) + elif status == "warning": + warning_hosts.add(host_name) + + checks_by_name_final = {k: dict(v) for k, v in checks_by_name.items()} + + if not all_check_records: + return { + "status": "partial_or_empty", + "message": "Hosts found but no agent_checks data returned", + "api_endpoint": endpoint, + "record_count": 0, + "data": [], + "summary": { + "total_check_runs": 0, + "checks_by_status": {}, + "unique_check_names": 0, + "hosts_with_critical_checks": 0, + "hosts_with_warning_checks": 0, + "checks_by_name": {}, + }, + "retrieved_at": current_timestamp(), + } + + return { + "status": "success", + "api_endpoint": endpoint, + "record_count": len(all_check_records), + "data": all_check_records, + "summary": { + "total_check_runs": len(all_check_records), + "checks_by_status": dict(status_counter), + "unique_check_names": len(checks_by_name_final), + "hosts_with_critical_checks": len(critical_hosts), + "hosts_with_warning_checks": len(warning_hosts), + "checks_by_name": checks_by_name_final, + }, + "retrieved_at": current_timestamp(), + } + + except RuntimeError as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except Exception as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + result = get_infra_agent_checks() + + output_json = output_dir / "datadog_infra_agent_checks.json" + with open(output_json, "w") as f: + json.dump(result, f, indent=2, default=str) + + logger.info("Evidence saved to %s", output_json) + + return 0 if result.get("status") in {"success", "partial_or_empty"} else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/datadog/infra_agent_checks/fetcher.yaml b/fetchers/datadog/infra_agent_checks/fetcher.yaml new file mode 100644 index 0000000..5fbc698 --- /dev/null +++ b/fetchers/datadog/infra_agent_checks/fetcher.yaml @@ -0,0 +1,27 @@ +name: datadog_infra_agent_checks +version: 0.1.0 +description: Retrieves per-host Datadog agent check results to prove configuration evaluation is actively running and passing across the fleet. +category: datadog + +supports_targets: false + +runtime: + type: python + entry: fetcher.py + +output: + type: json + path: datadog_infra_agent_checks.json + +secrets: +- name: api_key + env: DATADOG_API_KEY +- name: app_key + env: DATADOG_APP_KEY +- name: base_url + env: DATADOG_BASE_URL + +evidence_set: + reference_id: EVD-DD-INFRA-CHECKS + name: Datadog Infrastructure Agent Checks + instructions: 'Calls GET /api/v1/hosts with pagination; extracts meta.agent_checks from each host. Reports check status (ok/warning/critical/unknown) per host. Writes JSON check results.' diff --git a/fetchers/datadog/log_archives/fetcher.py b/fetchers/datadog/log_archives/fetcher.py new file mode 100644 index 0000000..63460eb --- /dev/null +++ b/fetchers/datadog/log_archives/fetcher.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +""" +DataDog Log Archives Retrieval + +Purpose: Retrieve long-term log archive configurations to prove logs are durably stored. +""" + +import json +import logging +import os +import sys +from collections import Counter +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +import requests +from dotenv import load_dotenv + +logger = logging.getLogger("datadog_log_archives") + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def get_env(name: str, default: Optional[str] = None) -> str: + value = os.environ.get(name, default) + if value is None or value == "": + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def get_dd_headers() -> Dict[str, str]: + return { + "Content-Type": "application/json", + "DD-API-KEY": get_env("DATADOG_API_KEY"), + "DD-APPLICATION-KEY": get_env("DATADOG_APP_KEY"), + } + + +def get_base_url() -> str: + return os.environ.get("DATADOG_BASE_URL", "https://api.ddog-gov.com").rstrip("/") + + +def extract_archive_fields(archive: Dict[str, Any]) -> Dict[str, Any]: + attrs = archive.get("attributes", archive) + destination = attrs.get("destination", {}) + dest_type = destination.get("type", "unknown") + + # Extract storage location without leaking full paths + storage_location = ( + destination.get("bucket") + or destination.get("container") + or destination.get("storageAccount") + or "" + ) + + return { + "id": archive.get("id"), + "name": attrs.get("name"), + "state": attrs.get("state"), + "destination_type": dest_type, + "storage_location": storage_location, + "destination_path": destination.get("path"), + "includeTags": attrs.get("includeTags"), + "query": attrs.get("query"), + } + + +def get_log_archives() -> Dict[str, Any]: + base_url = get_base_url() + headers = get_dd_headers() + endpoint = f"{base_url}/api/v2/logs/config/archives" + + try: + response = requests.get(endpoint, headers=headers, timeout=30) + response.raise_for_status() + payload = response.json() + + archives_raw = payload.get("data", []) + extracted = [extract_archive_fields(a) for a in archives_raw] + + enabled = [a for a in extracted if str(a.get("state", "")).upper() == "ENABLED"] + dest_type_counts: Dict[str, int] = dict(Counter( + a.get("destination_type", "unknown") for a in extracted + )) + + archives_summary = [ + { + "name": a["name"], + "state": a["state"], + "destination_type": a["destination_type"], + "bucket": a["storage_location"], + "destination_path": a.get("destination_path"), + } + for a in extracted + ] + + return { + "status": "success" if extracted else "partial_or_empty", + "api_endpoint": endpoint, + "record_count": len(extracted), + "data": extracted, + "summary": { + "total_archives": len(extracted), + "enabled_archives": len(enabled), + "archives_by_destination_type": dest_type_counts, + "archives_summary": archives_summary, + }, + "retrieved_at": current_timestamp(), + } + + except RuntimeError as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except Exception as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + result = get_log_archives() + + output_json = output_dir / "datadog_log_archives.json" + with open(output_json, "w") as f: + json.dump(result, f, indent=2, default=str) + + logger.info("Evidence saved to %s", output_json) + + return 0 if result.get("status") in {"success", "partial_or_empty"} else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/datadog/log_archives/fetcher.yaml b/fetchers/datadog/log_archives/fetcher.yaml new file mode 100644 index 0000000..3937dda --- /dev/null +++ b/fetchers/datadog/log_archives/fetcher.yaml @@ -0,0 +1,27 @@ +name: datadog_log_archives +version: 0.1.0 +description: Retrieves Datadog log archive configurations to prove logs are durably stored in long-term storage. +category: datadog + +supports_targets: false + +runtime: + type: python + entry: fetcher.py + +output: + type: json + path: datadog_log_archives.json + +secrets: +- name: api_key + env: DATADOG_API_KEY +- name: app_key + env: DATADOG_APP_KEY +- name: base_url + env: DATADOG_BASE_URL + +evidence_set: + reference_id: EVD-DD-LOG-ARCHIVES + name: Datadog Log Archives + instructions: 'Calls GET /api/v2/logs/config/archives. Extracts archive name, destination type, state, and filter query. Writes JSON archive config list.' diff --git a/fetchers/datadog/log_indexes/fetcher.py b/fetchers/datadog/log_indexes/fetcher.py new file mode 100644 index 0000000..0903515 --- /dev/null +++ b/fetchers/datadog/log_indexes/fetcher.py @@ -0,0 +1,156 @@ +#!/usr/bin/env python3 +""" +DataDog Log Indexes Retrieval + +Purpose: Retrieve log index retention configs to prove what data is retained and for how long. +""" + +import json +import logging +import os +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +import requests + +from dotenv import load_dotenv + +logger = logging.getLogger("datadog_log_indexes") + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def get_env(name: str, default: Optional[str] = None) -> str: + value = os.environ.get(name, default) + if value is None or value == "": + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def get_dd_headers() -> Dict[str, str]: + return { + "Content-Type": "application/json", + "DD-API-KEY": get_env("DATADOG_API_KEY"), + "DD-APPLICATION-KEY": get_env("DATADOG_APP_KEY"), + } + + +def get_base_url() -> str: + return os.environ.get("DATADOG_BASE_URL", "https://api.ddog-gov.com").rstrip("/") + + +def extract_index_fields(index: Dict[str, Any]) -> Dict[str, Any]: + # Real API returns snake_case; mocks use camelCase — handle both + retention = index.get("num_retention_days") or index.get("numRetentionDays") + flex_retention = index.get("num_flex_logs_retention_days") or index.get("numFlexLogsRetentionDays") + daily_limit = index.get("daily_limit") or index.get("dailyLimit") + is_rate_limited = index.get("is_rate_limited") if "is_rate_limited" in index else index.get("isRateLimited") + daily_limit_gb = round(daily_limit / 1_000_000_000, 2) if daily_limit else None + return { + "name": index.get("name"), + "numRetentionDays": retention, + "numFlexLogsRetentionDays": flex_retention, + "filter_query": index.get("filter", {}).get("query"), + "dailyLimit": daily_limit, + "isRateLimited": is_rate_limited, + } + + +def get_log_indexes() -> Dict[str, Any]: + base_url = get_base_url() + headers = get_dd_headers() + endpoint = f"{base_url}/api/v1/logs/config/indexes" + + try: + response = requests.get(endpoint, headers=headers, timeout=30) + response.raise_for_status() + payload = response.json() + + indexes = payload.get("indexes", payload) if isinstance(payload, dict) else payload + + extracted = [extract_index_fields(idx) for idx in indexes] + + retention_values = [ + idx["numRetentionDays"] + for idx in extracted + if idx.get("numRetentionDays") is not None + ] + + min_retention = min(retention_values) if retention_values else 0 + max_retention = max(retention_values) if retention_values else 0 + indexes_below_90 = sum(1 for v in retention_values if v < 90) + + audit_keywords = {"cloudtrail", "okta", "kubernetes.audit"} + + def is_audit_index(idx: Dict[str, Any]) -> bool: + query = (idx.get("filter_query") or "").lower() + return any(kw in query for kw in audit_keywords) + + audit_retentions = [ + idx["numRetentionDays"] + for idx in extracted + if is_audit_index(idx) and idx.get("numRetentionDays") is not None + ] + min_retention_audit = min(audit_retentions) if audit_retentions else None + audit_below_90 = sum(1 for v in audit_retentions if v < 90) + + indexes_summary = [ + { + "name": idx["name"], + "retention_days": idx["numRetentionDays"], + "filter": idx["filter_query"], + "daily_limit_gb": round(idx["dailyLimit"] / 1_000_000_000, 2) if idx.get("dailyLimit") else None, + } + for idx in extracted + ] + + return { + "status": "success" if extracted else "partial_or_empty", + "api_endpoint": endpoint, + "record_count": len(extracted), + "data": extracted, + "summary": { + "total_indexes": len(extracted), + "min_retention_days": min_retention, + "max_retention_days": max_retention, + "indexes_below_90_days": indexes_below_90, + "min_retention_days_audit_indexes": min_retention_audit, + "audit_indexes_below_90_days": audit_below_90, + "indexes_summary": indexes_summary, + }, + "retrieved_at": current_timestamp(), + } + + except RuntimeError as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except Exception as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + result = get_log_indexes() + + output_json = output_dir / "datadog_log_indexes.json" + with open(output_json, "w") as f: + json.dump(result, f, indent=2, default=str) + + logger.info("Evidence saved to %s", output_json) + + return 0 if result.get("status") in {"success", "partial_or_empty"} else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/datadog/log_indexes/fetcher.yaml b/fetchers/datadog/log_indexes/fetcher.yaml new file mode 100644 index 0000000..e22817b --- /dev/null +++ b/fetchers/datadog/log_indexes/fetcher.yaml @@ -0,0 +1,27 @@ +name: datadog_log_indexes +version: 0.1.0 +description: Retrieves Datadog log index retention configurations to prove what log data is retained and for how long. +category: datadog + +supports_targets: false + +runtime: + type: python + entry: fetcher.py + +output: + type: json + path: datadog_log_indexes.json + +secrets: +- name: api_key + env: DATADOG_API_KEY +- name: app_key + env: DATADOG_APP_KEY +- name: base_url + env: DATADOG_BASE_URL + +evidence_set: + reference_id: EVD-DD-LOG-INDEXES + name: Datadog Log Indexes + instructions: 'Calls GET /api/v1/logs/config/indexes. Extracts index name, retention days, and filter query. Writes JSON index config list.' diff --git a/fetchers/datadog/log_pipelines/fetcher.py b/fetchers/datadog/log_pipelines/fetcher.py new file mode 100644 index 0000000..363455d --- /dev/null +++ b/fetchers/datadog/log_pipelines/fetcher.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 +""" +DataDog Log Pipelines Retrieval + +Purpose: Retrieve log processing pipeline configurations to prove defined event types are ingested and processed. +""" + +import json +import logging +import os +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +import requests +from dotenv import load_dotenv + +logger = logging.getLogger("datadog_log_pipelines") + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def get_env(name: str, default: Optional[str] = None) -> str: + value = os.environ.get(name, default) + if value is None or value == "": + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def get_dd_headers() -> Dict[str, str]: + return { + "Content-Type": "application/json", + "DD-API-KEY": get_env("DATADOG_API_KEY"), + "DD-APPLICATION-KEY": get_env("DATADOG_APP_KEY"), + } + + +def get_base_url() -> str: + return os.environ.get("DATADOG_BASE_URL", "https://api.ddog-gov.com").rstrip("/") + + +def extract_pipeline_fields(pipeline: Dict[str, Any]) -> Dict[str, Any]: + processors = pipeline.get("processors", []) + return { + "id": pipeline.get("id"), + "name": pipeline.get("name"), + "isEnabled": pipeline.get("isEnabled"), + "isReadOnly": pipeline.get("isReadOnly"), + "filter_query": pipeline.get("filter", {}).get("query"), + "processor_count": len(processors), + } + + +def get_log_pipelines() -> Dict[str, Any]: + base_url = get_base_url() + headers = get_dd_headers() + endpoint = f"{base_url}/api/v1/logs/config/pipelines" + + try: + response = requests.get(endpoint, headers=headers, timeout=30) + response.raise_for_status() + pipelines = response.json() + + if not isinstance(pipelines, list): + pipelines = pipelines.get("data", pipelines.get("pipelines", [])) + + extracted = [extract_pipeline_fields(p) for p in pipelines] + + enabled = [p for p in extracted if p.get("isEnabled")] + disabled = [p for p in extracted if not p.get("isEnabled")] + read_only = [p for p in extracted if p.get("isReadOnly")] + disabled_zero_processors = sum( + 1 for p in disabled if p.get("processor_count", 0) == 0 + ) + + pipelines_summary = [ + { + "name": p["name"], + "isEnabled": p["isEnabled"], + "processor_count": p["processor_count"], + "filter": p["filter_query"], + } + for p in extracted + ] + + return { + "status": "success" if extracted else "partial_or_empty", + "api_endpoint": endpoint, + "record_count": len(extracted), + "data": extracted, + "summary": { + "total_pipelines": len(extracted), + "enabled_pipelines": len(enabled), + "disabled_pipelines": len(disabled), + "read_only_pipelines": len(read_only), + "disabled_pipelines_with_zero_processors": disabled_zero_processors, + "pipelines_summary": pipelines_summary, + }, + "retrieved_at": current_timestamp(), + } + + except RuntimeError as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except Exception as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + result = get_log_pipelines() + + output_json = output_dir / "datadog_log_pipelines.json" + with open(output_json, "w") as f: + json.dump(result, f, indent=2, default=str) + + logger.info("Evidence saved to %s", output_json) + + return 0 if result.get("status") in {"success", "partial_or_empty"} else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/datadog/log_pipelines/fetcher.yaml b/fetchers/datadog/log_pipelines/fetcher.yaml new file mode 100644 index 0000000..a3686ab --- /dev/null +++ b/fetchers/datadog/log_pipelines/fetcher.yaml @@ -0,0 +1,27 @@ +name: datadog_log_pipelines +version: 0.1.0 +description: Retrieves Datadog log processing pipeline configurations to prove defined event types are ingested and processed. +category: datadog + +supports_targets: false + +runtime: + type: python + entry: fetcher.py + +output: + type: json + path: datadog_log_pipelines.json + +secrets: +- name: api_key + env: DATADOG_API_KEY +- name: app_key + env: DATADOG_APP_KEY +- name: base_url + env: DATADOG_BASE_URL + +evidence_set: + reference_id: EVD-DD-LOG-PIPELINES + name: Datadog Log Pipelines + instructions: 'Calls GET /api/v1/logs/config/pipelines. Extracts pipeline name, type, enabled state, filter query, and processor count. Writes JSON pipeline config list.' diff --git a/fetchers/datadog/monitors_list/fetcher.py b/fetchers/datadog/monitors_list/fetcher.py new file mode 100644 index 0000000..f996bad --- /dev/null +++ b/fetchers/datadog/monitors_list/fetcher.py @@ -0,0 +1,147 @@ +#!/usr/bin/env python3 +""" +DataDog Monitors Retrieval + +Purpose: Retrieve monitor configurations to prove persistent alerting is defined for infrastructure and security events. +""" + +import json +import logging +import os +import sys +from collections import Counter +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +import requests +from dotenv import load_dotenv + +logger = logging.getLogger("datadog_monitors_list") + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def get_env(name: str, default: Optional[str] = None) -> str: + value = os.environ.get(name, default) + if value is None or value == "": + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def get_dd_headers() -> Dict[str, str]: + return { + "Content-Type": "application/json", + "DD-API-KEY": get_env("DATADOG_API_KEY"), + "DD-APPLICATION-KEY": get_env("DATADOG_APP_KEY"), + } + + +def get_base_url() -> str: + return os.environ.get("DATADOG_BASE_URL", "https://api.ddog-gov.com").rstrip("/") + + +def fetch_all_monitors(base_url: str, headers: Dict[str, str]) -> List[Dict[str, Any]]: + all_monitors: List[Dict[str, Any]] = [] + page = 0 + page_size = 100 + endpoint = f"{base_url}/api/v1/monitor" + + while True: + params = {"page": page, "page_size": page_size} + try: + response = requests.get(endpoint, headers=headers, params=params, timeout=30) + response.raise_for_status() + data = response.json() + if not isinstance(data, list) or not data: + break + all_monitors.extend(data) + if len(data) < page_size: + break + page += 1 + except requests.exceptions.RequestException as e: + logger.warning("Pagination interrupted at page %s: %s", page, e) + break + + return all_monitors + + +def extract_monitor_fields(monitor: Dict[str, Any]) -> Dict[str, Any]: + message = monitor.get("message", "") or "" + return { + "id": monitor.get("id"), + "name": monitor.get("name"), + "type": monitor.get("type"), + "overall_state": monitor.get("overall_state"), + "query": (monitor.get("query") or "")[:200], + "message": message[:200], + "tags": monitor.get("tags", []), + "thresholds": monitor.get("thresholds"), + "notify_no_data": monitor.get("notify_no_data"), + "no_data_timeframe": monitor.get("no_data_timeframe"), + } + + +def get_monitors_list() -> Dict[str, Any]: + base_url = get_base_url() + headers = get_dd_headers() + endpoint = f"{base_url}/api/v1/monitor" + + try: + monitors = fetch_all_monitors(base_url, headers) + extracted = [extract_monitor_fields(m) for m in monitors] + + status_counts: Dict[str, int] = dict(Counter( + m.get("overall_state", "Unknown") or "Unknown" for m in extracted + )) + type_counts: Dict[str, int] = dict(Counter( + m.get("type", "unknown") or "unknown" for m in extracted + )) + + monitors_in_alert = sum(1 for m in extracted if m.get("overall_state") == "Alert") + + return { + "status": "success" if extracted else "partial_or_empty", + "api_endpoint": endpoint, + "record_count": len(extracted), + "data": extracted, + "summary": { + "total_monitors": len(extracted), + "monitors_by_status": status_counts, + "monitors_by_type": type_counts, + "monitors_in_alert_count": monitors_in_alert, + }, + "retrieved_at": current_timestamp(), + } + + except RuntimeError as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except Exception as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + result = get_monitors_list() + + output_json = output_dir / "datadog_monitors_list.json" + with open(output_json, "w") as f: + json.dump(result, f, indent=2, default=str) + + logger.info("Evidence saved to %s", output_json) + + return 0 if result.get("status") in {"success", "partial_or_empty"} else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/datadog/monitors_list/fetcher.yaml b/fetchers/datadog/monitors_list/fetcher.yaml new file mode 100644 index 0000000..90a864f --- /dev/null +++ b/fetchers/datadog/monitors_list/fetcher.yaml @@ -0,0 +1,27 @@ +name: datadog_monitors_list +version: 0.1.0 +description: Retrieves Datadog monitor configurations to prove persistent alerting is defined for infrastructure and security events. +category: datadog + +supports_targets: false + +runtime: + type: python + entry: fetcher.py + +output: + type: json + path: datadog_monitors_list.json + +secrets: +- name: api_key + env: DATADOG_API_KEY +- name: app_key + env: DATADOG_APP_KEY +- name: base_url + env: DATADOG_BASE_URL + +evidence_set: + reference_id: EVD-DD-MONITORS + name: Datadog Monitors List + instructions: 'Calls GET /api/v1/monitor with pagination. Extracts monitor name, type, overall_state, query, thresholds, and notify settings. Writes JSON monitor list.' diff --git a/fetchers/datadog/siem_configuration/fetcher.py b/fetchers/datadog/siem_configuration/fetcher.py new file mode 100644 index 0000000..c2d8abf --- /dev/null +++ b/fetchers/datadog/siem_configuration/fetcher.py @@ -0,0 +1,164 @@ +#!/usr/bin/env python3 +""" +DataDog SIEM Configuration Retrieval + +Purpose: Retrieve SIEM operational state — suppression rules and notification integrations. +""" + +import json +import logging +import os +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional +from urllib.parse import urlparse + +import requests +from dotenv import load_dotenv + +logger = logging.getLogger("datadog_siem_configuration") + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def get_env(name: str, default: Optional[str] = None) -> str: + value = os.environ.get(name, default) + if value is None or value == "": + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def get_dd_headers() -> Dict[str, str]: + return { + "Content-Type": "application/json", + "DD-API-KEY": get_env("DATADOG_API_KEY"), + "DD-APPLICATION-KEY": get_env("DATADOG_APP_KEY"), + } + + +def get_base_url() -> str: + return os.environ.get("DATADOG_BASE_URL", "https://api.ddog-gov.com").rstrip("/") + + +def safe_get(url: str, headers: Dict[str, str], strict: bool = False) -> Optional[Dict[str, Any]]: + try: + response = requests.get(url, headers=headers, timeout=30) + response.raise_for_status() + return response.json() + except requests.exceptions.HTTPError as e: + if strict and e.response is not None and e.response.status_code in (403, 404): + raise + logger.warning("GET %s failed: %s", url, e) + return None + except requests.exceptions.RequestException as e: + logger.warning("GET %s failed: %s", url, e) + return None + + +def get_siem_configuration() -> Dict[str, Any]: + base_url = get_base_url() + headers = get_dd_headers() + suppression_endpoint = f"{base_url}/api/v2/security_monitoring/configuration/suppression_rules" + webhook_endpoint = f"{base_url}/api/v1/integration/webhook/configuration/webhooks" + + try: + # Suppression rules + suppression_payload = safe_get(suppression_endpoint, headers, strict=True) + suppression_data = [] + if suppression_payload: + raw = suppression_payload.get("data", []) + for item in raw: + attrs = item.get("attributes", {}) + suppression_data.append({ + "id": item.get("id"), + "name": attrs.get("name"), + "isEnabled": attrs.get("isEnabled", attrs.get("enabled")), + "startDate": attrs.get("startDate"), + "expirationDate": attrs.get("expirationDate"), + "rule_id": item.get("relationships", {}).get("rule", {}).get("data", {}).get("id"), + }) + + enabled_suppression = sum(1 for r in suppression_data if r.get("isEnabled")) + + # Webhook integrations — log domain only, never the full URL + webhook_payload = safe_get(webhook_endpoint, headers) + webhook_data = [] + notification_types = set() + if webhook_payload: + webhooks = webhook_payload.get("webhooks", []) + for wh in webhooks: + raw_url = wh.get("url", "") + domain = urlparse(raw_url).netloc if raw_url else "" + webhook_data.append({ + "name": wh.get("name"), + "url_domain": domain, + }) + notification_types.add("webhook") + + # Determine notification types from integration names (heuristic) + for wh in webhook_data: + name_lower = (wh.get("name") or "").lower() + if "slack" in name_lower: + notification_types.add("slack") + elif "pagerduty" in name_lower or "pd-" in name_lower: + notification_types.add("pagerduty") + + return { + "status": "success", + "api_endpoint": suppression_endpoint, + "api_endpoints_additional": [webhook_endpoint], + "record_count": len(suppression_data) + len(webhook_data), + "data": { + "suppression_rules": suppression_data, + "notification_integrations": webhook_data, + }, + "summary": { + "suppression_rules_count": len(suppression_data), + "enabled_suppression_rules": enabled_suppression, + "notification_integrations_count": len(webhook_data), + "notification_types": sorted(notification_types), + }, + "retrieved_at": current_timestamp(), + } + + except requests.exceptions.HTTPError as e: + if e.response is not None and e.response.status_code in (403, 404): + return { + "status": "not_available", + "message": "Security Monitoring (SIEM) is not enabled for this organization", + "api_endpoint": suppression_endpoint, + "retrieved_at": current_timestamp(), + } + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except RuntimeError as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except Exception as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + result = get_siem_configuration() + + output_json = output_dir / "datadog_siem_configuration.json" + with open(output_json, "w") as f: + json.dump(result, f, indent=2, default=str) + + logger.info("Evidence saved to %s", output_json) + + return 0 if result.get("status") in {"success", "partial_or_empty", "not_available"} else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/datadog/siem_configuration/fetcher.yaml b/fetchers/datadog/siem_configuration/fetcher.yaml new file mode 100644 index 0000000..00999c5 --- /dev/null +++ b/fetchers/datadog/siem_configuration/fetcher.yaml @@ -0,0 +1,27 @@ +name: datadog_siem_configuration +version: 0.1.0 +description: Retrieves Datadog SIEM operational state including suppression rules and notification integrations. +category: datadog + +supports_targets: false + +runtime: + type: python + entry: fetcher.py + +output: + type: json + path: datadog_siem_configuration.json + +secrets: +- name: api_key + env: DATADOG_API_KEY +- name: app_key + env: DATADOG_APP_KEY +- name: base_url + env: DATADOG_BASE_URL + +evidence_set: + reference_id: EVD-DD-SIEM-CONFIG + name: Datadog SIEM Configuration + instructions: 'Calls GET /api/v2/security_monitoring/suppressions and GET /api/v1/integration/webhooks. Writes JSON SIEM config summary.' diff --git a/fetchers/datadog/siem_detection_rules/fetcher.py b/fetchers/datadog/siem_detection_rules/fetcher.py new file mode 100644 index 0000000..e3d3f9f --- /dev/null +++ b/fetchers/datadog/siem_detection_rules/fetcher.py @@ -0,0 +1,182 @@ +#!/usr/bin/env python3 +""" +DataDog SIEM Detection Rules Retrieval + +Purpose: Retrieve custom SIEM detection rules to prove active threat detection is configured. +""" + +import json +import logging +import os +import sys +from collections import Counter +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +import requests +from dotenv import load_dotenv + +logger = logging.getLogger("datadog_siem_detection_rules") + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def get_env(name: str, default: Optional[str] = None) -> str: + value = os.environ.get(name, default) + if value is None or value == "": + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def get_dd_headers() -> Dict[str, str]: + return { + "Content-Type": "application/json", + "DD-API-KEY": get_env("DATADOG_API_KEY"), + "DD-APPLICATION-KEY": get_env("DATADOG_APP_KEY"), + } + + +def get_base_url() -> str: + return os.environ.get("DATADOG_BASE_URL", "https://api.ddog-gov.com").rstrip("/") + + +def fetch_all_rules(base_url: str, headers: Dict[str, str]) -> List[Dict[str, Any]]: + all_rules: List[Dict[str, Any]] = [] + page_number = 0 + page_size = 100 + endpoint = f"{base_url}/api/v2/security_monitoring/rules" + + while True: + params = { + "page[number]": page_number, + "page[size]": page_size, + "is_default": "false", + } + try: + response = requests.get(endpoint, headers=headers, params=params, timeout=30) + response.raise_for_status() + payload = response.json() + data = payload.get("data", []) + if not data: + break + all_rules.extend(data) + if len(data) < page_size: + break + page_number += 1 + except requests.exceptions.HTTPError as e: + if e.response is not None and e.response.status_code in (403, 404): + raise + logger.warning("Pagination interrupted at page %s: %s", page_number, e) + break + except requests.exceptions.RequestException as e: + logger.warning("Pagination interrupted at page %s: %s", page_number, e) + break + + return all_rules + + +def extract_rule_fields(rule: Dict[str, Any]) -> Dict[str, Any]: + cases = rule.get("cases", []) + severities = [c.get("status", "").lower() for c in cases if c.get("status")] + return { + "id": rule.get("id"), + "name": rule.get("name"), + "type": rule.get("type"), + "isEnabled": rule.get("isEnabled"), + "isDefault": rule.get("isDefault"), + "createdAt": rule.get("createdAt"), + "updatedAt": rule.get("updatedAt"), + "hasExtendedTitle": rule.get("hasExtendedTitle"), + "severities": severities, + } + + +def get_siem_detection_rules() -> Dict[str, Any]: + base_url = get_base_url() + headers = get_dd_headers() + endpoint = f"{base_url}/api/v2/security_monitoring/rules" + + try: + rules = fetch_all_rules(base_url, headers) + + extracted = [extract_rule_fields(r) for r in rules] + + enabled_rules = [r for r in extracted if r.get("isEnabled")] + disabled_rules = [r for r in extracted if not r.get("isEnabled")] + + type_counts: Dict[str, int] = dict(Counter(r.get("type", "unknown") for r in extracted)) + severity_counts: Dict[str, int] = {} + for r in extracted: + for sev in r.get("severities", []): + severity_counts[sev] = severity_counts.get(sev, 0) + 1 + + cutoff = datetime.now(timezone.utc) - timedelta(days=90) + rules_updated_recently = 0 + for r in extracted: + updated_at = r.get("updatedAt") + if updated_at: + try: + dt = datetime.strptime(updated_at, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) + if dt >= cutoff: + rules_updated_recently += 1 + except ValueError: + pass + + return { + "status": "success", + "api_endpoint": endpoint, + "record_count": len(extracted), + "data": extracted, + "summary": { + "total_rules": len(extracted), + "enabled_rules": len(enabled_rules), + "disabled_rules": len(disabled_rules), + "rules_never_triggered_count": 0, + "rules_updated_last_90_days": rules_updated_recently, + "rules_by_type": type_counts, + "rules_by_severity": severity_counts, + }, + "retrieved_at": current_timestamp(), + } + + except requests.exceptions.HTTPError as e: + if e.response is not None and e.response.status_code in (403, 404): + return { + "status": "not_available", + "message": "Security Monitoring (SIEM) is not enabled for this organization", + "api_endpoint": endpoint, + "retrieved_at": current_timestamp(), + } + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except RuntimeError as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except Exception as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + result = get_siem_detection_rules() + + output_json = output_dir / "datadog_siem_detection_rules.json" + with open(output_json, "w") as f: + json.dump(result, f, indent=2, default=str) + + logger.info("Evidence saved to %s", output_json) + + return 0 if result.get("status") in {"success", "partial_or_empty", "not_available"} else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/datadog/siem_detection_rules/fetcher.yaml b/fetchers/datadog/siem_detection_rules/fetcher.yaml new file mode 100644 index 0000000..8607ec7 --- /dev/null +++ b/fetchers/datadog/siem_detection_rules/fetcher.yaml @@ -0,0 +1,27 @@ +name: datadog_siem_detection_rules +version: 0.1.0 +description: Retrieves custom Datadog SIEM detection rules to prove active threat detection is configured. +category: datadog + +supports_targets: false + +runtime: + type: python + entry: fetcher.py + +output: + type: json + path: datadog_siem_detection_rules.json + +secrets: +- name: api_key + env: DATADOG_API_KEY +- name: app_key + env: DATADOG_APP_KEY +- name: base_url + env: DATADOG_BASE_URL + +evidence_set: + reference_id: EVD-DD-SIEM-RULES + name: Datadog SIEM Detection Rules + instructions: 'Calls GET /api/v2/security_monitoring/rules with pagination, filtered to custom rules. Extracts rule name, type, severity, enabled state, and detection method. Writes JSON rules list.' diff --git a/fetchers/datadog/siem_signals/fetcher.py b/fetchers/datadog/siem_signals/fetcher.py new file mode 100644 index 0000000..294afbf --- /dev/null +++ b/fetchers/datadog/siem_signals/fetcher.py @@ -0,0 +1,182 @@ +#!/usr/bin/env python3 +""" +DataDog SIEM Signals Retrieval + +Purpose: Retrieve security signals generated by detection rules to prove events are being detected and reviewed. +""" + +import json +import logging +import os +import sys +from collections import Counter +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +import requests + +from dotenv import load_dotenv + +logger = logging.getLogger("datadog_siem_signals") + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def get_env(name: str, default: Optional[str] = None) -> str: + value = os.environ.get(name, default) + if value is None or value == "": + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def get_dd_headers() -> Dict[str, str]: + return { + "Content-Type": "application/json", + "DD-API-KEY": get_env("DATADOG_API_KEY"), + "DD-APPLICATION-KEY": get_env("DATADOG_APP_KEY"), + } + + +def get_base_url() -> str: + return os.environ.get("DATADOG_BASE_URL", "https://api.ddog-gov.com").rstrip("/") + + +def fetch_all_signals( + base_url: str, + headers: Dict[str, str], + from_ts: str, + to_ts: str, +) -> List[Dict[str, Any]]: + all_signals: List[Dict[str, Any]] = [] + cursor: Optional[str] = None + endpoint = f"{base_url}/api/v2/security_monitoring/signals/search" + + while True: + body: Dict[str, Any] = { + "filter": {"from": from_ts, "to": to_ts}, + "page": {"limit": 100}, + } + if cursor: + body["page"]["cursor"] = cursor + + try: + response = requests.post(endpoint, headers=headers, json=body, timeout=30) + response.raise_for_status() + payload = response.json() + data = payload.get("data", []) + all_signals.extend(data) + + cursor = payload.get("meta", {}).get("page", {}).get("after") + if not cursor or not data: + break + except requests.exceptions.HTTPError as e: + if e.response is not None and e.response.status_code in (403, 404): + raise + logger.warning("Pagination interrupted: %s", e) + break + except requests.exceptions.RequestException as e: + logger.warning("Pagination interrupted: %s", e) + break + + return all_signals + + +def extract_signal_fields(signal: Dict[str, Any]) -> Dict[str, Any]: + attrs = signal.get("attributes", {}) + return { + "id": signal.get("id"), + "status": attrs.get("status"), + "severity": attrs.get("severity"), + "timestamp": attrs.get("timestamp"), + "rule_name": attrs.get("rule", {}).get("name"), + "rule_id": attrs.get("rule", {}).get("id"), + } + + +def get_siem_signals() -> Dict[str, Any]: + base_url = get_base_url() + headers = get_dd_headers() + endpoint = f"{base_url}/api/v2/security_monitoring/signals/search" + + lookback_days = int(os.environ.get("DATADOG_SIGNALS_LOOKBACK_DAYS", "30")) + now = datetime.now(timezone.utc) + from_dt = now - timedelta(days=lookback_days) + from_ts = from_dt.strftime("%Y-%m-%dT%H:%M:%SZ") + to_ts = now.strftime("%Y-%m-%dT%H:%M:%SZ") + + try: + signals = fetch_all_signals(base_url, headers, from_ts, to_ts) + extracted = [extract_signal_fields(s) for s in signals] + + severity_counts: Dict[str, int] = dict(Counter( + s.get("severity", "unknown") or "unknown" for s in extracted + )) + status_counts: Dict[str, int] = dict(Counter( + s.get("status", "unknown") or "unknown" for s in extracted + )) + + rule_hits: Counter = Counter( + s.get("rule_name", "unknown") or "unknown" for s in extracted + ) + top_triggering_rules = [ + {"rule_name": name, "count": count} + for name, count in rule_hits.most_common(10) + ] + + return { + "status": "success", + "api_endpoint": endpoint, + "record_count": len(extracted), + "data": extracted, + "summary": { + "total_signals": len(extracted), + "severity_counts": severity_counts, + "status_counts": status_counts, + "top_triggering_rules": top_triggering_rules, + "lookback_days": lookback_days, + "date_range": {"from": from_ts, "to": to_ts}, + }, + "retrieved_at": current_timestamp(), + } + + except requests.exceptions.HTTPError as e: + if e.response is not None and e.response.status_code in (403, 404): + return { + "status": "not_available", + "message": "Security Monitoring (SIEM) is not enabled for this organization", + "api_endpoint": endpoint, + "retrieved_at": current_timestamp(), + } + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except RuntimeError as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + except Exception as e: + return {"status": "error", "message": str(e), "retrieved_at": current_timestamp()} + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + result = get_siem_signals() # keep the original get_XXX() call + + output_json = output_dir / "datadog_siem_signals.json" + with open(output_json, "w") as f: + json.dump(result, f, indent=2, default=str) + + logger.info("Evidence saved to %s", output_json) + + return 0 if result.get("status") in {"success", "partial_or_empty", "not_available"} else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/datadog/siem_signals/fetcher.yaml b/fetchers/datadog/siem_signals/fetcher.yaml new file mode 100644 index 0000000..a5e570e --- /dev/null +++ b/fetchers/datadog/siem_signals/fetcher.yaml @@ -0,0 +1,28 @@ +name: datadog_siem_signals +version: 0.1.0 +description: Retrieves Datadog security signals generated by SIEM detection rules to prove events are being detected and reviewed. +category: datadog + +supports_targets: false + +runtime: + type: python + entry: fetcher.py + +output: + type: json + path: datadog_siem_signals.json + +secrets: +- name: api_key + env: DATADOG_API_KEY +- name: app_key + env: DATADOG_APP_KEY +- name: base_url + env: DATADOG_BASE_URL +- name: signals_lookback_days + env: DATADOG_SIGNALS_LOOKBACK_DAYS +evidence_set: + reference_id: EVD-DD-SIEM-SIGNALS + name: Datadog SIEM Signals + instructions: 'Calls POST /api/v2/security_monitoring/signals/search with a lookback window. Lookback controlled by DATADOG_SIGNALS_LOOKBACK_DAYS (default 30). Writes JSON signals list.' diff --git a/manifest.yaml b/manifest.yaml index 5f7f773..c4e5d58 100644 --- a/manifest.yaml +++ b/manifest.yaml @@ -1,7 +1,71 @@ run: output_dir: ./evidence fetchers: - - use: knowbe4_module_based_summary + - use: datadog_agent_hosts secrets: - api_key: ${env:KNOWBE4_API_KEY} - region: ${env:KNOWBE4_REGION} + api_key: ${env:DATADOG_API_KEY} + app_key: ${env:DATADOG_APP_KEY} + base_url: ${env:DATADOG_BASE_URL} + - use: datadog_apm_services + secrets: + api_key: ${env:DATADOG_API_KEY} + app_key: ${env:DATADOG_APP_KEY} + base_url: ${env:DATADOG_BASE_URL} + - use: datadog_containers + secrets: + api_key: ${env:DATADOG_API_KEY} + app_key: ${env:DATADOG_APP_KEY} + base_url: ${env:DATADOG_BASE_URL} + - use: datadog_incident_timelines + secrets: + api_key: ${env:DATADOG_API_KEY} + app_key: ${env:DATADOG_APP_KEY} + base_url: ${env:DATADOG_BASE_URL} + incidents_lookback_days: ${env:DATADOG_INCIDENTS_LOOKBACK_DAYS} + - use: datadog_incidents_list + secrets: + api_key: ${env:DATADOG_API_KEY} + app_key: ${env:DATADOG_APP_KEY} + base_url: ${env:DATADOG_BASE_URL} + incidents_lookback_days: ${env:DATADOG_INCIDENTS_LOOKBACK_DAYS} + - use: datadog_infra_agent_checks + secrets: + api_key: ${env:DATADOG_API_KEY} + app_key: ${env:DATADOG_APP_KEY} + base_url: ${env:DATADOG_BASE_URL} + - use: datadog_log_archives + secrets: + api_key: ${env:DATADOG_API_KEY} + app_key: ${env:DATADOG_APP_KEY} + base_url: ${env:DATADOG_BASE_URL} + - use: datadog_log_indexes + secrets: + api_key: ${env:DATADOG_API_KEY} + app_key: ${env:DATADOG_APP_KEY} + base_url: ${env:DATADOG_BASE_URL} + - use: datadog_log_pipelines + secrets: + api_key: ${env:DATADOG_API_KEY} + app_key: ${env:DATADOG_APP_KEY} + base_url: ${env:DATADOG_BASE_URL} + - use: datadog_monitors_list + secrets: + api_key: ${env:DATADOG_API_KEY} + app_key: ${env:DATADOG_APP_KEY} + base_url: ${env:DATADOG_BASE_URL} + - use: datadog_siem_configuration + secrets: + api_key: ${env:DATADOG_API_KEY} + app_key: ${env:DATADOG_APP_KEY} + base_url: ${env:DATADOG_BASE_URL} + - use: datadog_siem_detection_rules + secrets: + api_key: ${env:DATADOG_API_KEY} + app_key: ${env:DATADOG_APP_KEY} + base_url: ${env:DATADOG_BASE_URL} + - use: datadog_siem_signals + secrets: + api_key: ${env:DATADOG_API_KEY} + app_key: ${env:DATADOG_APP_KEY} + base_url: ${env:DATADOG_BASE_URL} + signals_lookback_days: ${env:DATADOG_SIGNALS_LOOKBACK_DAYS}