diff --git a/prowler/changelog.d/csv-formula-injection.security.md b/prowler/changelog.d/csv-formula-injection.security.md new file mode 100644 index 0000000000..c8fd6b9efd --- /dev/null +++ b/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 diff --git a/prowler/changelog.d/html-compliance-escape.security.md b/prowler/changelog.d/html-compliance-escape.security.md new file mode 100644 index 0000000000..98b4b12a04 --- /dev/null +++ b/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 diff --git a/prowler/lib/outputs/compliance/compliance_output.py b/prowler/lib/outputs/compliance/compliance_output.py index f4f84561f4..60fc5d6749 100644 --- a/prowler/lib/outputs/compliance/compliance_output.py +++ b/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() diff --git a/prowler/lib/outputs/compliance/universal/universal_output.py b/prowler/lib/outputs/compliance/universal/universal_output.py index 9c7c76da47..505e855b57 100644 --- a/prowler/lib/outputs/compliance/universal/universal_output.py +++ b/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: diff --git a/prowler/lib/outputs/csv/csv.py b/prowler/lib/outputs/csv/csv.py index bcb50d9433..1c10b20af7 100644 --- a/prowler/lib/outputs/csv/csv.py +++ b/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: diff --git a/prowler/lib/outputs/html/html.py b/prowler/lib/outputs/html/html.py index 60428bede7..da3faa8b1c 100644 --- a/prowler/lib/outputs/html/html.py +++ b/prowler/lib/outputs/html/html.py @@ -116,7 +116,7 @@ class HTML(Output): {str(escape(finding.status_extended)).replace("_", "_")}

{HTML.process_markdown(str(escape(finding.metadata.Risk)))}

{HTML.process_markdown(str(escape(finding.metadata.Remediation.Recommendation.Text)))}

-

{parse_html_string(unroll_dict(finding.compliance, separator=": "))}

+

{parse_html_string(str(escape(unroll_dict(finding.compliance, separator=": "))))}

""") except Exception as error: diff --git a/prowler/lib/outputs/utils.py b/prowler/lib/outputs/utils.py index 9750c4f7e2..de0ffce5c3 100644 --- a/prowler/lib/outputs/utils.py +++ b/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}" diff --git a/prowler/providers/aws/lib/quick_inventory/quick_inventory.py b/prowler/providers/aws/lib/quick_inventory/quick_inventory.py index 8ffdd80e7b..07327aac76 100644 --- a/prowler/providers/aws/lib/quick_inventory/quick_inventory.py +++ b/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( diff --git a/tests/lib/outputs/compliance/generic/generic_aws_test.py b/tests/lib/outputs/compliance/generic/generic_aws_test.py index 335a9ad17d..36ac977715 100644 --- a/tests/lib/outputs/compliance/generic/generic_aws_test.py +++ b/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. diff --git a/tests/lib/outputs/compliance/universal/universal_output_test.py b/tests/lib/outputs/compliance/universal/universal_output_test.py index dd951cf272..00f577060a 100644 --- a/tests/lib/outputs/compliance/universal/universal_output_test.py +++ b/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): diff --git a/tests/lib/outputs/csv/csv_test.py b/tests/lib/outputs/csv/csv_test.py index b2b5d390df..dd132ad593 100644 --- a/tests/lib/outputs/csv/csv_test.py +++ b/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 diff --git a/tests/lib/outputs/html/html_test.py b/tests/lib/outputs/html/html_test.py index 1590d29742..6c39e43249 100644 --- a/tests/lib/outputs/html/html_test.py +++ b/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 = '' + 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 ( + "•Framework<img src=x " + "onerror="window.PROWLER_COMPLIANCE_XSS=1">-1.0: " + "<img src=x onerror="window.PROWLER_COMPLIANCE_XSS=1">, 1.1" + ) in output_data + def test_transform_escapes_metadata_fields(self): finding = generate_finding_output() finding.metadata.Severity = MagicMock(value='') diff --git a/tests/lib/outputs/outputs_test.py b/tests/lib/outputs/outputs_test.py index fad6f4d286..b404e1c032 100644 --- a/tests/lib/outputs/outputs_test.py +++ b/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 diff --git a/tests/providers/aws/lib/quick_inventory/quick_inventory_test.py b/tests/providers/aws/lib/quick_inventory/quick_inventory_test.py new file mode 100644 index 0000000000..52375c67b8 --- /dev/null +++ b/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"