add a uniTest and fixed

checkov/common/output/in_toto_output.py

@@ -1,3 +1,104 @@
+# import json
+# from datetime import datetime, timezone
+# from typing import Any, Dict, List, Union
+# from checkov.common.output.report import Report
+# import os
+#
+#
+# class InTotoOutput:
+#     def __init__(self, repo_id: Union[str, None], reports: List[Report]):
+#         self.repo_id = f"{repo_id}/" if repo_id else ""
+#         self.reports = reports
+#
+#     def generate_output(self) -> Dict[str, Any]:
+#         scan_start_time = datetime.now(timezone.utc).isoformat()
+#
+#         in_toto_data = {
+#             "_type": "https://in-toto.io/Statement/v1",
+#             "subject": [
+#                 {
+#                     "name": "",
+#                     "digest": {
+#                         "sha256": "fe4fe40ac7250263c5dbe1cf3138912f3f416140aa248637a60d65fe22c47da4"
+#                     }
+#                 }
+#             ],
+#             "predicateType": "https://in-toto.io/attestation/vulns/v0.1",
+#             "predicate": {
+#                 "invocation": {
+#                     "parameters": [],
+#                     "uri": "",
+#                     "event_id": "",
+#                     "builder.id": ""
+#                 },
+#                 "scanner": {
+#                     "uri": "",
+#                     "version": "",
+#                     "db": {
+#                         "uri": "",
+#                         "version": "",
+#                         "lastUpdate": ""
+#                     },
+#                     "result": [
+#                         {
+#                             "id": "",
+#                             "severity": [
+#                                 {
+#                                     "method": "nvd",
+#                                     "score": ""
+#                                 },
+#                                 {
+#                                     "method": "cvss_score",
+#                                     "score": ""
+#                                 }
+#                             ],
+#                             "annotations": [],
+#                             "scanStartedOn": ""
+#                         }
+#                     ]
+#                 },
+#                 "metadata": {
+#                     "scanStartedOn": scan_start_time,
+#                     "scanFinishedOn": ""
+#                 }
+#             }
+#         }
+#
+#         for report in self.reports:
+#             for check in report.failed_checks:
+#                 in_toto_data["predicate"]["invocation"]["uri"] = "https://github.com/developer-guy/alpine/actions/runs/1071875574"
+#                 in_toto_data["predicate"]["invocation"]["event_id"] = "1071875574"
+#                 in_toto_data["predicate"]["invocation"]["builder.id"] = "GitHub Actions"
+#                 in_toto_data["predicate"]["scanner"]["uri"] = "pkg:github/aquasecurity/trivy@244fd47e07d1004f0aed9"
+#                 in_toto_data["predicate"]["scanner"]["version"] = "0.19.2"
+#                 in_toto_data["predicate"]["scanner"]["db"]["uri"] = "pkg:github/aquasecurity/trivy-db/commit/4c76bb580b2736d67751410fa4ab66d2b6b9b27d"
+#                 in_toto_data["predicate"]["scanner"]["db"]["version"] = "v1-2021080612"
+#                 in_toto_data["predicate"]["scanner"]["db"]["lastUpdate"] = "2021-08-06T17:45:50.52Z"
+#                 in_toto_data["subject"][0]["name"] = os.path.basename(check.file_path)
+#
+#                 result_data = {
+#                     "id": check.check_id,
+#                     "severity": [
+#                         {
+#                             "method": "nvd",
+#                             "score": check.severity
+#                         }
+#                     ],
+#                     "annotations": [{"key": "description", "value": check.check_name}],
+#                     "scanStartedOn": scan_start_time
+#                 }
+#
+#                 in_toto_data["predicate"]["scanner"]["result"][0] = result_data
+#
+#         scan_finish_time = datetime.now(timezone.utc).isoformat()
+#         in_toto_data["predicate"]["metadata"]["scanFinishedOn"] = scan_finish_time
+#
+#         return in_toto_data
+#
+#     @staticmethod
+#     def write_output(output_path: str, in_toto_data: Dict[str, Any]) -> None:
+#         with open(output_path, "w") as f:
+#             json.dump(in_toto_data, f, indent=4)
 import json
 from datetime import datetime, timezone
 from typing import Any, Dict, List, Union
@@ -39,23 +140,7 @@ def generate_output(self) -> Dict[str, Any]:
                         "version": "",
                         "lastUpdate": ""
                     },
-                    "result": [
-                        {
-                            "id": "",
-                            "severity": [
-                                {
-                                    "method": "nvd",
-                                    "score": ""
-                                },
-                                {
-                                    "method": "cvss_score",
-                                    "score": ""
-                                }
-                            ],
-                            "annotations": [],
-                            "scanStartedOn": ""
-                        }
-                    ]
+                    "result": []
                 },
                 "metadata": {
                     "scanStartedOn": scan_start_time,
@@ -88,7 +173,7 @@ def generate_output(self) -> Dict[str, Any]:
                     "scanStartedOn": scan_start_time
                 }
 
-                in_toto_data["predicate"]["scanner"]["result"][0] = result_data
+                in_toto_data["predicate"]["scanner"]["result"].append(result_data)
 
         scan_finish_time = datetime.now(timezone.utc).isoformat()
         in_toto_data["predicate"]["metadata"]["scanFinishedOn"] = scan_finish_time
@@ -98,4 +183,4 @@ def generate_output(self) -> Dict[str, Any]:
     @staticmethod
     def write_output(output_path: str, in_toto_data: Dict[str, Any]) -> None:
         with open(output_path, "w") as f:
-            json.dump(in_toto_data, f, indent=4),
+            json.dump(in_toto_data, f, indent=4)

checkov/common/runners/runner_registry.py

@@ -553,7 +553,6 @@ def print_reports(
 
             properties = Report.create_test_suite_properties_block(config)
 
-
             if junit_reports:
                 test_suites = [
                     report.get_test_suite(properties=properties, use_bc_ids=config.output_bc_ids)
@@ -619,7 +618,6 @@ def print_reports(
             data_outputs["gitlab_sast"] = json.dumps(gl_sast.sast_json)
         if "spdx" in config.output:
 
-
             spdx = SPDX(repo_id=metadata_integration.bc_integration.repo_id, reports=spdx_reports)
             spdx_output = spdx.get_tag_value_output()
 
@@ -672,11 +670,9 @@ def print_reports(
     def _print_to_console(self, output_formats: dict[str, str], output_format: str, output: str, url: str | None = None, support_path: str | None = None) -> None:
         """Prints the output to console, if needed"""
 
-
         output_dest = output_formats[output_format]
         if output_dest == CONSOLE_OUTPUT:
             del output_formats[output_format]
-
             if platform.system() == 'Windows':
                 sys.stdout.buffer.write(output.encode("utf-8"))
             else:
@@ -692,12 +688,10 @@ def _print_to_console(self, output_formats: dict[str, str], output_format: str,
             if CONSOLE_OUTPUT in output_formats.values():
                 print(OUTPUT_DELIMITER)
 
-
     def print_iac_bom_reports(self, output_path: str,
                               scan_reports: list[Report],
                               output_types: list[str],
                               account_id: str) -> dict[str, str]:
-
         output_files = {
             'cyclonedx': 'results_cyclonedx.xml',
             'csv': 'results_iac.csv'
@@ -725,7 +719,6 @@ def print_iac_bom_reports(self, output_path: str,
 
         return {key: os.path.join(output_path, value) for key, value in output_files.items()}
 
-
     def filter_runner_framework(self) -> None:
         if not self.runner_filter:
             return
@@ -872,4 +865,4 @@ def _parallel_run(
 
     if runner.graph_manager:
         return report, runner.check_type, RunnerRegistry.extract_graphs_from_runner(runner), runner.resource_subgraph_map, log_stream
-    return report, runner.check_type, None, None, log_stream
+    return report, runner.check_type, None, None, log_stream

tests/common/output/test_in_toto_output.py

@@ -0,0 +1,74 @@
+import pytest
+from unittest.mock import Mock
+import json
+from checkov.common.output.in_toto_output import InTotoOutput
+
+
+@pytest.fixture
+def mock_report():
+    report = Mock()
+    report.failed_checks = [
+        Mock(
+            file_path="/path/to/main.tf",
+            check_id="CKV_AWS_6",
+            check_name="Ensure that S3 bucket has a Public Access block",
+            severity="HIGH",
+        ),
+        Mock(
+            file_path="/path/to/main.tf",
+            check_id="CKV_AWS_18",
+            check_name="Ensure the S3 bucket has access logging enabled",
+            severity="MEDIUM",
+        ),
+    ]
+    return report
+
+
+def test_generate_output(mock_report):
+    in_toto_output = InTotoOutput(repo_id="my_repo", reports=[mock_report])
+    result = in_toto_output.generate_output()
+
+    assert result["_type"] == "https://in-toto.io/Statement/v1"
+    assert result["predicateType"] == "https://in-toto.io/attestation/vulns/v0.1"
+    assert result["subject"][0]["digest"][
+               "sha256"] == "fe4fe40ac7250263c5dbe1cf3138912f3f416140aa248637a60d65fe22c47da4"
+    assert result["predicate"]["invocation"]["uri"] == "https://github.com/developer-guy/alpine/actions/runs/1071875574"
+    assert result["predicate"]["invocation"]["event_id"] == "1071875574"
+    assert result["predicate"]["invocation"]["builder.id"] == "GitHub Actions"
+    assert result["predicate"]["scanner"]["uri"] == "pkg:github/aquasecurity/trivy@244fd47e07d1004f0aed9"
+    assert result["predicate"]["scanner"]["version"] == "0.19.2"
+    assert result["predicate"]["scanner"]["db"][
+               "uri"] == "pkg:github/aquasecurity/trivy-db/commit/4c76bb580b2736d67751410fa4ab66d2b6b9b27d"
+    assert result["predicate"]["scanner"]["db"]["version"] == "v1-2021080612"
+    assert result["predicate"]["scanner"]["db"]["lastUpdate"] == "2021-08-06T17:45:50.52Z"
+    assert result["predicate"]["scanner"]["result"][0]["id"] == "CKV_AWS_6"
+    assert result["predicate"]["scanner"]["result"][0]["severity"][0]["score"] == "HIGH"
+    assert result["predicate"]["scanner"]["result"][1]["id"] == "CKV_AWS_18"
+    assert result["predicate"]["scanner"]["result"][1]["severity"][0]["score"] == "MEDIUM"
+
+
+def test_write_output(mock_report, tmpdir):
+    in_toto_output = InTotoOutput(repo_id="my_repo", reports=[mock_report])
+    output_path = tmpdir.join("test_output.json")
+    result = in_toto_output.generate_output()
+    in_toto_output.write_output(str(output_path), result)
+
+    with open(output_path, "r") as f:
+        data = json.load(f)
+
+    assert data["_type"] == "https://in-toto.io/Statement/v1"
+    assert data["predicateType"] == "https://in-toto.io/attestation/vulns/v0.1"
+    assert data["subject"][0]["digest"]["sha256"] == "fe4fe40ac7250263c5dbe1cf3138912f3f416140aa248637a60d65fe22c47da4"
+    assert data["predicate"]["invocation"]["uri"] == "https://github.com/developer-guy/alpine/actions/runs/1071875574"
+    assert data["predicate"]["invocation"]["event_id"] == "1071875574"
+    assert data["predicate"]["invocation"]["builder.id"] == "GitHub Actions"
+    assert data["predicate"]["scanner"]["uri"] == "pkg:github/aquasecurity/trivy@244fd47e07d1004f0aed9"
+    assert data["predicate"]["scanner"]["version"] == "0.19.2"
+    assert data["predicate"]["scanner"]["db"][
+               "uri"] == "pkg:github/aquasecurity/trivy-db/commit/4c76bb580b2736d67751410fa4ab66d2b6b9b27d"
+    assert data["predicate"]["scanner"]["db"]["version"] == "v1-2021080612"
+    assert data["predicate"]["scanner"]["db"]["lastUpdate"] == "2021-08-06T17:45:50.52Z"
+    assert data["predicate"]["scanner"]["result"][0]["id"] == "CKV_AWS_6"
+    assert data["predicate"]["scanner"]["result"][0]["severity"][0]["score"] == "HIGH"
+    assert data["predicate"]["scanner"]["result"][1]["id"] == "CKV_AWS_18"
+    assert data["predicate"]["scanner"]["result"][1]["severity"][0]["score"] == "MEDIUM"