feat(sdk): add --export-ocsf flag for OCSF ingestion to Prowler Cloud (#10095)

Author: AdriiiPRodri

poetry.lock

@@ -22,6 +22,8 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - OpenStack compute 7 new checks [(#9944)](https://github.com/prowler-cloud/prowler/pull/9944)
 - CSA CCM 4.0 for the Alibaba Cloud provider [(#10061)](https://github.com/prowler-cloud/prowler/pull/10061)
 - ECS Exec (ECS-006) privilege escalation detection via `ecs:ExecuteCommand` + `ecs:DescribeTasks` [(#10066)](https://github.com/prowler-cloud/prowler/pull/10066)
+- `--export-ocsf` CLI flag to upload OCSF scan results to Prowler Cloud [(#10095)](https://github.com/prowler-cloud/prowler/pull/10095)
+- `scan_id` field in OCSF `unmapped` output for ingestion correlation [(#10095)](https://github.com/prowler-cloud/prowler/pull/10095)
 - `defenderxdr_endpoint_privileged_user_exposed_credentials` check for M365 provider [(#10084)](https://github.com/prowler-cloud/prowler/pull/10084)
 - `defenderxdr_critical_asset_management_pending_approvals` check for M365 provider [(#10085)](https://github.com/prowler-cloud/prowler/pull/10085)
 - `entra_seamless_sso_disabled` check for m365 provider [(#10086)](https://github.com/prowler-cloud/prowler/pull/10086)

prowler/CHANGELOG.md

@@ -2,13 +2,16 @@
 # -*- coding: utf-8 -*-
 
 import sys
+import tempfile
 from os import environ
 
+import requests
 from colorama import Fore, Style
 from colorama import init as colorama_init
 
 from prowler.config.config import (
     EXTERNAL_TOOL_PROVIDERS,
+    cloud_api_base_url,
     csv_file_suffix,
     get_available_compliance_frameworks,
     html_file_suffix,
@@ -110,6 +113,7 @@
 from prowler.lib.outputs.csv.csv import CSV
 from prowler.lib.outputs.finding import Finding
 from prowler.lib.outputs.html.html import HTML
+from prowler.lib.outputs.ocsf.ingestion import send_ocsf_to_api
 from prowler.lib.outputs.ocsf.ocsf import OCSF
 from prowler.lib.outputs.outputs import extract_findings_statistics, report
 from prowler.lib.outputs.slack.slack import Slack
@@ -477,6 +481,7 @@ def streaming_callback(findings_batch):
             sys.exit(1)
 
     generated_outputs = {"regular": [], "compliance": []}
+    ocsf_output = None
 
     if args.output_formats:
         for mode in args.output_formats:
@@ -507,6 +512,7 @@ def streaming_callback(findings_batch):
                     file_path=f"{filename}{json_ocsf_file_suffix}",
                 )
                 generated_outputs["regular"].append(json_output)
+                ocsf_output = json_output
                 json_output.batch_write_data_to_file()
             if mode == "html":
                 html_output = HTML(
@@ -518,6 +524,57 @@ def streaming_callback(findings_batch):
                     provider=global_provider, stats=stats
                 )
 
+    if getattr(args, "export_ocsf", False):
+        if not ocsf_output or not getattr(ocsf_output, "file_path", None):
+            tmp_ocsf = tempfile.NamedTemporaryFile(
+                suffix=json_ocsf_file_suffix, delete=False
+            )
+            ocsf_output = OCSF(
+                findings=finding_outputs,
+                file_path=tmp_ocsf.name,
+            )
+            tmp_ocsf.close()
+            ocsf_output.batch_write_data_to_file()
+        print(
+            f"{Style.BRIGHT}\nExporting OCSF to Prowler Cloud, please wait...{Style.RESET_ALL}"
+        )
+        try:
+            response = send_ocsf_to_api(ocsf_output.file_path)
+        except ValueError:
+            logger.warning(
+                "OCSF export skipped: no API key configured. "
+                "Set the PROWLER_API_KEY environment variable to enable it. "
+                f"Scan results were saved to {ocsf_output.file_path}"
+            )
+        except requests.ConnectionError:
+            logger.warning(
+                "OCSF export skipped: could not reach the Prowler Cloud API at "
+                f"{cloud_api_base_url}. Check the URL and your network connection. "
+                f"Scan results were saved to {ocsf_output.file_path}"
+            )
+        except requests.HTTPError as http_err:
+            logger.warning(
+                f"OCSF export failed: the API returned HTTP {http_err.response.status_code}. "
+                "Verify your API key is valid and has the right permissions. "
+                f"Scan results were saved to {ocsf_output.file_path}"
+            )
+        except Exception as error:
+            logger.warning(
+                f"OCSF export failed unexpectedly: {error}. "
+                f"Scan results were saved to {ocsf_output.file_path}"
+            )
+        else:
+            job_id = response.get("data", {}).get("id") if response else None
+            if job_id:
+                print(
+                    f"{Style.BRIGHT}{Fore.GREEN}\nOCSF export accepted. Ingestion job: {job_id}{Style.RESET_ALL}"
+                )
+            else:
+                logger.warning(
+                    "OCSF export: unexpected API response (missing ingestion job ID). "
+                    f"Scan results were saved to {ocsf_output.file_path}"
+                )
+
     # Compliance Frameworks
     input_compliance_frameworks = set(output_options.output_modes).intersection(
         get_available_compliance_frameworks(provider)

prowler/__main__.py

@@ -120,6 +120,11 @@ def get_available_compliance_frameworks(provider=None):
 encoding_format_utf_8 = "utf-8"
 available_output_formats = ["csv", "json-asff", "json-ocsf", "html"]
 
+# Prowler Cloud API settings
+cloud_api_base_url = os.getenv("PROWLER_CLOUD_API_BASE", "https://api.prowler.com")
+cloud_api_key = os.getenv("PROWLER_API_KEY", "")
+cloud_api_ingestion_path = "/api/v1/ingestions"
+
 
 def set_output_timestamp(
     new_timestamp: datetime,

prowler/config/config.py

@@ -215,6 +215,14 @@ def __init_outputs_parser__(self):
             default=False,
             help="Set the output timestamp format as unix timestamps instead of iso format timestamps (default mode).",
         )
+        common_outputs_parser.add_argument(
+            "--export-ocsf",
+            action="store_true",
+            help=(
+                "Send OCSF output to Prowler Cloud ingestion endpoint. "
+                "Requires PROWLER_API_KEY environment variable."
+            ),
+        )
 
     def __init_logging_parser__(self):
         # Logging Options

prowler/lib/cli/parser.py

@@ -0,0 +1,65 @@
+import os
+from typing import Any, Dict, Optional
+
+import requests
+
+from prowler.config.config import (
+    cloud_api_base_url,
+    cloud_api_ingestion_path,
+    cloud_api_key,
+)
+
+
+def send_ocsf_to_api(
+    file_path: str,
+    *,
+    base_url: Optional[str] = None,
+    api_key: Optional[str] = None,
+    timeout: int = 60,
+) -> Dict[str, Any]:
+    """Send OCSF file to the Prowler Cloud ingestion endpoint.
+
+    Args:
+        file_path: Path to the OCSF JSON file to upload.
+        base_url: API base URL. Falls back to PROWLER_CLOUD_API_BASE env var,
+                  then to https://api.prowler.com.
+        api_key: API key. Falls back to PROWLER_API_KEY env var.
+        timeout: Request timeout in seconds.
+
+    Returns:
+        Parsed JSON:API response dict.
+
+    Raises:
+        FileNotFoundError: If the OCSF file does not exist.
+        ValueError: If no API key is available.
+        requests.HTTPError: If the API returns an error status.
+    """
+    if not file_path:
+        raise ValueError("No OCSF file path provided.")
+
+    if not os.path.isfile(file_path):
+        raise FileNotFoundError(f"OCSF file not found: {file_path}")
+
+    api_key = api_key or cloud_api_key
+    if not api_key:
+        raise ValueError("Missing API key. Set PROWLER_API_KEY environment variable.")
+
+    base_url = base_url or cloud_api_base_url
+    base_url = base_url.rstrip("/")
+    if not base_url.lower().startswith(("http://", "https://")):
+        base_url = f"https://{base_url}"
+
+    url = f"{base_url}{cloud_api_ingestion_path}"
+
+    with open(file_path, "rb") as fh:
+        response = requests.post(
+            url,
+            headers={
+                "Authorization": f"Api-Key {api_key}",
+                "Accept": "application/vnd.api+json",
+            },
+            files={"file": (os.path.basename(file_path), fh, "application/json")},
+            timeout=timeout,
+        )
+    response.raise_for_status()
+    return response.json() if response.text else {}

prowler/lib/outputs/ocsf/ingestion.py

@@ -1,6 +1,7 @@
 import json
 import os
-from datetime import datetime
+from datetime import datetime, timezone
+from random import getrandbits
 from typing import List
 
 from py_ocsf_models.events.base_event import SeverityID, StatusID
@@ -17,6 +18,7 @@
 from py_ocsf_models.objects.product import Product
 from py_ocsf_models.objects.remediation import Remediation
 from py_ocsf_models.objects.resource_details import ResourceDetails
+from uuid6 import UUID
 
 from prowler.lib.logger import logger
 from prowler.lib.outputs.finding import Finding
@@ -52,6 +54,10 @@ def transform(self, findings: List[Finding]) -> None:
             findings (List[Finding]): a list of Finding objects
         """
         try:
+            if not findings:
+                return
+
+            scan_id = _uuid7_from_timestamp(findings[0].timestamp)
             for finding in findings:
                 finding_activity = ActivityID.Create
                 cloud_account_type = self.get_account_type_id_by_provider(
@@ -163,6 +169,7 @@ def transform(self, findings: List[Finding]) -> None:
                         "additional_urls": finding.metadata.AdditionalURLs,
                         "notes": finding.metadata.Notes,
                         "compliance": finding.compliance,
+                        "scan_id": str(scan_id),
                     },
                 )
                 if finding.provider != "kubernetes":
@@ -295,3 +302,26 @@ def get_finding_status_id(muted: bool) -> StatusID:
         if muted:
             status_id = StatusID.Suppressed
         return status_id
+
+
+# NOTE: Copied from api/src/backend/api/uuid_utils.py (datetime_to_uuid7)
+# Adapted to accept datetime/epoch inputs.
+def _uuid7_from_timestamp(value) -> UUID:
+    if isinstance(value, datetime):
+        dt = value
+    else:
+        dt = datetime.fromtimestamp(int(value), tz=timezone.utc)
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+
+    timestamp_ms = int(dt.timestamp() * 1000) & 0xFFFFFFFFFFFF
+    rand_seq = getrandbits(12)
+    rand_node = getrandbits(62)
+
+    uuid_int = timestamp_ms << 80
+    uuid_int |= 0x7 << 76
+    uuid_int |= rand_seq << 64
+    uuid_int |= 0x2 << 62
+    uuid_int |= rand_node
+
+    return UUID(int=uuid_int)

prowler/lib/outputs/ocsf/ocsf.py

@@ -70,6 +70,7 @@ dependencies = [
   "slack-sdk==3.39.0",
   "tabulate==0.9.0",
   "tzlocal==5.3.1",
+  "uuid6==2024.7.10",
   "py-iam-expand==0.1.0",
   "h2==4.3.0",
   "oci==2.160.3",

pyproject.toml

@@ -2,6 +2,7 @@
 from datetime import datetime, timezone
 from io import StringIO
 from typing import Optional
+from uuid import UUID
 
 import requests
 from freezegun import freeze_time
@@ -101,7 +102,10 @@ def test_transform(self):
             output_data.type_name
             == f"Detection Finding: {DetectionFindingTypeID.Create.name}"
         )
-        assert output_data.unmapped == {
+        unmapped = output_data.unmapped
+        scan_id = unmapped.pop("scan_id")
+        assert UUID(scan_id)  # Valid UUID
+        assert unmapped == {
             "related_url": findings[0].metadata.RelatedUrl,
             "categories": findings[0].metadata.Categories,
             "depends_on": findings[0].metadata.DependsOn,
@@ -260,7 +264,11 @@ def test_batch_write_data_to_file(self):
 
         mock_file.seek(0)
         content = mock_file.read()
-        assert json.loads(content) == expected_json_output
+        actual_output = json.loads(content)
+        # scan_id is non-deterministic (UUID7), validate and remove before comparison
+        actual_scan_id = actual_output[0]["unmapped"].pop("scan_id")
+        assert UUID(actual_scan_id)
+        assert actual_output == expected_json_output
 
     def test_batch_write_data_to_file_without_findings(self):
         assert not OCSF([])._file_descriptor
@@ -318,7 +326,10 @@ def test_finding_output_cloud_pass_low_muted(self):
         assert finding_ocsf.risk_details == finding_output.metadata.Risk
 
         # Unmapped Data
-        assert finding_ocsf.unmapped == {
+        unmapped = finding_ocsf.unmapped
+        scan_id = unmapped.pop("scan_id")
+        assert UUID(scan_id)  # Valid UUID
+        assert unmapped == {
             "related_url": finding_output.metadata.RelatedUrl,
             "categories": finding_output.metadata.Categories,
             "depends_on": finding_output.metadata.DependsOn,