From 9c1ec799599fa64b72209224f7ab61808d6e9643 Mon Sep 17 00:00:00 2001 From: OpenCode Date: Fri, 14 Aug 2026 17:46:19 +0000 Subject: [PATCH 1/2] Add read-only Security Hub CSPM collector --- scripts/README.md | 20 ++++ scripts/requirements.txt | 1 + scripts/securityhub_collect.py | 195 +++++++++++++++++++++++++++++++++ 3 files changed, 216 insertions(+) create mode 100644 scripts/README.md create mode 100644 scripts/requirements.txt create mode 100644 scripts/securityhub_collect.py diff --git a/scripts/README.md b/scripts/README.md new file mode 100644 index 00000000..361e067b --- /dev/null +++ b/scripts/README.md @@ -0,0 +1,20 @@ +# Security Hub/CSPM collector + +This read-only script exports AWS Security Hub findings, including CSPM +control findings, to JSON for review or a later Jira importer. + +```bash +python -m pip install -r scripts/requirements.txt +aws sso login --profile my-sandbox +python scripts/securityhub_collect.py \ + --profile my-sandbox --region us-east-1 \ + --severity HIGH,CRITICAL --status NEW \ + --output securityhub-findings.json +``` + +Required permissions are `securityhub:DescribeHub`, +`securityhub:GetFindings`, and `sts:GetCallerIdentity`. The script does not +create, update, or resolve findings. Use a read-only/sandbox role first. + +The output includes a stable `key` per finding (account, region, product, and +finding ID), which is intended for idempotent Jira creation later. diff --git a/scripts/requirements.txt b/scripts/requirements.txt new file mode 100644 index 00000000..011ba23b --- /dev/null +++ b/scripts/requirements.txt @@ -0,0 +1 @@ +boto3>=1.34.0 diff --git a/scripts/securityhub_collect.py b/scripts/securityhub_collect.py new file mode 100644 index 00000000..08dc614b --- /dev/null +++ b/scripts/securityhub_collect.py @@ -0,0 +1,195 @@ +#!/usr/bin/env python3 +"""Export AWS Security Hub/CSPM findings as Jira-ready JSON. + +The collector is read-only. It uses the Security Hub ``get_findings`` API, +which is also the API used for CSPM control findings. Credentials and +secrets are never written to the output. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable + +try: + import boto3 +except ImportError: # pragma: no cover - gives a useful CLI error + boto3 = None + + +SEVERITIES = {"INFORMATIONAL", "LOW", "MEDIUM", "HIGH", "CRITICAL"} +STATUSES = {"NEW", "NOTIFIED", "RESOLVED", "SUPPRESSED"} + + +def _value(obj: Any, *keys: str, default: Any = None) -> Any: + for key in keys: + if isinstance(obj, dict) and obj.get(key) is not None: + return obj[key] + return default + + +def _finding_key(finding: dict[str, Any], account_id: str, region: str) -> str: + """Return a stable key suitable for deduplication in a later Jira step.""" + product = _value( + finding.get("ProductFields", {}), "aws/securityhub/ProductName", default="" + ) + return "|".join((account_id, region, str(product), str(finding.get("Id", "")))) + + +def normalize_finding( + finding: dict[str, Any], account_id: str, region: str +) -> dict[str, Any]: + severity = _value(finding.get("Severity", {}), "Label", default="UNKNOWN") + workflow = _value(finding.get("Workflow", {}), "Status", default="UNKNOWN") + resources = finding.get("Resources") or [] + return { + "key": _finding_key(finding, account_id, region), + "id": finding.get("Id"), + "product_arn": finding.get("ProductArn"), + "product_name": _value( + finding.get("ProductFields", {}), "aws/securityhub/ProductName" + ), + "control_id": _value( + finding.get("ProductFields", {}), "ControlId", "aws/securityhub/ControlId" + ), + "title": finding.get("Title"), + "description": finding.get("Description"), + "remediation": finding.get("Remediation", {}) + .get("Recommendation", {}) + .get("Text"), + "severity": str(severity).upper(), + "status": str(workflow).upper(), + "compliance": finding.get("Compliance", {}), + "account_id": account_id, + "region": region, + "resource_ids": [r.get("Id") for r in resources if r.get("Id")], + "resource_types": [r.get("Type") for r in resources if r.get("Type")], + "created_at": finding.get("CreatedAt"), + "updated_at": finding.get("UpdatedAt"), + "generator_id": finding.get("GeneratorId"), + "aws_console_url": ( + f"https://{region}.console.aws.amazon.com/securityhub/home" + f"?region={region}#/findings?search=Id%3D{finding.get('Id', '')}" + ), + } + + +def collect_findings( + client: Any, + account_id: str, + region: str, + severities: set[str] | None = None, + statuses: set[str] | None = None, + max_findings: int | None = None, +) -> list[dict[str, Any]]: + filters: dict[str, list[dict[str, Any]]] = {} + if severities: + filters["SeverityLabel"] = [ + {"Value": value, "Comparison": "EQUALS"} for value in sorted(severities) + ] + if statuses: + filters["WorkflowStatus"] = [ + {"Value": value, "Comparison": "EQUALS"} for value in sorted(statuses) + ] + + findings: list[dict[str, Any]] = [] + request: dict[str, Any] = {"MaxResults": 100} + if filters: + request["Filters"] = filters + while True: + response = client.get_findings(**request) + findings.extend( + normalize_finding(item, account_id, region) + for item in response.get("Findings", []) + ) + if max_findings and len(findings) >= max_findings: + return findings[:max_findings] + token = response.get("NextToken") + if not token: + return findings + request["NextToken"] = token + + +def collect( + region: str, + profile: str | None, + severities: set[str] | None, + statuses: set[str] | None, + max_findings: int | None, +) -> dict[str, Any]: + if boto3 is None: + raise RuntimeError( + "boto3 is required; install it with: python -m pip install boto3" + ) + session = boto3.Session(profile_name=profile, region_name=region) + client = session.client("securityhub", region_name=region) + identity = session.client("sts", region_name=region).get_caller_identity() + account_id = identity["Account"] + try: + hub = client.describe_hub() + except client.exceptions.ResourceNotFoundException: + hub = None + return { + "schema_version": 1, + "collected_at": datetime.now(timezone.utc).isoformat(), + "account_id": account_id, + "region": region, + "hub": hub, + "findings": collect_findings( + client, account_id, region, severities, statuses, max_findings + ), + } + + +def _csv_values(value: str | None, allowed: set[str], name: str) -> set[str] | None: + if not value: + return None + values = {item.strip().upper() for item in value.split(",") if item.strip()} + invalid = values - allowed + if invalid: + raise ValueError(f"invalid {name}: {', '.join(sorted(invalid))}") + return values + + +def main(argv: Iterable[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--region", default=None, help="AWS region (default: SDK/config region)" + ) + parser.add_argument("--profile", help="AWS profile name") + parser.add_argument("--severity", help="comma-separated severity labels") + parser.add_argument("--status", help="comma-separated workflow statuses") + parser.add_argument("--max-findings", type=int, help="maximum findings to export") + parser.add_argument( + "--output", type=Path, default=Path("securityhub-findings.json") + ) + args = parser.parse_args(list(argv) if argv is not None else None) + try: + severities = _csv_values(args.severity, SEVERITIES, "severity") + statuses = _csv_values(args.status, STATUSES, "status") + if args.max_findings is not None and args.max_findings < 1: + raise ValueError("--max-findings must be positive") + if boto3 is None: + raise RuntimeError( + "boto3 is required; install it with: python -m pip install boto3" + ) + region = args.region or boto3.Session(profile_name=args.profile).region_name + if not region: + raise ValueError("no region configured; pass --region") + result = collect(region, args.profile, severities, statuses, args.max_findings) + args.output.write_text( + json.dumps(result, indent=2, default=str) + "\n", encoding="utf-8" + ) + print(f"Wrote {len(result['findings'])} findings to {args.output}") + return 0 + except Exception as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) From 1a37b0fea08586ef590593c0d64fe22ebd042cf9 Mon Sep 17 00:00:00 2001 From: OpenCode Date: Fri, 14 Aug 2026 17:48:52 +0000 Subject: [PATCH 2/2] Harden Security Hub collector and add tests --- scripts/securityhub_collect.py | 26 +++++++++-- scripts/test_securityhub_collect.py | 70 +++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+), 4 deletions(-) create mode 100644 scripts/test_securityhub_collect.py diff --git a/scripts/securityhub_collect.py b/scripts/securityhub_collect.py index 08dc614b..7dcdc100 100644 --- a/scripts/securityhub_collect.py +++ b/scripts/securityhub_collect.py @@ -9,6 +9,7 @@ from __future__ import annotations import argparse +import hashlib import json import sys from datetime import datetime, timezone @@ -37,7 +38,26 @@ def _finding_key(finding: dict[str, Any], account_id: str, region: str) -> str: product = _value( finding.get("ProductFields", {}), "aws/securityhub/ProductName", default="" ) - return "|".join((account_id, region, str(product), str(finding.get("Id", "")))) + identity = { + "account_id": account_id, + "region": region, + "product": product, + "id": finding.get("Id", ""), + } + encoded = json.dumps( + identity, sort_keys=True, separators=(",", ":"), default=str + ).encode() + return hashlib.sha256(encoded).hexdigest() + + +def _recommendation_text(finding: dict[str, Any]) -> Any: + remediation = finding.get("Remediation") + if not isinstance(remediation, dict): + return None + recommendation = remediation.get("Recommendation") + if not isinstance(recommendation, dict): + return None + return recommendation.get("Text") def normalize_finding( @@ -58,9 +78,7 @@ def normalize_finding( ), "title": finding.get("Title"), "description": finding.get("Description"), - "remediation": finding.get("Remediation", {}) - .get("Recommendation", {}) - .get("Text"), + "remediation": _recommendation_text(finding), "severity": str(severity).upper(), "status": str(workflow).upper(), "compliance": finding.get("Compliance", {}), diff --git a/scripts/test_securityhub_collect.py b/scripts/test_securityhub_collect.py new file mode 100644 index 00000000..0c661709 --- /dev/null +++ b/scripts/test_securityhub_collect.py @@ -0,0 +1,70 @@ +import importlib.util +from pathlib import Path + + +SPEC = importlib.util.spec_from_file_location( + "securityhub_collect", Path(__file__).with_name("securityhub_collect.py") +) +module = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(module) + + +class FakeClient: + class exceptions: + ResourceNotFoundException = type("ResourceNotFoundException", (Exception,), {}) + + def __init__(self, pages): + self.pages = iter(pages) + self.requests = [] + + def get_findings(self, **request): + self.requests.append(request) + return next(self.pages) + + +def finding(identifier="finding-1", remediation=None): + value = {"Id": identifier, "Title": "Example", "Resources": []} + if remediation is not None: + value["Remediation"] = remediation + return value + + +def test_collect_findings_paginates_and_applies_filters(): + client = FakeClient( + [ + {"Findings": [finding()], "NextToken": "next"}, + {"Findings": [finding("finding-2")]}, + ] + ) + results = module.collect_findings(client, "123", "us-east-1", {"HIGH"}, {"NEW"}) + assert [item["id"] for item in results] == ["finding-1", "finding-2"] + assert client.requests[0]["Filters"] == { + "SeverityLabel": [{"Value": "HIGH", "Comparison": "EQUALS"}], + "WorkflowStatus": [{"Value": "NEW", "Comparison": "EQUALS"}], + } + assert client.requests[1]["NextToken"] == "next" + + +def test_normalize_handles_missing_and_null_remediation(): + assert ( + module.normalize_finding(finding(), "123", "us-east-1")["remediation"] is None + ) + assert ( + module.normalize_finding(finding("2", None), "123", "us-east-1")["remediation"] + is None + ) + assert ( + module.normalize_finding( + finding("3", {"Recommendation": None}), "123", "us-east-1" + )["remediation"] + is None + ) + + +def test_finding_key_is_collision_safe_and_stable(): + first = finding("a|b") + second = finding("a", {"Recommendation": {"Text": "b"}}) + key1 = module._finding_key(first, "account|1", "region") + key2 = module._finding_key(second, "account", "1|region") + assert key1 != key2 + assert key1 == module._finding_key(first, "account|1", "region")