ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 20:42:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(tags): convert tags to a dictionary (#4598)

Browse Source 
Co-authored-by: Pepe Fagoaga <pepe@prowler.com>
This commit is contained in:
Sergio Garcia
2024-08-05 02:17:43 -04:00
committed by GitHub
parent b0709d08cd
commit 95ec3d91b4
24 changed files with 357 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/lib/outputs/asff/asff.py
+7 -40
View File
@@ -25,7 +25,6 @@ class ASFF(Output):
- transform(findings: list[Finding]) -> None: Transforms a list of findings into ASFF format.
- batch_write_data_to_file() -> None: Writes the findings data to a file in JSON ASFF format.
- generate_status(status: str, muted: bool = False) -> str: Generates the ASFF status based on the provided status and muted flag.
- format_resource_tags(tags: str) -> dict: Transforms a string of tags into a dictionary format.
References:
- AWS Security Hub API Reference: https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_Compliance.html
@@ -62,7 +61,6 @@ class ASFF(Output):
if finding.status == "MANUAL":
continue
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
resource_tags = ASFF.format_resource_tags(finding.resource_tags)
associated_standards, compliance_summary = ASFF.format_compliance(
finding.compliance
@@ -70,7 +68,6 @@ class ASFF(Output):
# Ensures finding_status matches allowed values in ASFF
finding_status = ASFF.generate_status(finding.status, finding.muted)
self._data.append(
AWSSecurityFindingFormat(
# The following line cannot be changed because it is the format we use to generate unique findings for AWS Security Hub
@@ -99,7 +96,7 @@ class ASFF(Output):
Type=finding.resource_type,
Partition=finding.partition,
Region=finding.region,
Tags=resource_tags,
Tags=finding.resource_tags,
)
],
Compliance=Compliance(
@@ -195,42 +192,6 @@ class ASFF(Output):
return json_asff_status
@staticmethod
def format_resource_tags(tags: str) -> dict:
"""
Transforms a string of tags into a dictionary format.
Parameters:
- tags (str): A string containing tags separated by ' | ' and key-value pairs separated by '='.
Returns:
- dict: A dictionary where keys are tag names and values are tag values.
Notes:
- If the input string is empty or None, it returns None.
- Each tag in the input string should be in the format 'key=value'.
- If the input string is not formatted correctly, it logs an error and returns None.
"""
try:
tags_dict = None
if tags:
tags = tags.split(" | ")
tags_dict = {}
for tag in tags:
value = tag.split("=")
tags_dict[value[0]] = value[1]
return tags_dict
except IndexError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return None
except AttributeError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return None
@staticmethod
def format_compliance(compliance: dict) -> tuple[list[dict], list[str]]:
"""
@@ -316,6 +277,12 @@ class Resource(BaseModel):
Region: str
Tags: Optional[dict]
@validator("Tags", pre=True, always=True)
def tags_cannot_be_empty_dict(tags):
if not tags:
return None
return tags
class Compliance(BaseModel):
"""
prowler/lib/outputs/csv/csv.py
+8 -3
View File
@@ -3,7 +3,7 @@ from csv import DictWriter
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 unroll_dict
class CSV(Output):
@@ -17,8 +17,13 @@ class CSV(Output):
try:
for finding in findings:
finding_dict = {k.upper(): v for k, v in finding.dict().items()}
finding_dict["COMPLIANCE"] = unroll_dict(finding.compliance)
finding_dict["ACCOUNT_TAGS"] = unroll_list(finding.account_tags)
finding_dict["RESOURCE_TAGS"] = unroll_dict(finding.resource_tags)
finding_dict["COMPLIANCE"] = unroll_dict(
finding.compliance, separator=": "
)
finding_dict["ACCOUNT_TAGS"] = unroll_dict(
finding.account_tags, separator=":"
)
finding_dict["STATUS"] = finding.status.value
finding_dict["SEVERITY"] = finding.severity.value
self._data.append(finding_dict)
prowler/lib/outputs/finding.py
+2 -2
View File
@@ -50,7 +50,7 @@ class Finding(BaseModel):
# Optional since it depends on permissions
account_organization_name: Optional[str]
# Optional since it depends on permissions
account_tags: Optional[list[str]]
account_tags: dict = {}
finding_uid: str
provider: str
check_id: str
@@ -66,7 +66,7 @@ class Finding(BaseModel):
resource_uid: str
resource_name: str
resource_details: str
resource_tags: str
resource_tags: dict = {}
# Only present for AWS and Azure
partition: Optional[str]
region: str
prowler/lib/outputs/html/html.py
+2 -2
View File
@@ -45,11 +45,11 @@ class HTML(Output):
<td>{finding.check_id.replace("_", "<wbr />_")}</td>
<td>{finding.check_title}</td>
<td>{finding.resource_uid.replace("<", "&lt;").replace(">", "&gt;").replace("_", "<wbr />_")}</td>
<td>{parse_html_string(finding.resource_tags)}</td>
<td>{parse_html_string(unroll_dict(finding.resource_tags))}</td>
<td>{finding.status_extended.replace("<", "&lt;").replace(">", "&gt;").replace("_", "<wbr />_")}</td>
<td><p class="show-read-more">{html.escape(finding.risk)}</p></td>
<td><p class="show-read-more">{html.escape(finding.remediation_recommendation_text)}</p> <a class="read-more" href="{finding.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))}</p></td>
<td><p class="show-read-more">{parse_html_string(unroll_dict(finding.compliance, separator=": "))}</p></td>
</tr>
"""
)
prowler/lib/outputs/ocsf/ocsf.py
+3 -7
View File
@@ -20,6 +20,7 @@ from py_ocsf_models.objects.resource_details import ResourceDetails
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_to_list
class OCSF(Output):
@@ -97,12 +98,7 @@ class OCSF(Output):
risk_details=finding.risk,
resources=[
ResourceDetails(
# TODO: Check labels for other providers
labels=(
finding.resource_tags.split(",")
if finding.resource_tags
else []
),
labels=unroll_dict_to_list(finding.resource_tags),
name=finding.resource_name,
uid=finding.resource_uid,
group=Group(name=finding.service_name),
@@ -148,7 +144,7 @@ class OCSF(Output):
type_id=cloud_account_type.value,
type=cloud_account_type.name,
uid=finding.account_uid,
labels=finding.account_tags,
labels=unroll_dict_to_list(finding.account_tags),
),
org=Organization(
uid=finding.account_organization_uid,
prowler/lib/outputs/utils.py
+128 -44
View File
@@ -1,4 +1,24 @@
def unroll_list(listed_items: list, separator: str = "|"):
def unroll_list(listed_items: list, separator: str = "|") -> str:
"""
Unrolls a list of items into a single string, separated by a specified separator.
Args:
listed_items (list): The list of items to be unrolled.
separator (str, optional): The separator to be used between the items. Defaults to "|".
Returns:
str: The unrolled string.
Examples:
>>> unroll_list(['apple', 'banana', 'orange'])
'apple | banana | orange'
>>> unroll_list(['apple', 'banana', 'orange'], separator=',')
'apple, banana, orange'
>>> unroll_list([])
''
"""
unrolled_items = ""
if listed_items:
for item in listed_items:
@@ -13,70 +33,118 @@ def unroll_list(listed_items: list, separator: str = "|"):
return unrolled_items
def unroll_tags(tags: list):
unrolled_items = ""
separator = "|"
def unroll_tags(tags: list) -> dict:
"""
Unrolls a list of tags into a dictionary.
Args:
tags (list): A list of tags.
Returns:
dict: A dictionary containing the unrolled tags.
Examples:
>>> tags = [{"key": "name", "value": "John"}, {"key": "age", "value": "30"}]
>>> unroll_tags(tags)
{'name': 'John', 'age': '30'}
>>> tags = [{"Key": "name", "Value": "John"}, {"Key": "age", "Value": "30"}]
>>> unroll_tags(tags)
{'name': 'John', 'age': '30'}
>>> tags = [{"name": "John", "age": "30"}]
>>> unroll_tags(tags)
{'name': 'John', 'age': '30'}
>>> tags = []
>>> unroll_tags(tags)
{}
"""
if tags and tags != [{}] and tags != [None]:
for item in tags:
# Check if there are tags in list
if isinstance(item, dict):
for key, value in item.items():
if not unrolled_items:
# Check the pattern of tags (Key:Value or Key:key/Value:value)
if "Key" != key and "Value" != key:
unrolled_items = f"{key}={value}"
else:
if "Key" == key:
unrolled_items = f"{value}="
else:
unrolled_items = f"{value}"
else:
if "Key" != key and "Value" != key:
unrolled_items = (
f"{unrolled_items} {separator} {key}={value}"
)
else:
if "Key" == key:
unrolled_items = (
f"{unrolled_items} {separator} {value}="
)
else:
unrolled_items = f"{unrolled_items}{value}"
elif not unrolled_items:
unrolled_items = f"{item}"
else:
unrolled_items = f"{unrolled_items} {separator} {item}"
return unrolled_items
if "key" in tags[0]:
return {item["key"]: item["value"] for item in tags}
elif "Key" in tags[0]:
return {item["Key"]: item["Value"] for item in tags}
else:
return {key: value for d in tags for key, value in d.items()}
return {}
def unroll_dict(dict: dict):
def unroll_dict(dict: dict, separator: str = "=") -> str:
"""
Unrolls a dictionary into a string representation.
Args:
dict (dict): The dictionary to be unrolled.
Returns:
str: The unrolled string representation of the dictionary.
Examples:
>>> my_dict = {'name': 'John', 'age': 30, 'hobbies': ['reading', 'coding']}
>>> unroll_dict(my_dict)
'name: John | age: 30 | hobbies: reading, coding'
"""
unrolled_items = ""
separator = "|"
for key, value in dict.items():
if isinstance(value, list):
value = ", ".join(value)
if not unrolled_items:
unrolled_items = f"{key}: {value}"
unrolled_items = f"{key}{separator}{value}"
else:
unrolled_items = f"{unrolled_items} {separator} {key}: {value}"
unrolled_items = f"{unrolled_items} | {key}{separator}{value}"
return unrolled_items
def unroll_dict_to_list(dict: dict):
def unroll_dict_to_list(dict: dict) -> list:
"""
Unrolls a dictionary into a list of key-value pairs.
Args:
dict (dict): The dictionary to be unrolled.
Returns:
list: A list of key-value pairs, where each pair is represented as a string.
Examples:
>>> my_dict = {'name': 'John', 'age': 30, 'hobbies': ['reading', 'coding']}
>>> unroll_dict_to_list(my_dict)
['name: John', 'age: 30', 'hobbies: reading, coding']
"""
dict_list = []
for key, value in dict.items():
if isinstance(value, list):
value = ", ".join(value)
dict_list.append(f"{key}: {value}")
dict_list.append(f"{key}:{value}")
else:
dict_list.append(f"{key}: {value}")
dict_list.append(f"{key}:{value}")
return dict_list
def parse_json_tags(tags: list):
def parse_json_tags(tags: list) -> dict[str, str]:
"""
Parses a list of JSON tags and returns a dictionary of key-value pairs.
Args:
tags (list): A list of JSON tags.
Returns:
dict: A dictionary containing the parsed key-value pairs from the tags.
Examples:
>>> tags = [
... {"Key": "Name", "Value": "John"},
... {"Key": "Age", "Value": "30"},
... {"Key": "City", "Value": "New York"}
... ]
>>> parse_json_tags(tags)
{'Name': 'John', 'Age': '30', 'City': 'New York'}
"""
dict_tags = {}
if tags and tags != [{}] and tags != [None]:
for tag in tags:
@@ -88,7 +156,23 @@ def parse_json_tags(tags: list):
return dict_tags
def parse_html_string(str: str):
def parse_html_string(str: str) -> str:
"""
Parses a string and returns a formatted HTML string.
This function takes an input string and splits it using the delimiter " | ".
It then formats each element of the split string as a bullet point in HTML format.
Args:
str (str): The input string to be parsed.
Returns:
str: The formatted HTML string.
Example:
>>> parse_html_string("item1 | item2 | item3")
'\n&#x2022;item1\n\n&#x2022;item2\n\n&#x2022;item3\n'
"""
string = ""
for elem in str.split(" | "):
if elem:
prowler/providers/aws/lib/mutelist/mutelist.py
+4 -4
View File
@@ -1,13 +1,13 @@
import re
from typing import Any
import yaml
from boto3 import Session
from boto3.dynamodb.conditions import Attr
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.lib.mutelist.mutelist import Mutelist
from prowler.lib.outputs.utils import unroll_tags
from prowler.lib.outputs.utils import unroll_dict, unroll_tags
class AWSMutelist(Mutelist):
@@ -45,7 +45,7 @@ class AWSMutelist(Mutelist):
def is_finding_muted(
self,
finding: Any,
finding: Check_Report_AWS,
aws_account_id: str,
) -> bool:
return self.is_muted(
@@ -53,7 +53,7 @@ class AWSMutelist(Mutelist):
finding.check_metadata.CheckID,
finding.region,
finding.resource_id,
unroll_tags(finding.resource_tags),
unroll_dict(unroll_tags(finding.resource_tags)),
)
def get_mutelist_file_from_s3(self, aws_session: Session = None):
prowler/providers/aws/lib/organizations/organizations.py
+2 -2
View File
@@ -30,9 +30,9 @@ def get_organizations_metadata(
def parse_organizations_metadata(metadata: dict, tags: dict) -> AWSOrganizationsInfo:
try:
# Convert Tags dictionary to String
account_details_tags = []
account_details_tags = {}
for tag in tags.get("Tags", {}):
account_details_tags.append(f"{tag['Key']}:{tag['Value']}")
account_details_tags[tag["Key"]] = tag["Value"]
account_details = metadata.get("Account", {})
prowler/providers/azure/lib/mutelist/mutelist.py
+4 -5
View File
@@ -1,18 +1,17 @@
from typing import Any
from prowler.lib.check.models import Check_Report_Azure
from prowler.lib.mutelist.mutelist import Mutelist
from prowler.lib.outputs.utils import unroll_tags
from prowler.lib.outputs.utils import unroll_dict, unroll_tags
class AzureMutelist(Mutelist):
def is_finding_muted(
self,
finding: Any,
finding: Check_Report_Azure,
) -> bool:
return self.is_muted(
finding.subscription,
finding.check_metadata.CheckID,
finding.location,
finding.resource_name,
unroll_tags(finding.resource_tags),
unroll_dict(unroll_tags(finding.resource_tags)),
)
prowler/providers/gcp/gcp_provider.py
+2 -2
View File
@@ -279,9 +279,9 @@ class GcpProvider(Provider):
response = request.execute()
for project in response.get("projects", []):
labels = []
labels = {}
for key, value in project.get("labels", {}).items():
labels.append(f"{key}:{value}")
labels[key] = value
project_id = project["projectId"]
gcp_project = GCPProject(
prowler/providers/gcp/lib/mutelist/mutelist.py
+4 -5
View File
@@ -1,18 +1,17 @@
from typing import Any
from prowler.lib.check.models import Check_Report_GCP
from prowler.lib.mutelist.mutelist import Mutelist
from prowler.lib.outputs.utils import unroll_tags
from prowler.lib.outputs.utils import unroll_dict, unroll_tags
class GCPMutelist(Mutelist):
def is_finding_muted(
self,
finding: Any,
finding: Check_Report_GCP,
) -> bool:
return self.is_muted(
finding.project_id,
finding.check_metadata.CheckID,
finding.location,
finding.resource_name,
unroll_tags(finding.resource_tags),
unroll_dict(unroll_tags(finding.resource_tags)),
)
prowler/providers/gcp/models.py
+1 -1
View File
@@ -22,7 +22,7 @@ class GCPProject(BaseModel):
id: str
name: str
organization: Optional[GCPOrganization]
labels: list[str]
labels: dict
lifecycle_state: str
prowler/providers/kubernetes/lib/mutelist/mutelist.py
+4 -5
View File
@@ -1,13 +1,12 @@
from typing import Any
from prowler.lib.check.models import Check_Report_Kubernetes
from prowler.lib.mutelist.mutelist import Mutelist
from prowler.lib.outputs.utils import unroll_tags
from prowler.lib.outputs.utils import unroll_dict, unroll_tags
class KubernetesMutelist(Mutelist):
def is_finding_muted(
self,
finding: Any,
finding: Check_Report_Kubernetes,
cluster: str,
) -> bool:
return self.is_muted(
@@ -15,5 +14,5 @@ class KubernetesMutelist(Mutelist):
finding.check_metadata.CheckID,
finding.namespace,
finding.resource_name,
unroll_tags(finding.resource_tags),
unroll_dict(unroll_tags(finding.resource_tags)),
)
tests/lib/outputs/asff/asff_test.py
+74 -20
View File
@@ -40,7 +40,7 @@ class TestASFF:
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags="key1=value1",
resource_tags={"key1": "value1"},
)
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
@@ -70,7 +70,7 @@ class TestASFF:
Type=finding.resource_type,
Partition=AWS_COMMERCIAL_PARTITION,
Region=AWS_REGION_EU_WEST_1,
Tags=ASFF.format_resource_tags(finding.resource_tags),
Tags={"key1": "value1"},
)
],
Compliance=Compliance(
@@ -103,7 +103,7 @@ class TestASFF:
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags="key1=value1",
resource_tags={"key1": "value1"},
)
finding.remediation_recommendation_url = ""
@@ -136,7 +136,72 @@ class TestASFF:
Type=finding.resource_type,
Partition=AWS_COMMERCIAL_PARTITION,
Region=AWS_REGION_EU_WEST_1,
Tags=ASFF.format_resource_tags(finding.resource_tags),
Tags={"key1": "value1"},
)
],
Compliance=Compliance(
Status=ASFF.generate_status(status),
RelatedRequirements=compliance_summary,
AssociatedStandards=associated_standards,
),
Remediation=Remediation(
Recommendation=Recommendation(
Text=finding.remediation_recommendation_text,
Url="https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html",
)
),
Description=finding.description,
)
asff = ASFF(findings=[finding])
assert len(asff.data) == 1
asff_finding = asff.data[0]
assert asff_finding == expected
def test_asff_without_resource_tags(self):
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
)
finding.remediation_recommendation_url = ""
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
associated_standards, compliance_summary = ASFF.format_compliance(
finding.compliance
)
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
expected = AWSSecurityFindingFormat(
Id=f"prowler-{finding.check_id}-{AWS_ACCOUNT_NUMBER}-{AWS_REGION_EU_WEST_1}-{hash_sha512(finding.resource_uid)}",
ProductArn=f"arn:{AWS_COMMERCIAL_PARTITION}:securityhub:{AWS_REGION_EU_WEST_1}::product/prowler/prowler",
ProductFields=ProductFields(
ProviderVersion=prowler_version,
ProwlerResourceName=finding.resource_uid,
),
GeneratorId="prowler-" + finding.check_id,
AwsAccountId=AWS_ACCOUNT_NUMBER,
Types=finding.check_type.split(","),
FirstObservedAt=timestamp,
UpdatedAt=timestamp,
CreatedAt=timestamp,
Severity=Severity(Label=finding.severity),
Title=finding.check_title,
Resources=[
Resource(
Id=finding.resource_uid,
Type=finding.resource_type,
Partition=AWS_COMMERCIAL_PARTITION,
Region=AWS_REGION_EU_WEST_1,
Tags=None,
)
],
Compliance=Compliance(
@@ -171,7 +236,7 @@ class TestASFF:
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags="key1=value1",
resource_tags={"key1": "value1"},
)
finding.remediation_recommendation_url = ""
finding.remediation_recommendation_text = "x" * 513
@@ -205,7 +270,7 @@ class TestASFF:
Type=finding.resource_type,
Partition=AWS_COMMERCIAL_PARTITION,
Region=AWS_REGION_EU_WEST_1,
Tags=ASFF.format_resource_tags(finding.resource_tags),
Tags={"key1": "value1"},
)
],
Compliance=Compliance(
@@ -239,7 +304,7 @@ class TestASFF:
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags="key1=value1",
resource_tags={"key1": "value1"},
compliance={
"CISA": ["your-systems-3", "your-data-2"],
"SOC2": ["cc_2_1", "cc_7_2", "cc_a_1_2"],
@@ -412,7 +477,7 @@ class TestASFF:
Type=finding.resource_type,
Partition=AWS_COMMERCIAL_PARTITION,
Region=AWS_REGION_EU_WEST_1,
Tags=ASFF.format_resource_tags(finding.resource_tags),
Tags={"key1": "value1"},
)
],
Compliance=Compliance(
@@ -448,7 +513,7 @@ class TestASFF:
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags="key1=value1",
resource_tags={"key1": "value1"},
)
finding.remediation_recommendation_url = ""
@@ -517,14 +582,3 @@ class TestASFF:
assert ASFF.generate_status("FAIL") == "FAILED"
assert ASFF.generate_status("FAIL", True) == "WARNING"
assert ASFF.generate_status("SOMETHING ELSE") == "NOT_AVAILABLE"
def test_asff_format_resource_tags(self):
assert ASFF.format_resource_tags(None) is None
assert ASFF.format_resource_tags("") is None
assert ASFF.format_resource_tags([]) is None
assert ASFF.format_resource_tags([{}]) is None
assert ASFF.format_resource_tags("key1=value1") == {"key1": "value1"}
assert ASFF.format_resource_tags("key1=value1 | key2=value2") == {
"key1": "value1",
"key2": "value2",
}
tests/lib/outputs/csv/csv_test.py
+2 -2
View File
@@ -25,7 +25,7 @@ class TestCSV:
resource_uid="resource-123",
resource_name="Example Resource",
resource_details="Detailed information about the resource",
resource_tags="tag1,tag2",
resource_tags={"tag1": "value1", "tag2": "value2"},
partition="aws",
description="Description of the finding",
risk="High",
@@ -78,7 +78,7 @@ class TestCSV:
assert (
output_data["RESOURCE_DETAILS"] == "Detailed information about the resource"
)
assert output_data["RESOURCE_TAGS"] == "tag1,tag2"
assert output_data["RESOURCE_TAGS"] == "tag1=value1 | tag2=value2"
assert output_data["PARTITION"] == "aws"
assert output_data["REGION"] == AWS_REGION_EU_WEST_1
assert output_data["DESCRIPTION"] == "Description of the finding"
tests/lib/outputs/finding_test.py
+16 -16
View File
@@ -12,7 +12,7 @@ def mock_get_provider_data_mapping_aws(_):
"account_email": "mock_account_email",
"account_organization_uid": "mock_account_org_uid",
"account_organization_name": "mock_account_org_name",
"account_tags": ["tag1", "tag2"],
"account_tags": {"tag1": "value1"},
"finding_uid": "mock_finding_uid",
"provider": "aws",
"check_id": "mock_check_id",
@@ -28,7 +28,7 @@ def mock_get_provider_data_mapping_aws(_):
"resource_uid": "mock_resource_uid",
"resource_name": "mock_resource_name",
"resource_details": "mock_resource_details",
"resource_tags": "mock_resource_tags",
"resource_tags": {"tag1": "value1"},
"partition": None,
"region": "mock_region",
"description": "mock_description",
@@ -58,7 +58,7 @@ def mock_get_provider_data_mapping_azure(_):
"account_email": "mock_account_email",
"account_organization_uid": "mock_account_org_uid",
"account_organization_name": "mock_account_org_name",
"account_tags": ["tag1", "tag2"],
"account_tags": {"tag1": "value1"},
"finding_uid": "mock_finding_uid",
"provider": "azure",
"check_id": "mock_check_id",
@@ -74,7 +74,7 @@ def mock_get_provider_data_mapping_azure(_):
"resource_uid": "mock_resource_uid",
"resource_name": "mock_resource_name",
"resource_details": "mock_resource_details",
"resource_tags": "mock_resource_tags",
"resource_tags": {"tag1": "value1"},
"partition": None,
"description": "mock_description",
"risk": "mock_risk",
@@ -103,7 +103,7 @@ def mock_get_provider_data_mapping_gcp(_):
"account_email": "mock_account_email",
"account_organization_uid": "mock_account_org_uid",
"account_organization_name": "mock_account_org_name",
"account_tags": ["tag1", "tag2"],
"account_tags": {"tag1": "value1"},
"finding_uid": "mock_finding_uid",
"provider": "gcp",
"check_id": "mock_check_id",
@@ -119,7 +119,7 @@ def mock_get_provider_data_mapping_gcp(_):
"resource_uid": "mock_resource_uid",
"resource_name": "mock_resource_name",
"resource_details": "mock_resource_details",
"resource_tags": "mock_resource_tags",
"resource_tags": {"tag1": "value1"},
"partition": None,
"description": "mock_description",
"risk": "mock_risk",
@@ -148,7 +148,7 @@ def mock_get_provider_data_mapping_kubernetes(_):
"account_email": "mock_account_email",
"account_organization_uid": "mock_account_org_uid",
"account_organization_name": "mock_account_org_name",
"account_tags": ["tag1", "tag2"],
"account_tags": {"tag1": "value1"},
"finding_uid": "mock_finding_uid",
"provider": "kubernetes",
"check_id": "mock_check_id",
@@ -164,7 +164,7 @@ def mock_get_provider_data_mapping_kubernetes(_):
"resource_uid": "mock_resource_uid",
"resource_name": "mock_resource_name",
"resource_details": "mock_resource_details",
"resource_tags": "mock_resource_tags",
"resource_tags": {"tag1": "value1"},
"partition": None,
"description": "mock_description",
"risk": "mock_risk",
@@ -240,7 +240,7 @@ class TestFinding:
assert finding_output.subservice_name == "mock_subservice_name"
assert finding_output.severity == Severity.high
assert finding_output.resource_type == "mock_resource_type"
assert finding_output.resource_tags == "mock_resource_tags"
assert finding_output.resource_tags == {"tag1": "value1"}
assert finding_output.partition is None
assert finding_output.description == "mock_description"
assert finding_output.risk == "mock_risk"
@@ -260,7 +260,7 @@ class TestFinding:
assert finding_output.account_email == "mock_account_email"
assert finding_output.account_organization_uid == "mock_account_org_uid"
assert finding_output.account_organization_name == "mock_account_org_name"
assert finding_output.account_tags == ["tag1", "tag2"]
assert finding_output.account_tags == {"tag1": "value1"}
assert finding_output.prowler_version == "1.0.0"
@patch(
@@ -318,7 +318,7 @@ class TestFinding:
assert finding_output.subservice_name == "mock_subservice_name"
assert finding_output.severity == Severity.high
assert finding_output.resource_type == "mock_resource_type"
assert finding_output.resource_tags == "mock_resource_tags"
assert finding_output.resource_tags == {"tag1": "value1"}
assert finding_output.partition is None
assert finding_output.description == "mock_description"
assert finding_output.risk == "mock_risk"
@@ -353,7 +353,7 @@ class TestFinding:
organization.display_name = "mock_organization_name"
project.id = "mock_project_id"
project.name = "mock_project_name"
project.labels = ["label1", "label2"]
project.labels = {"tag1": "value1"}
project.organization = organization
provider.projects = {"mock_project_id": project}
@@ -388,7 +388,7 @@ class TestFinding:
assert finding_output.subservice_name == "mock_subservice_name"
assert finding_output.severity == Severity.high
assert finding_output.resource_type == "mock_resource_type"
assert finding_output.resource_tags == "mock_resource_tags"
assert finding_output.resource_tags == {"tag1": "value1"}
assert finding_output.partition is None
assert finding_output.description == "mock_description"
assert finding_output.risk == "mock_risk"
@@ -408,7 +408,7 @@ class TestFinding:
assert finding_output.account_email == "mock_account_email"
assert finding_output.account_organization_uid == "mock_organization_id"
assert finding_output.account_organization_name == "mock_account_org_name"
assert finding_output.account_tags == ["label1", "label2"]
assert finding_output.account_tags == {"tag1": "value1"}
assert finding_output.prowler_version == "1.0.0"
assert finding_output.timestamp == 1622520000
@@ -459,7 +459,7 @@ class TestFinding:
assert finding_output.subservice_name == "mock_subservice_name"
assert finding_output.severity == Severity.high
assert finding_output.resource_type == "mock_resource_type"
assert finding_output.resource_tags == "mock_resource_tags"
assert finding_output.resource_tags == {"tag1": "value1"}
assert finding_output.partition is None
assert finding_output.description == "mock_description"
assert finding_output.risk == "mock_risk"
@@ -479,6 +479,6 @@ class TestFinding:
assert finding_output.account_email == "mock_account_email"
assert finding_output.account_organization_uid == "mock_account_org_uid"
assert finding_output.account_organization_name == "mock_account_org_name"
assert finding_output.account_tags == ["tag1", "tag2"]
assert finding_output.account_tags == {"tag1": "value1"}
assert finding_output.prowler_version == "1.0.0"
assert finding_output.timestamp == 1622520000
tests/lib/outputs/fixtures/fixtures.py
+10 -6
View File
@@ -16,7 +16,7 @@ def generate_finding_output(
resource_details: str = "",
resource_uid: str = "",
resource_name: str = "",
resource_tags: str = "",
resource_tags: dict = {},
compliance: dict = {"test-compliance": "test-compliance"},
timestamp: datetime = None,
provider: str = "aws",
@@ -34,6 +34,10 @@ def generate_finding_output(
depends_on: str = "test-dependency",
related_to: str = "test-related-to",
notes: str = "test-notes",
service_name: str = "test-service",
check_id: str = "test-check-id",
check_title: str = "test-check-id",
check_type: str = "test-type",
) -> Finding:
return Finding(
auth_method="profile: default",
@@ -43,16 +47,16 @@ def generate_finding_output(
account_email="",
account_organization_uid="test-organization-id",
account_organization_name="test-organization",
account_tags=["test-tag:test-value"],
account_tags={"test-tag": "test-value"},
finding_uid="test-unique-finding",
provider=provider,
check_id="test-check-id",
check_title="test-check-id",
check_type="test-type",
check_id=check_id,
check_title=check_title,
check_type=check_type,
status=status,
status_extended=status_extended,
muted=muted,
service_name="test-service",
service_name=service_name,
subservice_name="",
severity=severity,
resource_type="test-resource",
tests/lib/outputs/html/html_test.py
+25 -5
View File
@@ -45,11 +45,15 @@ fail_html_finding = """
<td>eu-west-1</td>
<td>test-check-id</td>
<td>test-check-id</td>
<td></td>
<td></td>
<td></td>
<td>test-resource-uid</td>
<td>
&#x2022;key1=value1
&#x2022;key2=value2
</td>
<td>test-status-extended</td>
<td><p class="show-read-more">test-risk</p></td>
<td><p class="show-read-more"></p> <a class="read-more" href=""><i class="fas fa-external-link-alt"></i></a></td>
<td><p class="show-read-more">test-remediation-recommendation-text</p> <a class="read-more" href=""><i class="fas fa-external-link-alt"></i></a></td>
<td><p class="show-read-more">
&#x2022;test-compliance: test-compliance
</p></td>
@@ -421,7 +425,23 @@ html_footer = """
class TestHTML:
def test_transform_fail_finding(self):
findings = [generate_finding_output(status="FAIL")]
findings = [
generate_finding_output(
status="FAIL",
resource_tags={"key1": "value1", "key2": "value2"},
severity="high",
service_name="test-service",
region=AWS_REGION_EU_WEST_1,
check_id="test-check-id",
check_title="test-check-id",
resource_uid="test-resource-uid",
status_extended="test-status-extended",
risk="test-risk",
remediation_recommendation_text="test-remediation-recommendation-text",
compliance={"test-compliance": "test-compliance"},
)
]
html = HTML(findings)
output_data = html.data[0]
assert isinstance(output_data, str)
tests/lib/outputs/ocsf/ocsf_test.py
+13 -5
View File
@@ -30,7 +30,11 @@ class TestOCSF:
def test_transform(self):
findings = [
generate_finding_output(
status="FAIL", severity="low", muted=False, region=AWS_REGION_EU_WEST_1
status="FAIL",
severity="low",
muted=False,
region=AWS_REGION_EU_WEST_1,
resource_tags={"Name": "test", "Environment": "dev"},
)
]
@@ -58,7 +62,7 @@ class TestOCSF:
assert output_data.status_code == findings[0].status
assert output_data.status_detail == findings[0].status_extended
assert output_data.risk_details == findings[0].risk
assert output_data.resources[0].labels == []
assert output_data.resources[0].labels == ["Name:test", "Environment:dev"]
assert output_data.resources[0].name == findings[0].resource_name
assert output_data.resources[0].uid == findings[0].resource_uid
assert output_data.resources[0].type == findings[0].resource_type
@@ -190,7 +194,11 @@ class TestOCSF:
def test_finding_output_cloud_pass_low_muted(self):
finding_output = generate_finding_output(
status="PASS", severity="low", muted=True, region=AWS_REGION_EU_WEST_1
status="PASS",
severity="low",
muted=True,
region=AWS_REGION_EU_WEST_1,
resource_tags={"Name": "test", "Environment": "dev"},
)
finding_ocsf = OCSF([finding_output])
@@ -248,7 +256,7 @@ class TestOCSF:
assert len(resource_details) == 1
assert isinstance(resource_details, list)
assert isinstance(resource_details[0], ResourceDetails)
assert resource_details[0].labels == []
assert resource_details[0].labels == ["Name:test", "Environment:dev"]
assert resource_details[0].name == finding_output.resource_name
assert resource_details[0].uid == finding_output.resource_uid
assert resource_details[0].type == finding_output.resource_type
@@ -287,7 +295,7 @@ class TestOCSF:
assert cloud_account.type_id == TypeID.AWS_Account
assert cloud_account.type == TypeID.AWS_Account.name
assert cloud_account.uid == finding_output.account_uid
assert cloud_account.labels == finding_output.account_tags
assert cloud_account.labels == ["test-tag:test-value"]
cloud_organization = cloud.org
assert isinstance(cloud_organization, Organization)
tests/lib/outputs/outputs_test.py
+32 -11
View File
@@ -98,6 +98,15 @@ class TestOutputs:
{"Key": "environment", "Value": "dev"},
{"Key": "terraform", "Value": "true"},
]
assert unroll_tags(dict_list) == {
"environment": "dev",
"name": "test",
"project": "prowler",
"terraform": "true",
}
def test_unroll_tags_unique(self):
unique_dict_list = [
{
"test1": "value1",
@@ -105,14 +114,26 @@ class TestOutputs:
"test3": "value3",
}
]
assert (
unroll_tags(dict_list)
== "name=test | project=prowler | environment=dev | terraform=true"
)
assert (
unroll_tags(unique_dict_list)
== "test1=value1 | test2=value2 | test3=value3"
)
assert unroll_tags(unique_dict_list) == {
"test1": "value1",
"test2": "value2",
"test3": "value3",
}
def test_unroll_tags_lowercase(self):
dict_list = [
{"key": "name", "value": "test"},
{"key": "project", "value": "prowler"},
{"key": "environment", "value": "dev"},
{"key": "terraform", "value": "true"},
]
assert unroll_tags(dict_list) == {
"environment": "dev",
"name": "test",
"project": "prowler",
"terraform": "true",
}
def test_unroll_dict(self):
test_compliance_dict = {
@@ -156,18 +177,18 @@ class TestOutputs:
"FedRAMP-Low-Revision-4": ["sc-13"],
}
assert (
unroll_dict(test_compliance_dict)
unroll_dict(test_compliance_dict, separator=": ")
== "CISA: your-systems-3, your-data-1, your-data-2 | CIS-1.4: 2.1.1 | CIS-1.5: 2.1.1 | GDPR: article_32 | AWS-Foundational-Security-Best-Practices: s3 | HIPAA: 164_308_a_1_ii_b, 164_308_a_4_ii_a, 164_312_a_2_iv, 164_312_c_1, 164_312_c_2, 164_312_e_2_ii | GxP-21-CFR-Part-11: 11.10-c, 11.30 | GxP-EU-Annex-11: 7.1-data-storage-damage-protection | NIST-800-171-Revision-2: 3_3_8, 3_5_10, 3_13_11, 3_13_16 | NIST-800-53-Revision-4: sc_28 | NIST-800-53-Revision-5: au_9_3, cm_6_a, cm_9_b, cp_9_d, cp_9_8, pm_11_b, sc_8_3, sc_8_4, sc_13_a, sc_16_1, sc_28_1, si_19_4 | ENS-RD2022: mp.si.2.aws.s3.1 | NIST-CSF-1.1: ds_1 | RBI-Cyber-Security-Framework: annex_i_1_3 | FFIEC: d3-pc-am-b-12 | PCI-3.2.1: s3 | FedRamp-Moderate-Revision-4: sc-13, sc-28 | FedRAMP-Low-Revision-4: sc-13"
)
def test_unroll_dict_to_list(self):
dict_A = {"A": "B"}
list_A = ["A: B"]
list_A = ["A:B"]
assert unroll_dict_to_list(dict_A) == list_A
dict_B = {"A": ["B", "C"]}
list_B = ["A: B, C"]
list_B = ["A:B, C"]
assert unroll_dict_to_list(dict_B) == list_B
tests/providers/aws/aws_provider_test.py
+2 -2
View File
@@ -270,7 +270,7 @@ class TestAWSProvider:
assert isinstance(aws_provider.organizations_metadata, AWSOrganizationsInfo)
assert aws_provider.organizations_metadata.account_email == "master@example.com"
assert aws_provider.organizations_metadata.account_name == "master"
assert aws_provider.organizations_metadata.account_tags == ["tagged:true"]
assert aws_provider.organizations_metadata.account_tags == {"tagged": "true"}
assert (
aws_provider.organizations_metadata.organization_account_arn
== f"arn:aws:organizations::{AWS_ACCOUNT_NUMBER}:account/{organization['Id']}/{AWS_ACCOUNT_NUMBER}"
@@ -351,7 +351,7 @@ class TestAWSProvider:
assert isinstance(aws_provider.organizations_metadata, AWSOrganizationsInfo)
assert aws_provider.organizations_metadata.account_email == "master@example.com"
assert aws_provider.organizations_metadata.account_name == "master"
assert aws_provider.organizations_metadata.account_tags == ["tagged:true"]
assert aws_provider.organizations_metadata.account_tags == {"tagged": "true"}
assert (
aws_provider.organizations_metadata.organization_account_arn
== f"arn:aws:organizations::{AWS_ACCOUNT_NUMBER}:account/{organization['Id']}/{AWS_ACCOUNT_NUMBER}"
tests/providers/aws/lib/organizations/organizations_test.py
+2 -2
View File
@@ -42,7 +42,7 @@ class Test_AWS_Organizations:
== f"arn:aws:organizations::{AWS_ACCOUNT_NUMBER}:organization/{org_id}"
)
assert org.organization_id == org_id
assert org.account_tags == ["key:value"]
assert org.account_tags == {"key": "value"}
def test_parse_organizations_metadata(self):
tags = {"Tags": [{"Key": "test-key", "Value": "test-value"}]}
@@ -70,4 +70,4 @@ class Test_AWS_Organizations:
== f"arn:aws:organizations::{AWS_ACCOUNT_NUMBER}:account/{organization_name}/{AWS_ACCOUNT_NUMBER}"
)
assert org.organization_arn == arn
assert org.account_tags == ["test-key:test-value"]
assert org.account_tags == {"test-key": "test-value"}
tests/providers/aws/lib/s3/s3_test.py
+1 -1
View File
@@ -26,7 +26,7 @@ FINDING = generate_finding_output(
resource_uid="resource-123",
resource_name="Example Resource",
resource_details="Detailed information about the resource",
resource_tags="tag1,tag2",
resource_tags={"key1": "tag1", "key2": "tag2"},
partition="aws",
description="Description of the finding",
risk="High",
tests/providers/gcp/gcp_provider_test.py
+9 -9
View File
@@ -29,7 +29,7 @@ class TestGCPProvider:
number="55555555",
id="project/55555555",
name="test-project",
labels=["test:value"],
labels={"test": "value"},
lifecycle_state="",
)
}
@@ -75,7 +75,7 @@ class TestGCPProvider:
number="55555555",
id="project/55555555",
name="test-project",
labels=["test:value"],
labels={"test": "value"},
lifecycle_state="",
)
}
@@ -148,7 +148,7 @@ class TestGCPProvider:
number="55555555",
id="project/55555555",
name="test-project",
labels=["test:value"],
labels={"test": "value"},
lifecycle_state="",
)
}
@@ -200,7 +200,7 @@ class TestGCPProvider:
number="55555555",
id="project/55555555",
name="test-project",
labels=["test:value"],
labels={"test": "value"},
lifecycle_state="",
)
}
@@ -242,7 +242,7 @@ class TestGCPProvider:
number="55555555",
id="project/55555555",
name="test-project",
labels=["test:value"],
labels={"test": "value"},
lifecycle_state="",
)
}
@@ -292,7 +292,7 @@ class TestGCPProvider:
number="55555555",
id="project/55555555",
name="test-project",
labels=["test:value"],
labels={"test": "value"},
lifecycle_state="",
)
}
@@ -341,7 +341,7 @@ class TestGCPProvider:
number="55555555",
id="project/55555555",
name="test-project",
labels=["test:value"],
labels={"test": "value"},
lifecycle_state="",
)
}
@@ -390,14 +390,14 @@ class TestGCPProvider:
number="55555555",
id="project/55555555",
name="test-project",
labels=["test:value"],
labels={"test": "value"},
lifecycle_state="",
),
"test-excluded-project": GCPProject(
number="12345678",
id="project/12345678",
name="test-excluded-project",
labels=["test:value"],
labels={"test": "value"},
lifecycle_state="",
),
}