fix: Some minor fixes in several parts (#4237)

Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>
This commit is contained in:
Pepe Fagoaga
2024-06-17 22:54:54 +02:00
committed by GitHub
parent 1252e6163b
commit e296d6e5c1
17 changed files with 64 additions and 54 deletions
+1 -1
View File
@@ -12,7 +12,7 @@ prowler <provider> --scan-unused-services
## Services that are ignored
### AWS
#### ACM
You can have certificates in ACM that is not in use by any AWS resource.
You can have certificates in ACM that are not in use by any AWS resource.
Prowler will check if every certificate is going to expire soon, if this certificate is not in use by default it is not going to be check if it is expired, is going to expire soon or it is good.
- `acm_certificates_expiration_check`
+3 -5
View File
@@ -64,6 +64,7 @@ default_config_file_path = (
default_fixer_config_file_path = (
f"{pathlib.Path(os.path.dirname(os.path.realpath(__file__)))}/fixer_config.yaml"
)
enconding_format_utf_8 = "utf-8"
available_output_formats = ["csv", "json-asff", "json-ocsf", "html"]
@@ -109,7 +110,7 @@ def load_and_validate_config_file(provider: str, config_file_path: str) -> dict:
dict: The configuration dictionary for the specified provider.
"""
try:
with open(config_file_path, "r", encoding="utf-8") as f:
with open(config_file_path, "r", encoding=enconding_format_utf_8) as f:
config_file = yaml.safe_load(f)
# Not to introduce a breaking change, allow the old format config file without any provider keys
@@ -156,12 +157,9 @@ def load_and_validate_fixer_config_file(
Returns:
dict: The fixer configuration dictionary for the specified provider.
Raises:
SystemExit: If there is an error reading or parsing the fixer configuration file.
"""
try:
with open(fixer_config_file_path, "r", encoding="utf-8") as f:
with open(fixer_config_file_path, "r", encoding=enconding_format_utf_8) as f:
fixer_config_file = yaml.safe_load(f)
return fixer_config_file.get(provider, {})
+7 -5
View File
@@ -4,6 +4,7 @@ import sys
from os import path
from prowler.config.config import (
enconding_format_utf_8,
html_file_suffix,
html_logo_url,
prowler_version,
@@ -11,6 +12,7 @@ from prowler.config.config import (
timestamp,
)
from prowler.lib.logger import logger
from prowler.lib.outputs.common_models import FindingOutput
from prowler.lib.outputs.utils import parse_html_string, unroll_dict
from prowler.lib.utils.utils import open_file
@@ -132,10 +134,10 @@ def add_html_header(file_descriptor, provider):
sys.exit(1)
def fill_html(file_descriptor, finding):
def fill_html(file_descriptor, finding: FindingOutput):
try:
row_class = "p-3 mb-2 bg-success-custom"
finding_status = finding.status.split(".")[0]
finding_status = finding.status.value
# Change the status of the finding if it's muted
if finding.muted:
finding_status = f"MUTED ({finding_status})"
@@ -149,7 +151,7 @@ def fill_html(file_descriptor, finding):
f"""
<tr class="{row_class}">
<td>{finding_status}</td>
<td>{finding.severity.split(".")[0]}</td>
<td>{finding.severity.value}</td>
<td>{finding.service_name}</td>
<td>{finding.region.lower()}</td>
<td>{finding.check_id.replace("_", "<wbr />_")}</td>
@@ -176,7 +178,7 @@ def fill_html_overview_statistics(stats, output_filename, output_directory):
# Read file
if path.isfile(filename):
with open(filename, "r", encoding="utf-8") as file:
with open(filename, "r", encoding=enconding_format_utf_8) as file:
filedata = file.read()
# Replace statistics
@@ -194,7 +196,7 @@ def fill_html_overview_statistics(stats, output_filename, output_directory):
filedata = filedata.replace("TOTAL_FAIL", str(stats.get("total_fail", 0)))
# Write file
with open(filename, "w", encoding="utf-8") as file:
with open(filename, "w", encoding=enconding_format_utf_8) as file:
file.write(filedata)
except FileNotFoundError as error:
+3 -2
View File
@@ -22,13 +22,14 @@ from colorama import Style
from detect_secrets import SecretsCollection
from detect_secrets.settings import default_settings
from prowler.config.config import enconding_format_utf_8
from prowler.lib.logger import logger
def open_file(input_file: str, mode: str = "r") -> TextIOWrapper:
"""open_file returns a handler to the file using the specified mode."""
try:
f = open(input_file, mode, encoding="utf-8")
f = open(input_file, mode, encoding=enconding_format_utf_8)
except OSError as os_error:
if os_error.strerror == "Too many open files":
logger.critical(
@@ -76,7 +77,7 @@ def file_exists(filename: str):
def hash_sha512(string: str) -> str:
"""hash_sha512 returns the first 9 bytes of the SHA512 representation for the given string."""
return sha512(string.encode("utf-8")).hexdigest()[0:9]
return sha512(string.encode(enconding_format_utf_8)).hexdigest()[0:9]
def detect_secrets_scan(data):
@@ -6,6 +6,7 @@ from base64 import b64decode
from detect_secrets import SecretsCollection
from detect_secrets.settings import default_settings
from prowler.config.config import enconding_format_utf_8
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.autoscaling.autoscaling_client import (
autoscaling_client,
@@ -27,10 +28,10 @@ class autoscaling_find_secrets_ec2_launch_configuration(Check):
if user_data[0:2] == b"\x1f\x8b": # GZIP magic number
user_data = zlib.decompress(user_data, zlib.MAX_WBITS | 32).decode(
"utf-8"
enconding_format_utf_8
)
else:
user_data = user_data.decode("utf-8")
user_data = user_data.decode(enconding_format_utf_8)
temp_user_data_file.write(
bytes(user_data, encoding="raw_unicode_escape")
@@ -6,6 +6,7 @@ from base64 import b64decode
from detect_secrets import SecretsCollection
from detect_secrets.settings import default_settings
from prowler.config.config import enconding_format_utf_8
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
@@ -26,9 +27,9 @@ class ec2_instance_secrets_user_data(Check):
if user_data[0:2] == b"\x1f\x8b": # GZIP magic number
user_data = zlib.decompress(
user_data, zlib.MAX_WBITS | 32
).decode("utf-8")
).decode(enconding_format_utf_8)
else:
user_data = user_data.decode("utf-8")
user_data = user_data.decode(enconding_format_utf_8)
temp_user_data_file.write(
bytes(user_data, encoding="raw_unicode_escape")
@@ -6,6 +6,7 @@ from base64 import b64decode
from detect_secrets import SecretsCollection
from detect_secrets.settings import default_settings
from prowler.config.config import enconding_format_utf_8
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
@@ -30,10 +31,10 @@ class ec2_launch_template_no_secrets(Check):
if user_data[0:2] == b"\x1f\x8b": # GZIP magic number
user_data = zlib.decompress(user_data, zlib.MAX_WBITS | 32).decode(
"utf-8"
enconding_format_utf_8
)
else:
user_data = user_data.decode("utf-8")
user_data = user_data.decode(enconding_format_utf_8)
temp_user_data_file.write(
bytes(user_data, encoding="raw_unicode_escape")
@@ -5,6 +5,7 @@ from typing import Optional
from botocore.client import ClientError
from pydantic import BaseModel
from prowler.config.config import enconding_format_utf_8
from prowler.lib.logger import logger
from prowler.lib.scan_filters.scan_filters import is_resource_filtered
from prowler.providers.aws.lib.service.service import AWSService
@@ -144,7 +145,9 @@ class IAM(AWSService):
if report_status["State"] == "COMPLETE":
report_is_completed = True
# Convert credential report to list of dictionaries
credential = self.client.get_credential_report()["Content"].decode("utf-8")
credential = self.client.get_credential_report()["Content"].decode(
enconding_format_utf_8
)
credential_lines = credential.split("\n")
csv_reader = csv.DictReader(credential_lines, delimiter=",")
credential_list = list(csv_reader)
@@ -11,7 +11,7 @@ class rds_instance_event_subscription_security_groups(Check):
report.status = "FAIL"
report.status_extended = "RDS security group event categories of configuration change and failure are not subscribed."
report.resource_id = rds_client.audited_account
report.resource_arn = rds_client.__get_trail_arn_template__(
report.resource_arn = rds_client.__get_rds_arn_template__(
db_event.region
)
report.region = db_event.region
@@ -33,7 +33,7 @@ class RDS(AWSService):
self.__threading_call__(self.__describe_db_engine_versions__)
self.__threading_call__(self.__describe_db_event_subscriptions__)
def __get_trail_arn_template__(self, region):
def __get_rds_arn_template__(self, region):
return (
f"arn:{self.audited_partition}:rds:{region}:{self.audited_account}:account"
if region
-8
View File
@@ -3,7 +3,6 @@ import os
import pathlib
from unittest import mock
import pytest
from requests import Response
from prowler.config.config import (
@@ -351,7 +350,6 @@ class Test_Config:
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
config_test_file = f"{path}/fixtures/config.yaml"
provider = "aws"
print(load_and_validate_config_file(provider, config_test_file))
assert load_and_validate_config_file(provider, config_test_file) == config_aws
def test_load_and_validate_config_file_gcp(self):
@@ -365,7 +363,6 @@ class Test_Config:
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
config_test_file = f"{path}/fixtures/config.yaml"
provider = "kubernetes"
print(load_and_validate_config_file(provider, config_test_file))
assert (
load_and_validate_config_file(provider, config_test_file)
== config_kubernetes
@@ -381,7 +378,6 @@ class Test_Config:
def test_load_and_validate_config_file_old_format(self):
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
config_test_file = f"{path}/fixtures/config_old.yaml"
print(load_and_validate_config_file("aws", config_test_file))
assert load_and_validate_config_file("aws", config_test_file) == old_config_aws
assert load_and_validate_config_file("gcp", config_test_file) == {}
assert load_and_validate_config_file("azure", config_test_file) == {}
@@ -396,8 +392,6 @@ class Test_Config:
assert "FileNotFoundError" in caplog.text
assert result == {}
assert pytest is not None
def test_load_and_validate_fixer_config_aws(self):
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
config_test_file = f"{path}/fixtures/fixer_config.yaml"
@@ -434,5 +428,3 @@ class Test_Config:
result = load_and_validate_fixer_config_file(provider, fixer_config_path)
assert "FileNotFoundError" in caplog.text
assert result == {}
assert pytest is not None
@@ -7,6 +7,7 @@ from boto3 import client, resource
from mock import patch
from moto import mock_aws
from prowler.config.config import enconding_format_utf_8
from prowler.providers.aws.lib.mutelist.mutelist import (
get_mutelist_file_from_dynamodb,
get_mutelist_file_from_lambda,
@@ -45,7 +46,7 @@ def mock_make_api_call(self, operation_name, kwarg):
}
}
}
).encode("utf-8")
).encode(enconding_format_utf_8)
)
}
@@ -73,6 +73,7 @@ class Test_acm_certificates_expiration_check:
assert result[0].resource_arn == certificate_arn
assert result[0].region == AWS_REGION
assert result[0].resource_tags == []
assert result[0].check_metadata.Severity == "medium"
def test_acm_certificate_expirated_long_time(self):
certificate_id = str(uuid.uuid4())
@@ -119,6 +120,7 @@ class Test_acm_certificates_expiration_check:
assert result[0].resource_arn == certificate_arn
assert result[0].region == AWS_REGION
assert result[0].resource_tags == []
assert result[0].check_metadata.Severity == "high"
def test_acm_certificate_not_expirated(self):
certificate_id = str(uuid.uuid4())
@@ -252,3 +254,4 @@ class Test_acm_certificates_expiration_check:
assert result[0].resource_arn == certificate_arn
assert result[0].region == AWS_REGION
assert result[0].resource_tags == []
assert result[0].check_metadata.Severity == "high"
@@ -3,6 +3,7 @@ from base64 import b64decode
from boto3 import client
from moto import mock_aws
from prowler.config.config import enconding_format_utf_8
from prowler.providers.aws.services.autoscaling.autoscaling_service import AutoScaling
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
@@ -72,7 +73,9 @@ class Test_AutoScaling_Service:
assert len(autoscaling.launch_configurations) == 2
assert autoscaling.launch_configurations[0].name == "tester1"
assert (
b64decode(autoscaling.launch_configurations[0].user_data).decode("utf-8")
b64decode(autoscaling.launch_configurations[0].user_data).decode(
enconding_format_utf_8
)
== "DB_PASSWORD=foobar123"
)
assert autoscaling.launch_configurations[0].image_id == "ami-12c6146b"
@@ -6,6 +6,7 @@ from unittest import mock
from boto3 import client
from moto import mock_aws
from prowler.config.config import enconding_format_utf_8
from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
ACTUAL_DIRECTORY = Path(path.dirname(path.realpath(__file__)))
@@ -49,9 +50,9 @@ class Test_ec2_launch_template_no_secrets:
VersionDescription="Launch Template without secrets",
LaunchTemplateData={
"InstanceType": "t1.micro",
"UserData": b64encode("This is some user_data".encode("utf-8")).decode(
"utf-8"
),
"UserData": b64encode(
"This is some user_data".encode(enconding_format_utf_8)
).decode(enconding_format_utf_8),
},
)
@@ -103,7 +104,7 @@ class Test_ec2_launch_template_no_secrets:
VersionDescription="Launch Template with secrets",
LaunchTemplateData={
"InstanceType": "t1.micro",
"UserData": b64encode(secrets).decode("utf-8"),
"UserData": b64encode(secrets).decode(enconding_format_utf_8),
},
)
@@ -159,7 +160,7 @@ class Test_ec2_launch_template_no_secrets:
VersionDescription="Launch Template with secrets",
LaunchTemplateData={
"InstanceType": "t1.micro",
"UserData": b64encode(secrets).decode("utf-8"),
"UserData": b64encode(secrets).decode(enconding_format_utf_8),
},
)
@@ -168,7 +169,7 @@ class Test_ec2_launch_template_no_secrets:
VersionDescription="Second Launch Template version with secrets",
LaunchTemplateData={
"InstanceType": "t1.micro",
"UserData": b64encode(secrets).decode("utf-8"),
"UserData": b64encode(secrets).decode(enconding_format_utf_8),
},
)
@@ -227,7 +228,7 @@ class Test_ec2_launch_template_no_secrets:
VersionDescription="Launch Template with secrets",
LaunchTemplateData={
"InstanceType": "t1.micro",
"UserData": b64encode(secrets).decode("utf-8"),
"UserData": b64encode(secrets).decode(enconding_format_utf_8),
},
)
@@ -291,7 +292,7 @@ class Test_ec2_launch_template_no_secrets:
VersionDescription="Launch Template with secrets",
LaunchTemplateData={
"InstanceType": "t1.micro",
"UserData": b64encode(secrets).decode("utf-8"),
"UserData": b64encode(secrets).decode(enconding_format_utf_8),
},
)
@@ -393,7 +394,7 @@ class Test_ec2_launch_template_no_secrets:
VersionDescription="Launch Template with secrets",
LaunchTemplateData={
"InstanceType": "t1.micro",
"UserData": b64encode(secrets).decode("utf-8"),
"UserData": b64encode(secrets).decode(enconding_format_utf_8),
},
)
@@ -410,7 +411,7 @@ class Test_ec2_launch_template_no_secrets:
VersionDescription="Launch Template without secrets",
LaunchTemplateData={
"InstanceType": "t1.micro",
"UserData": b64encode(b"Test").decode("utf-8"),
"UserData": b64encode(b"Test").decode(enconding_format_utf_8),
},
)
@@ -10,6 +10,7 @@ from dateutil.tz import tzutc
from freezegun import freeze_time
from moto import mock_aws
from prowler.config.config import enconding_format_utf_8
from prowler.providers.aws.services.ec2.ec2_service import EC2
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
@@ -320,7 +321,9 @@ class Test_EC2_Service:
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
ec2 = EC2(aws_provider)
assert user_data == b64decode(ec2.instances[0].user_data).decode("utf-8")
assert user_data == b64decode(ec2.instances[0].user_data).decode(
enconding_format_utf_8
)
# Test EC2 Get EBS Encryption by default
@mock_aws
@@ -632,9 +635,9 @@ class Test_EC2_Service:
VersionDescription="Test EC Launch Template 1 (Secret in UserData)",
LaunchTemplateData={
"InstanceType": TEMPLATE_INSTANCE_TYPE,
"UserData": b64encode(KNOWN_SECRET_USER_DATA.encode("utf-8")).decode(
"utf-8"
),
"UserData": b64encode(
KNOWN_SECRET_USER_DATA.encode(enconding_format_utf_8)
).decode(enconding_format_utf_8),
},
)
@@ -677,9 +680,9 @@ class Test_EC2_Service:
VersionDescription="Updated Test EC Launch Template 1",
LaunchTemplateData={
"InstanceType": TEMPLATE_INSTANCE_TYPE,
"UserData": b64encode(KNOWN_SECRET_USER_DATA.encode("utf-8")).decode(
"utf-8"
),
"UserData": b64encode(
KNOWN_SECRET_USER_DATA.encode(enconding_format_utf_8)
).decode(enconding_format_utf_8),
},
)
@@ -701,6 +704,6 @@ class Test_EC2_Service:
assert version2.template_data["InstanceType"] == TEMPLATE_INSTANCE_TYPE
assert (
b64decode(version2.template_data["UserData"]).decode("utf-8")
b64decode(version2.template_data["UserData"]).decode(enconding_format_utf_8)
== KNOWN_SECRET_USER_DATA
)
@@ -11,7 +11,7 @@ from tests.providers.aws.utils import (
)
make_api_call = botocore.client.BaseClient._make_api_call
rds_account_arn = f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:account"
RDS_ACCOUNT_ARN = f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:account"
class Test_rds_instance__no_event_subscriptions:
@@ -45,7 +45,7 @@ class Test_rds_instance__no_event_subscriptions:
)
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_arn == rds_account_arn
assert result[0].resource_arn == RDS_ACCOUNT_ARN
@mock_aws
def test_rds_no_events_ignoring(self):
@@ -301,7 +301,7 @@ class Test_rds_instance__no_event_subscriptions:
)
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_arn == rds_account_arn
assert result[0].resource_arn == RDS_ACCOUNT_ARN
@mock_aws
def test_rds_no_event_subscription(self):
@@ -348,4 +348,4 @@ class Test_rds_instance__no_event_subscriptions:
)
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_arn == rds_account_arn
assert result[0].resource_arn == RDS_ACCOUNT_ARN