fix(outputs): neutralise injection in CSV and HTML reports (#12963)

prowler/changelog.d/csv-formula-injection.security.md

@@ -0,0 +1 @@
+CSV outputs (findings, compliance and AWS quick inventory) prefix cells starting with `=`, `+`, `-`, `@`, tab or carriage return with a single quote so spreadsheet applications treat them as text instead of formulas; signed numbers such as `-1` are left untouched

prowler/changelog.d/html-compliance-escape.security.md

@@ -0,0 +1 @@
+HTML report compliance column is now escaped, so framework names and requirement IDs containing markup are shown as text instead of being rendered by the browser

prowler/lib/outputs/compliance/compliance_output.py

@@ -6,6 +6,7 @@ from prowler.lib.check.compliance_models import Compliance
 from prowler.lib.logger import logger
 from prowler.lib.outputs.finding import Finding
 from prowler.lib.outputs.output import Output
+from prowler.lib.outputs.utils import sanitize_csv_value
 
 
 class ComplianceOutput(Output):
@@ -83,7 +84,10 @@ class ComplianceOutput(Output):
                     csv_writer.writeheader()
                 for finding in self._data:
                     csv_writer.writerow(
-                        {k.upper(): v for k, v in finding.dict().items()}
+                        {
+                            k.upper(): sanitize_csv_value(v)
+                            for k, v in finding.dict().items()
+                        }
                     )
                 if self.close_file or self._from_cli:
                     self._file_descriptor.close()

prowler/lib/outputs/compliance/universal/universal_output.py

@@ -11,6 +11,7 @@ from prowler.lib.check.compliance_config_eval import (
 )
 from prowler.lib.check.compliance_models import ComplianceFramework
 from prowler.lib.logger import logger
+from prowler.lib.outputs.utils import sanitize_csv_value
 from prowler.lib.utils.utils import open_file
 
 if TYPE_CHECKING:
@@ -318,7 +319,12 @@ class UniversalComplianceOutput:
                 if self._file_descriptor.tell() == 0:
                     csv_writer.writeheader()
                 for row in self._data:
-                    csv_writer.writerow({k.upper(): v for k, v in row.dict().items()})
+                    csv_writer.writerow(
+                        {
+                            k.upper(): sanitize_csv_value(v)
+                            for k, v in row.dict().items()
+                        }
+                    )
                 if self.close_file or self._from_cli:
                     self._file_descriptor.close()
         except Exception as error:

prowler/lib/outputs/csv/csv.py

@@ -4,7 +4,7 @@ from typing import List
 from prowler.lib.logger import logger
 from prowler.lib.outputs.finding import Finding
 from prowler.lib.outputs.output import Output
-from prowler.lib.outputs.utils import unroll_dict, unroll_list
+from prowler.lib.outputs.utils import sanitize_csv_value, unroll_dict, unroll_list
 
 
 class CSV(Output):
@@ -106,7 +106,9 @@ class CSV(Output):
                 if self._file_descriptor.tell() == 0:
                     csv_writer.writeheader()
                 for finding in self._data:
-                    csv_writer.writerow(finding)
+                    csv_writer.writerow(
+                        {k: sanitize_csv_value(v) for k, v in finding.items()}
+                    )
                 if self.close_file or self._from_cli:
                     self._file_descriptor.close()
         except Exception as error:

prowler/lib/outputs/html/html.py

@@ -116,7 +116,7 @@ class HTML(Output):
                             <td>{str(escape(finding.status_extended)).replace("_", "<wbr />_")}</td>
                             <td><p class="show-read-more">{HTML.process_markdown(str(escape(finding.metadata.Risk)))}</p></td>
                             <td><p class="show-read-more">{HTML.process_markdown(str(escape(finding.metadata.Remediation.Recommendation.Text)))}</p> <a class="read-more" href="{str(escape(_safe_url(finding.metadata.Remediation.Recommendation.Url)))}"><i class="fas fa-external-link-alt"></i></a></td>
-                            <td><p class="show-read-more">{parse_html_string(unroll_dict(finding.compliance, separator=": "))}</p></td>
+                            <td><p class="show-read-more">{parse_html_string(str(escape(unroll_dict(finding.compliance, separator=": "))))}</p></td>
                         </tr>
                         """)
         except Exception as error:

prowler/lib/outputs/utils.py

@@ -1,3 +1,6 @@
+from math import isfinite
+
+
 def unroll_list(listed_items: list, separator: str = "|") -> str:
     """
     Unrolls a list of items into a single string, separated by a specified separator.
@@ -199,3 +202,21 @@ def parse_html_string(str: str) -> str:
             string += f"\n•{elem}\n"
 
     return string
+
+
+def sanitize_csv_value(value):
+    """Prefix a cell with `'` when a spreadsheet would evaluate it as a formula; numbers are left as they are."""
+    if not isinstance(value, str) or not (
+        value.startswith(("\t", "\r"))
+        or value.lstrip().startswith(("=", "+", "-", "@"))
+    ):
+        return value
+    try:
+        number = float(value)
+    except ValueError:
+        return f"'{value}"
+    # float() also accepts nan/inf and whitespace-padded forms, which a spreadsheet
+    # renders as text rather than as a number
+    if isfinite(number) and value == value.strip():
+        return value
+    return f"'{value}"

prowler/providers/aws/lib/quick_inventory/quick_inventory.py

@@ -14,6 +14,7 @@ from prowler.config.config import (
     output_file_timestamp,
 )
 from prowler.lib.logger import logger
+from prowler.lib.outputs.utils import sanitize_csv_value
 from prowler.providers.aws.aws_provider import AwsProvider
 from prowler.providers.aws.lib.arn.models import get_arn_resource_type
 
@@ -293,7 +294,7 @@ def create_output(resources: list, provider: AwsProvider, args):
                 header = data.keys()
                 csv_writer.writerow(header)
                 count += 1
-            csv_writer.writerow(data.values())
+            csv_writer.writerow([sanitize_csv_value(v) for v in data.values()])
 
         csv_file.close()
         print(

tests/lib/outputs/compliance/generic/generic_aws_test.py

@@ -136,6 +136,28 @@ class TestAWSGenericCompliance:
 
         assert content == expected_csv
 
+    def test_batch_write_neutralises_formula_initiators(self):
+        mock_file = StringIO()
+        findings = [
+            generate_finding_output(
+                status_extended="=cmd",
+                resource_uid="-1",
+                compliance={"NIST-800-53-Revision-4": "ac_2_4"},
+            )
+        ]
+        output = GenericCompliance(findings, NIST_800_53_REVISION_4_AWS)
+        output._file_descriptor = mock_file
+
+        with patch.object(mock_file, "close", return_value=None):
+            output.batch_write_data_to_file()
+
+        mock_file.seek(0)
+        header, row, _ = mock_file.read().splitlines()
+        cells = dict(zip(header.split(";"), row.split(";")))
+
+        assert cells["STATUSEXTENDED"] == "'=cmd"
+        assert cells["RESOURCEID"] == "-1"
+
     def test_csv_row_count_matches_framework_checks_not_stored_compliance(self):
         """Regression test for PROWLER-1763.
 

tests/lib/outputs/compliance/universal/universal_output_test.py

@@ -228,6 +228,34 @@ class TestCSVFileWrite:
         assert "REQUIREMENTS_ATTRIBUTES_SECTION" in content
         assert "IAM" in content
 
+    def test_batch_write_neutralises_formula_initiators(self, tmp_path):
+        reqs = [
+            UniversalComplianceRequirement(
+                id="1.1",
+                description="test",
+                attributes={"Section": "IAM"},
+                checks={"aws": ["check_a"]},
+            ),
+        ]
+        fw = _make_framework(reqs, [AttributeMetadata(key="Section", type="str")])
+
+        finding = _make_finding("check_a", "PASS", {"TestFW-1.0": ["1.1"]})
+        finding.status_extended = "=cmd"
+        finding.resource_name = "-1"
+        filepath = str(tmp_path / "test.csv")
+
+        output = UniversalComplianceOutput(
+            findings=[finding], framework=fw, file_path=filepath
+        )
+        output.batch_write_data_to_file()
+
+        with open(filepath, "r") as f:
+            header, row = f.read().splitlines()
+        cells = dict(zip(header.split(";"), row.split(";")))
+
+        assert cells["STATUSEXTENDED"] == "'=cmd"
+        assert cells["RESOURCENAME"] == "-1"
+
 
 class TestNoFindings:
     def test_empty_findings_no_data(self, tmp_path):

tests/lib/outputs/csv/csv_test.py

@@ -1,4 +1,5 @@
 import tempfile
+from csv import DictReader
 from datetime import datetime
 from io import StringIO, TextIOWrapper
 from typing import List
@@ -131,6 +132,29 @@ class TestCSV:
 
         assert content == expected_csv
 
+    def test_csv_write_neutralises_formula_initiators(self):
+        mock_file = StringIO()
+        findings = [
+            generate_finding_output(
+                resource_name='=HYPERLINK("http://attacker.example")',
+                resource_details="-1",
+                resource_tags={"=cmd": "x", "env": "prod"},
+            )
+        ]
+
+        output = CSV(findings)
+        output._file_descriptor = mock_file
+
+        with patch.object(mock_file, "close", return_value=None):
+            output.batch_write_data_to_file()
+
+        mock_file.seek(0)
+        cells = next(DictReader(mock_file, delimiter=";"))
+
+        assert cells["RESOURCE_NAME"] == '\'=HYPERLINK("http://attacker.example")'
+        assert cells["RESOURCE_DETAILS"] == "-1"
+        assert cells["RESOURCE_TAGS"] == "'=cmd=x | env=prod"
+
     def test_batch_write_data_to_file_without_findings(self):
         assert not CSV([])._file_descriptor
 

tests/lib/outputs/html/html_test.py

@@ -910,6 +910,23 @@ class TestHTML:
         assert "•key&<>'"<img" in output_data
         assert "=value&<>'"<img" in output_data
 
+    def test_transform_escapes_compliance(self):
+        xss_payload = '<img src=x onerror="window.PROWLER_COMPLIANCE_XSS=1">'
+        findings = [
+            generate_finding_output(
+                compliance={f"Framework{xss_payload}-1.0": [xss_payload, "1.1"]},
+            )
+        ]
+
+        output_data = HTML(findings).data[0]
+
+        assert xss_payload not in output_data
+        assert (
+            "&#x2022;Framework&lt;img src=x "
+            "onerror=&#34;window.PROWLER_COMPLIANCE_XSS=1&#34;&gt;-1.0: "
+            "&lt;img src=x onerror=&#34;window.PROWLER_COMPLIANCE_XSS=1&#34;&gt;, 1.1"
+        ) in output_data
+
     def test_transform_escapes_metadata_fields(self):
         finding = generate_finding_output()
         finding.metadata.Severity = MagicMock(value='<img data-field="severity" src=x>')

tests/lib/outputs/outputs_test.py

@@ -13,6 +13,7 @@ from prowler.lib.outputs.outputs import (
 from prowler.lib.outputs.utils import (
     parse_html_string,
     parse_json_tags,
+    sanitize_csv_value,
     unroll_dict,
     unroll_dict_to_list,
     unroll_list,
@@ -1313,3 +1314,51 @@ class TestReport:
         with mock.patch("builtins.print") as mocked_print:
             report([finding], provider, output_options)
             mocked_print.assert_called()
+
+
+class TestSanitizeCSVValue:
+    @pytest.mark.parametrize("initiator", ["=", "+", "-", "@", "\t", "\r"])
+    def test_leading_formula_initiator_is_prefixed(self, initiator):
+        assert sanitize_csv_value(f"{initiator}cmd") == f"'{initiator}cmd"
+
+    def test_value_containing_equals_is_untouched(self):
+        assert sanitize_csv_value("key=value") == "key=value"
+
+    @pytest.mark.parametrize("number", ["-1", "+1", "-1.5", "-1e3"])
+    def test_signed_number_is_untouched(self, number):
+        assert sanitize_csv_value(number) == number
+
+    @pytest.mark.parametrize("value", [None, True, -1, -1.5, ""])
+    def test_non_string_and_empty_values_are_untouched(self, value):
+        assert sanitize_csv_value(value) == value
+
+    @pytest.mark.parametrize(
+        "value", ["-nan", "+nan", "-inf", "+inf", "-infinity", "+Infinity"]
+    )
+    def test_non_finite_float_literal_is_prefixed(self, value):
+        assert sanitize_csv_value(value) == f"'{value}"
+
+    @pytest.mark.parametrize("value", ["-1 ", "+1\t", "-1\n"])
+    def test_padded_number_is_prefixed(self, value):
+        assert sanitize_csv_value(value) == f"'{value}"
+
+    @pytest.mark.parametrize(
+        "payload",
+        [
+            "=cmd|' /C calc'!A0",
+            "+cmd|' /C calc'!A0",
+            "-1+1|cmd|' /C calc'!A0",
+            "@SUM(1+1)*cmd|' /C calc'!A0",
+            "-0x1",
+        ],
+    )
+    def test_formula_payload_is_prefixed(self, payload):
+        assert sanitize_csv_value(payload) == f"'{payload}"
+
+    @pytest.mark.parametrize("value", [" =cmd", "\n=cmd", "  -1+1|cmd", " \n@SUM(1)"])
+    def test_whitespace_before_formula_initiator_is_prefixed(self, value):
+        assert sanitize_csv_value(value) == f"'{value}"
+
+    @pytest.mark.parametrize("value", [" ", "\n", "  text"])
+    def test_whitespace_without_initiator_is_untouched(self, value):
+        assert sanitize_csv_value(value) == value

tests/providers/aws/lib/quick_inventory/quick_inventory_test.py

@@ -0,0 +1,52 @@
+import csv
+from unittest.mock import MagicMock
+
+from prowler.providers.aws.lib.quick_inventory.quick_inventory import create_output
+
+
+def _args(tmp_path):
+    args = MagicMock()
+    args.output_directory = str(tmp_path)
+    args.output_filename = "inventory"
+    return args
+
+
+def _provider():
+    provider = MagicMock()
+    provider.identity.account = "123456789012"
+    return provider
+
+
+def _rows(tmp_path):
+    with open(tmp_path / "inventory.csv", newline="") as handle:
+        return list(csv.reader(handle))
+
+
+class TestQuickInventoryCsvOutput:
+    def test_formula_initiator_in_tags_is_neutralised(self, tmp_path):
+        resources = [
+            {
+                "arn": "arn:aws:s3:eu-west-1:123456789012:bucket-one",
+                "tags": '=cmd|" /C calc"!A0',
+            }
+        ]
+
+        create_output(resources, _provider(), _args(tmp_path))
+
+        header, row = _rows(tmp_path)
+        assert row[header.index("AWS_Tags")] == '\'=cmd|" /C calc"!A0'
+
+    def test_header_and_ordinary_values_are_untouched(self, tmp_path):
+        resources = [
+            {
+                "arn": "arn:aws:s3:eu-west-1:123456789012:bucket-one",
+                "tags": "env=prod",
+            }
+        ]
+
+        create_output(resources, _provider(), _args(tmp_path))
+
+        header, row = _rows(tmp_path)
+        assert header[0] == "AWS_AccountID"
+        assert row[header.index("AWS_Tags")] == "env=prod"
+        assert row[header.index("AWS_Region")] == "eu-west-1"