Security Hub CSPM に集約した GuardDuty の検知を DevOps Agent で調査し Slack 通知までを試してみた

"""
Lambda handler: forward Security Hub / GuardDuty findings to DevOps Agent
via Generic Webhook with HMAC-SHA256 authentication.

Event source  : EventBridge rule matching Security Hub Findings - Imported
                where ProductName = GuardDuty
Destination   : DevOps Agent Generic Webhook (HTTP POST, JSON body)
Dependencies  : standard library only (hmac, hashlib, base64, datetime,
                urllib, json, os, logging)
"""

import base64
import hashlib
import hmac
import json
import logging
import os
import urllib.error
import urllib.request
from datetime import datetime, timezone
from typing import Any

# ---------------------------------------------------------------------------
# Logger setup
# ---------------------------------------------------------------------------
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()
logging.basicConfig(
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
    level=getattr(logging, LOG_LEVEL, logging.INFO),
)
logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
WEBHOOK_URL: str = os.environ["DEVOPS_AGENT_WEBHOOK_URL"]
WEBHOOK_SECRET: str = os.environ["DEVOPS_AGENT_WEBHOOK_SECRET"]
REQUEST_TIMEOUT_SECONDS: int = 10

# ---------------------------------------------------------------------------
# Severity mapping: ASFF Severity Label -> DevOps Agent priority
# ---------------------------------------------------------------------------
_SEVERITY_TO_PRIORITY: dict[str, str] = {
    "INFORMATIONAL": "MINIMAL",
    "LOW": "LOW",
    "MEDIUM": "MEDIUM",
    "HIGH": "HIGH",
    "CRITICAL": "CRITICAL",
}


def _severity_to_priority(severity_label: str) -> str:
    """Map ASFF Severity Label to DevOps Agent priority.

    Falls back to MEDIUM for unrecognized values.
    """
    return _SEVERITY_TO_PRIORITY.get(severity_label.upper(), "MEDIUM")


def _now_iso8601() -> str:
    """Return current UTC time in ISO8601 format with milliseconds and Z suffix.

    Example: 2025-11-23T18:00:00.000Z
    """
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z")


def _build_payload(finding: dict[str, Any], timestamp: str) -> dict[str, Any]:
    """Build a DevOps Agent Generic Webhook payload from an ASFF finding.

    Parameters
    ----------
    finding   : ASFF finding dict from Security Hub event
    timestamp : ISO8601 timestamp string (same value used in HMAC header)

    Returns
    -------
    Payload dict conforming to the DevOps Agent Generic Webhook schema.
    """
    severity_label: str = finding.get("Severity", {}).get("Label", "")
    priority = _severity_to_priority(severity_label)

    return {
        "eventType": "incident",
        "incidentId": finding.get("Id", ""),
        "action": "created",
        "priority": priority,
        "title": finding.get("Title", ""),
        "description": finding.get("Description", ""),
        "timestamp": timestamp,
        "service": "GuardDuty",
        "data": {
            "aws_account_id": finding.get("AwsAccountId", ""),
            "region": finding.get("Region", ""),
            "types": finding.get("Types", []),
            "resources": finding.get("Resources", []),
            "asff_finding": finding,
        },
    }


def _compute_signature(secret: str, timestamp: str, payload_json: str) -> str:
    """Compute HMAC-SHA256 signature for the DevOps Agent Webhook request.

    Formula: base64( hmac_sha256(secret, f"{timestamp}:{payload_json}") )

    Parameters
    ----------
    secret       : Webhook HMAC secret (from environment variable)
    timestamp    : ISO8601 timestamp string (same value as x-amzn-event-timestamp header)
    payload_json : Serialized JSON body string (must match the POST body exactly)

    Returns
    -------
    Base64-encoded HMAC-SHA256 digest string.
    """
    message = f"{timestamp}:{payload_json}".encode("utf-8")
    digest = hmac.new(
        key=secret.encode("utf-8"),
        msg=message,
        digestmod=hashlib.sha256,
    ).digest()
    return base64.b64encode(digest).decode("utf-8")


def _post_to_webhook(payload: dict[str, Any], timestamp: str) -> int:
    """POST a JSON payload to the DevOps Agent Webhook URL with HMAC authentication.

    Uses deterministic JSON serialization (no extra spaces) to ensure the
    POST body matches exactly what was signed.

    Parameters
    ----------
    payload   : DevOps Agent payload dict
    timestamp : ISO8601 timestamp string used in both HMAC and header

    Returns
    -------
    HTTP status code.

    Raises
    ------
    urllib.error.HTTPError / urllib.error.URLError on failure or non-200 response.
    """
    # Deterministic serialization: separators=(",", ":") removes extra spaces
    payload_json = json.dumps(payload, separators=(",", ":"))
    body = payload_json.encode("utf-8")

    signature = _compute_signature(WEBHOOK_SECRET, timestamp, payload_json)

    request = urllib.request.Request(
        url=WEBHOOK_URL,
        data=body,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "x-amzn-event-timestamp": timestamp,
            "x-amzn-event-signature": signature,
        },
    )
    with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
        status: int = response.status
        if status != 200:
            raise urllib.error.HTTPError(
                url=WEBHOOK_URL,
                code=status,
                msg=f"Unexpected status code: {status}",
                hdrs=response.headers,
                fp=None,
            )
        logger.info("Webhook responded with HTTP %d", status)
        return status


def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
    """Entry point for the Lambda function.

    Iterates over findings in the Security Hub event and posts each one
    to the DevOps Agent Generic Webhook with HMAC-SHA256 authentication.

    Parameters
    ----------
    event   : EventBridge event wrapping Security Hub Findings - Imported
    context : Lambda context object (unused)

    Returns
    -------
    dict with statusCode and a summary of processed findings.
    """
    logger.debug("Received event: %s", json.dumps(event))

    findings: list[dict[str, Any]] = event.get("detail", {}).get("findings", [])
    if not findings:
        logger.warning("No findings in event; nothing to do.")
        return {"statusCode": 200, "processed": 0}

    logger.info("Processing %d finding(s)", len(findings))

    processed = 0
    for finding in findings:
        # Generate a fresh timestamp per finding for HMAC
        timestamp = _now_iso8601()
        payload = _build_payload(finding, timestamp)
        finding_id: str = payload["incidentId"]

        logger.info(
            "Posting finding id=%s title=%s priority=%s",
            finding_id,
            payload["title"],
            payload["priority"],
        )
        try:
            status = _post_to_webhook(payload, timestamp)
            logger.info("Successfully posted finding %s (HTTP %d)", finding_id, status)
            processed += 1
        except urllib.error.HTTPError as exc:
            logger.error(
                "HTTP error posting finding %s: %d %s",
                finding_id,
                exc.code,
                exc.reason,
            )
            raise
        except urllib.error.URLError as exc:
            logger.error(
                "URL error posting finding %s: %s", finding_id, exc.reason
            )
            raise

    logger.info("Done. processed=%d / total=%d", processed, len(findings))
    return {"statusCode": 200, "processed": processed}