mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-23 12:31:54 +00:00
chore(s3): create class and refactor (#4457)
Co-authored-by: pedrooot <pedromarting3@gmail.com> Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Pepe Fagoaga
+90
-71
@@ -66,7 +66,7 @@ from prowler.lib.outputs.ocsf.ocsf import OCSF
|
||||
from prowler.lib.outputs.outputs import extract_findings_statistics
|
||||
from prowler.lib.outputs.slack.slack import Slack
|
||||
from prowler.lib.outputs.summary_table import display_summary_table
|
||||
from prowler.providers.aws.lib.s3.s3 import send_to_s3_bucket
|
||||
from prowler.providers.aws.lib.s3.s3 import S3
|
||||
from prowler.providers.aws.lib.security_hub.security_hub import SecurityHub
|
||||
from prowler.providers.common.provider import Provider
|
||||
from prowler.providers.common.quick_inventory import run_provider_quick_inventory
|
||||
@@ -306,6 +306,8 @@ def prowler():
|
||||
Finding.generate_output(global_provider, finding) for finding in findings
|
||||
]
|
||||
|
||||
generated_outputs = {"regular": [], "compliance": []}
|
||||
|
||||
if args.output_formats:
|
||||
for mode in args.output_formats:
|
||||
filename = (
|
||||
@@ -318,6 +320,7 @@ def prowler():
|
||||
create_file_descriptor=True,
|
||||
file_path=f"{filename}{csv_file_suffix}",
|
||||
)
|
||||
generated_outputs["regular"].append(csv_output)
|
||||
# Write CSV Finding Object to file
|
||||
csv_output.batch_write_data_to_file()
|
||||
|
||||
@@ -327,6 +330,7 @@ def prowler():
|
||||
create_file_descriptor=True,
|
||||
file_path=f"{filename}{json_asff_file_suffix}",
|
||||
)
|
||||
generated_outputs["regular"].append(asff_output)
|
||||
# Write ASFF Finding Object to file
|
||||
asff_output.batch_write_data_to_file()
|
||||
|
||||
@@ -336,6 +340,7 @@ def prowler():
|
||||
create_file_descriptor=True,
|
||||
file_path=f"{filename}{json_ocsf_file_suffix}",
|
||||
)
|
||||
generated_outputs["regular"].append(json_output)
|
||||
json_output.batch_write_data_to_file()
|
||||
if mode == "html":
|
||||
html_output = HTML(
|
||||
@@ -343,28 +348,11 @@ def prowler():
|
||||
create_file_descriptor=True,
|
||||
file_path=f"{filename}{html_file_suffix}",
|
||||
)
|
||||
generated_outputs["regular"].append(html_output)
|
||||
html_output.batch_write_data_to_file(
|
||||
provider=global_provider, stats=stats
|
||||
)
|
||||
|
||||
# Send output to S3 if needed (-B / -D)
|
||||
if provider == "aws" and (
|
||||
args.output_bucket or args.output_bucket_no_assume
|
||||
):
|
||||
output_bucket = args.output_bucket
|
||||
bucket_session = global_provider.session.current_session
|
||||
# Check if -D was input
|
||||
if args.output_bucket_no_assume:
|
||||
output_bucket = args.output_bucket_no_assume
|
||||
bucket_session = global_provider.session.original_session
|
||||
send_to_s3_bucket(
|
||||
global_provider.output_options.output_filename,
|
||||
args.output_directory,
|
||||
mode,
|
||||
output_bucket,
|
||||
bucket_session,
|
||||
)
|
||||
|
||||
# Compliance Frameworks
|
||||
input_compliance_frameworks = set(
|
||||
global_provider.output_options.output_modes
|
||||
@@ -377,65 +365,70 @@ def prowler():
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
cis_finding = AWSCIS(
|
||||
cis = AWSCIS(
|
||||
findings=finding_outputs,
|
||||
compliance=bulk_compliance_frameworks[compliance_name],
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
cis_finding.batch_write_data_to_file()
|
||||
generated_outputs["compliance"].append(cis)
|
||||
cis.batch_write_data_to_file()
|
||||
elif compliance_name == "mitre_attack_aws":
|
||||
# Generate MITRE ATT&CK Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
mitre_attack_finding = AWSMitreAttack(
|
||||
mitre_attack = AWSMitreAttack(
|
||||
findings=finding_outputs,
|
||||
compliance=bulk_compliance_frameworks[compliance_name],
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
mitre_attack_finding.batch_write_data_to_file()
|
||||
generated_outputs["compliance"].append(mitre_attack)
|
||||
mitre_attack.batch_write_data_to_file()
|
||||
elif compliance_name.startswith("ens_"):
|
||||
# Generate ENS Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
ens_finding = AWSENS(
|
||||
ens = AWSENS(
|
||||
findings=finding_outputs,
|
||||
compliance=bulk_compliance_frameworks[compliance_name],
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
ens_finding.batch_write_data_to_file()
|
||||
generated_outputs["compliance"].append(ens)
|
||||
ens.batch_write_data_to_file()
|
||||
elif compliance_name.startswith("aws_well_architected_framework"):
|
||||
# Generate AWS Well-Architected Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
aws_well_architected_finding = AWSWellArchitected(
|
||||
aws_well_architected = AWSWellArchitected(
|
||||
findings=finding_outputs,
|
||||
compliance=bulk_compliance_frameworks[compliance_name],
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
aws_well_architected_finding.batch_write_data_to_file()
|
||||
generated_outputs["compliance"].append(aws_well_architected)
|
||||
aws_well_architected.batch_write_data_to_file()
|
||||
elif compliance_name.startswith("iso27001_"):
|
||||
# Generate ISO27001 Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
iso27001_finding = AWSISO27001(
|
||||
iso27001 = AWSISO27001(
|
||||
findings=finding_outputs,
|
||||
compliance=bulk_compliance_frameworks[compliance_name],
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
iso27001_finding.batch_write_data_to_file()
|
||||
generated_outputs["compliance"].append(iso27001)
|
||||
iso27001.batch_write_data_to_file()
|
||||
else:
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
@@ -447,6 +440,7 @@ def prowler():
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
generated_outputs["compliance"].append(generic_compliance)
|
||||
generic_compliance.batch_write_data_to_file()
|
||||
|
||||
elif provider == "azure":
|
||||
@@ -457,26 +451,28 @@ def prowler():
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
cis_finding = AzureCIS(
|
||||
cis = AzureCIS(
|
||||
findings=finding_outputs,
|
||||
compliance=bulk_compliance_frameworks[compliance_name],
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
cis_finding.batch_write_data_to_file()
|
||||
generated_outputs["compliance"].append(cis)
|
||||
cis.batch_write_data_to_file()
|
||||
elif compliance_name == "mitre_attack_azure":
|
||||
# Generate MITRE ATT&CK Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
mitre_attack_finding = AzureMitreAttack(
|
||||
mitre_attack = AzureMitreAttack(
|
||||
findings=finding_outputs,
|
||||
compliance=bulk_compliance_frameworks[compliance_name],
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
mitre_attack_finding.batch_write_data_to_file()
|
||||
generated_outputs["compliance"].append(mitre_attack)
|
||||
mitre_attack.batch_write_data_to_file()
|
||||
else:
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
@@ -488,6 +484,7 @@ def prowler():
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
generated_outputs["compliance"].append(generic_compliance)
|
||||
generic_compliance.batch_write_data_to_file()
|
||||
|
||||
elif provider == "gcp":
|
||||
@@ -498,26 +495,28 @@ def prowler():
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
cis_finding = GCPCIS(
|
||||
cis = GCPCIS(
|
||||
findings=finding_outputs,
|
||||
compliance=bulk_compliance_frameworks[compliance_name],
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
cis_finding.batch_write_data_to_file()
|
||||
generated_outputs["compliance"].append(cis)
|
||||
cis.batch_write_data_to_file()
|
||||
elif compliance_name == "mitre_attack_gcp":
|
||||
# Generate MITRE ATT&CK Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
mitre_attack_finding = GCPMitreAttack(
|
||||
mitre_attack = GCPMitreAttack(
|
||||
findings=finding_outputs,
|
||||
compliance=bulk_compliance_frameworks[compliance_name],
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
mitre_attack_finding.batch_write_data_to_file()
|
||||
generated_outputs["compliance"].append(mitre_attack)
|
||||
mitre_attack.batch_write_data_to_file()
|
||||
else:
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
@@ -529,6 +528,7 @@ def prowler():
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
generated_outputs["compliance"].append(generic_compliance)
|
||||
generic_compliance.batch_write_data_to_file()
|
||||
|
||||
elif provider == "kubernetes":
|
||||
@@ -539,13 +539,14 @@ def prowler():
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
cis_finding = KubernetesCIS(
|
||||
cis = KubernetesCIS(
|
||||
findings=finding_outputs,
|
||||
compliance=bulk_compliance_frameworks[compliance_name],
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
cis_finding.batch_write_data_to_file()
|
||||
generated_outputs["compliance"].append(cis)
|
||||
cis.batch_write_data_to_file()
|
||||
else:
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
@@ -557,45 +558,63 @@ def prowler():
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
generated_outputs["compliance"].append(generic_compliance)
|
||||
generic_compliance.batch_write_data_to_file()
|
||||
|
||||
# AWS Security Hub Integration
|
||||
if provider == "aws" and args.security_hub:
|
||||
print(
|
||||
f"{Style.BRIGHT}\nSending findings to AWS Security Hub, please wait...{Style.RESET_ALL}"
|
||||
)
|
||||
|
||||
security_hub_regions = (
|
||||
global_provider.get_available_aws_service_regions("securityhub")
|
||||
if not global_provider.identity.audited_regions
|
||||
else global_provider.identity.audited_regions
|
||||
)
|
||||
|
||||
security_hub = SecurityHub(
|
||||
aws_account_id=global_provider.identity.account,
|
||||
aws_partition=global_provider.identity.partition,
|
||||
aws_session=global_provider.session.current_session,
|
||||
findings=asff_output.data,
|
||||
status=global_provider.output_options.status,
|
||||
send_only_fails=global_provider.output_options.send_sh_only_fails,
|
||||
aws_security_hub_available_regions=security_hub_regions,
|
||||
)
|
||||
# Send the findings to Security Hub
|
||||
findings_sent_to_security_hub = security_hub.batch_send_to_security_hub()
|
||||
print(
|
||||
f"{Style.BRIGHT}{Fore.GREEN}\n{findings_sent_to_security_hub} findings sent to AWS Security Hub!{Style.RESET_ALL}"
|
||||
)
|
||||
|
||||
# Resolve previous fails of Security Hub
|
||||
if not args.skip_sh_update:
|
||||
print(
|
||||
f"{Style.BRIGHT}\nArchiving previous findings in AWS Security Hub, please wait...{Style.RESET_ALL}"
|
||||
if provider == "aws":
|
||||
# Send output to S3 if needed (-B / -D) for all the output formats
|
||||
if args.output_bucket or args.output_bucket_no_assume:
|
||||
output_bucket = args.output_bucket
|
||||
bucket_session = global_provider.session.current_session
|
||||
# Check if -D was input
|
||||
if args.output_bucket_no_assume:
|
||||
output_bucket = args.output_bucket_no_assume
|
||||
bucket_session = global_provider.session.original_session
|
||||
s3 = S3(
|
||||
session=bucket_session,
|
||||
bucket_name=output_bucket,
|
||||
output_directory=args.output_directory,
|
||||
)
|
||||
findings_archived_in_security_hub = security_hub.archive_previous_findings()
|
||||
s3.send_to_bucket(generated_outputs)
|
||||
if args.security_hub:
|
||||
print(
|
||||
f"{Style.BRIGHT}{Fore.GREEN}\n{findings_archived_in_security_hub} findings archived in AWS Security Hub!{Style.RESET_ALL}"
|
||||
f"{Style.BRIGHT}\nSending findings to AWS Security Hub, please wait...{Style.RESET_ALL}"
|
||||
)
|
||||
|
||||
security_hub_regions = (
|
||||
global_provider.get_available_aws_service_regions("securityhub")
|
||||
if not global_provider.identity.audited_regions
|
||||
else global_provider.identity.audited_regions
|
||||
)
|
||||
|
||||
security_hub = SecurityHub(
|
||||
aws_account_id=global_provider.identity.account,
|
||||
aws_partition=global_provider.identity.partition,
|
||||
aws_session=global_provider.session.current_session,
|
||||
findings=asff_output.data,
|
||||
status=global_provider.output_options.status,
|
||||
send_only_fails=global_provider.output_options.send_sh_only_fails,
|
||||
aws_security_hub_available_regions=security_hub_regions,
|
||||
)
|
||||
# Send the findings to Security Hub
|
||||
findings_sent_to_security_hub = security_hub.batch_send_to_security_hub()
|
||||
print(
|
||||
f"{Style.BRIGHT}{Fore.GREEN}\n{findings_sent_to_security_hub} findings sent to AWS Security Hub!{Style.RESET_ALL}"
|
||||
)
|
||||
|
||||
# Resolve previous fails of Security Hub
|
||||
if not args.skip_sh_update:
|
||||
print(
|
||||
f"{Style.BRIGHT}\nArchiving previous findings in AWS Security Hub, please wait...{Style.RESET_ALL}"
|
||||
)
|
||||
findings_archived_in_security_hub = (
|
||||
security_hub.archive_previous_findings()
|
||||
)
|
||||
print(
|
||||
f"{Style.BRIGHT}{Fore.GREEN}\n{findings_archived_in_security_hub} findings archived in AWS Security Hub!{Style.RESET_ALL}"
|
||||
)
|
||||
|
||||
# Display summary table
|
||||
if not args.only_logs:
|
||||
display_summary_table(
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
from csv import DictWriter
|
||||
from io import TextIOWrapper
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
|
||||
from prowler.lib.check.compliance_models import ComplianceBaseModel
|
||||
@@ -25,17 +25,21 @@ class ComplianceOutput(Output):
|
||||
create_file_descriptor: Method to create a file descriptor for writing data to a file.
|
||||
"""
|
||||
|
||||
_data: list
|
||||
_file_descriptor: TextIOWrapper
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
findings: List[Finding],
|
||||
compliance: ComplianceBaseModel,
|
||||
create_file_descriptor: bool = False,
|
||||
file_path: str = None,
|
||||
file_extension: str = "",
|
||||
) -> None:
|
||||
self._data = []
|
||||
|
||||
if not file_extension and file_path:
|
||||
self._file_extension = "".join(Path(file_path).suffixes)
|
||||
if file_extension:
|
||||
self._file_extension = file_extension
|
||||
|
||||
if findings:
|
||||
# Get the compliance name of the model
|
||||
compliance_name = (
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from io import TextIOWrapper
|
||||
from os import path
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
@@ -40,13 +40,13 @@ class Output(ABC):
|
||||
self._data = []
|
||||
|
||||
if not file_extension and file_path:
|
||||
_, self._file_extension = path.splitext(file_path)
|
||||
self._file_extension = "".join(Path(file_path).suffixes)
|
||||
if file_extension:
|
||||
self._file_extension = file_extension
|
||||
|
||||
if findings:
|
||||
self.transform(findings)
|
||||
if create_file_descriptor:
|
||||
if create_file_descriptor and file_path:
|
||||
self.create_file_descriptor(file_path)
|
||||
|
||||
@property
|
||||
@@ -73,7 +73,7 @@ class Output(ABC):
|
||||
def batch_write_data_to_file(self) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
def create_file_descriptor(self, file_path) -> None:
|
||||
def create_file_descriptor(self, file_path: str) -> None:
|
||||
"""
|
||||
Creates a file descriptor for writing data to a file.
|
||||
|
||||
|
||||
@@ -16,276 +16,292 @@ from prowler.config.config import (
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.providers.aws.aws_provider import AwsProvider
|
||||
from prowler.providers.aws.lib.arn.models import get_arn_resource_type
|
||||
from prowler.providers.aws.lib.s3.s3 import send_to_s3_bucket
|
||||
|
||||
|
||||
def quick_inventory(provider: AwsProvider, args):
|
||||
resources = []
|
||||
global_resources = []
|
||||
total_resources_per_region = {}
|
||||
iam_was_scanned = False
|
||||
# If not inputed regions, check all of them
|
||||
if not provider.identity.audited_regions:
|
||||
# EC2 client for describing all regions
|
||||
ec2_client = provider.session.current_session.client(
|
||||
"ec2", region_name=provider.identity.profile_region
|
||||
)
|
||||
# Get all the available regions
|
||||
provider.identity.audited_regions = [
|
||||
region["RegionName"] for region in ec2_client.describe_regions()["Regions"]
|
||||
]
|
||||
try:
|
||||
resources = []
|
||||
global_resources = []
|
||||
total_resources_per_region = {}
|
||||
iam_was_scanned = False
|
||||
# If not inputed regions, check all of them
|
||||
if not provider.identity.audited_regions:
|
||||
# EC2 client for describing all regions
|
||||
ec2_client = provider.session.current_session.client(
|
||||
"ec2", region_name=provider.identity.profile_region
|
||||
)
|
||||
# Get all the available regions
|
||||
provider.identity.audited_regions = [
|
||||
region["RegionName"]
|
||||
for region in ec2_client.describe_regions()["Regions"]
|
||||
]
|
||||
|
||||
with alive_bar(
|
||||
total=len(provider.identity.audited_regions),
|
||||
ctrl_c=False,
|
||||
bar="blocks",
|
||||
spinner="classic",
|
||||
stats=False,
|
||||
enrich_print=False,
|
||||
) as bar:
|
||||
for region in sorted(provider.identity.audited_regions):
|
||||
bar.title = f"Inventorying AWS Account {orange_color}{provider.identity.account}{Style.RESET_ALL}"
|
||||
resources_in_region = []
|
||||
# {
|
||||
# eu-west-1: 100,...
|
||||
# }
|
||||
with alive_bar(
|
||||
total=len(provider.identity.audited_regions),
|
||||
ctrl_c=False,
|
||||
bar="blocks",
|
||||
spinner="classic",
|
||||
stats=False,
|
||||
enrich_print=False,
|
||||
) as bar:
|
||||
for region in sorted(provider.identity.audited_regions):
|
||||
bar.title = f"Inventorying AWS Account {orange_color}{provider.identity.account}{Style.RESET_ALL}"
|
||||
resources_in_region = []
|
||||
# {
|
||||
# eu-west-1: 100,...
|
||||
# }
|
||||
|
||||
try:
|
||||
# Scan IAM only once
|
||||
if not iam_was_scanned:
|
||||
global_resources.extend(
|
||||
get_iam_resources(provider.session.current_session)
|
||||
)
|
||||
iam_was_scanned = True
|
||||
|
||||
# Get regional S3 buckets since none-tagged buckets are not supported by the resourcegroupstaggingapi
|
||||
resources_in_region.extend(get_regional_buckets(provider, region))
|
||||
|
||||
client = provider.session.current_session.client(
|
||||
"resourcegroupstaggingapi", region_name=region
|
||||
)
|
||||
# Get all the resources
|
||||
resources_count = 0
|
||||
try:
|
||||
get_resources_paginator = client.get_paginator("get_resources")
|
||||
for page in get_resources_paginator.paginate():
|
||||
resources_count += len(page["ResourceTagMappingList"])
|
||||
for resource in page["ResourceTagMappingList"]:
|
||||
# Avoid adding S3 buckets again:
|
||||
if resource["ResourceARN"].split(":")[2] != "s3":
|
||||
# Check if region is not in ARN --> Global service
|
||||
if not resource["ResourceARN"].split(":")[3]:
|
||||
global_resources.append(
|
||||
{
|
||||
"arn": resource["ResourceARN"],
|
||||
"tags": resource["Tags"],
|
||||
}
|
||||
)
|
||||
else:
|
||||
resources_in_region.append(
|
||||
{
|
||||
"arn": resource["ResourceARN"],
|
||||
"tags": resource["Tags"],
|
||||
}
|
||||
)
|
||||
# Scan IAM only once
|
||||
if not iam_was_scanned:
|
||||
global_resources.extend(
|
||||
get_iam_resources(provider.session.current_session)
|
||||
)
|
||||
iam_was_scanned = True
|
||||
|
||||
# Get regional S3 buckets since none-tagged buckets are not supported by the resourcegroupstaggingapi
|
||||
resources_in_region.extend(get_regional_buckets(provider, region))
|
||||
|
||||
client = provider.session.current_session.client(
|
||||
"resourcegroupstaggingapi", region_name=region
|
||||
)
|
||||
# Get all the resources
|
||||
resources_count = 0
|
||||
try:
|
||||
get_resources_paginator = client.get_paginator("get_resources")
|
||||
for page in get_resources_paginator.paginate():
|
||||
resources_count += len(page["ResourceTagMappingList"])
|
||||
for resource in page["ResourceTagMappingList"]:
|
||||
# Avoid adding S3 buckets again:
|
||||
if resource["ResourceARN"].split(":")[2] != "s3":
|
||||
# Check if region is not in ARN --> Global service
|
||||
if not resource["ResourceARN"].split(":")[3]:
|
||||
global_resources.append(
|
||||
{
|
||||
"arn": resource["ResourceARN"],
|
||||
"tags": resource["Tags"],
|
||||
}
|
||||
)
|
||||
else:
|
||||
resources_in_region.append(
|
||||
{
|
||||
"arn": resource["ResourceARN"],
|
||||
"tags": resource["Tags"],
|
||||
}
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
bar()
|
||||
if len(resources_in_region) > 0:
|
||||
total_resources_per_region[region] = len(resources_in_region)
|
||||
bar.text = f"-> Found {Fore.GREEN}{len(resources_in_region)}{Style.RESET_ALL} resources in {region}"
|
||||
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
bar()
|
||||
if len(resources_in_region) > 0:
|
||||
total_resources_per_region[region] = len(resources_in_region)
|
||||
bar.text = f"-> Found {Fore.GREEN}{len(resources_in_region)}{Style.RESET_ALL} resources in {region}"
|
||||
bar()
|
||||
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
bar()
|
||||
resources.extend(resources_in_region)
|
||||
bar.title = f"-> {Fore.GREEN}Quick Inventory completed!{Style.RESET_ALL}"
|
||||
|
||||
resources.extend(resources_in_region)
|
||||
bar.title = f"-> {Fore.GREEN}Quick Inventory completed!{Style.RESET_ALL}"
|
||||
resources.extend(global_resources)
|
||||
total_resources_per_region["global"] = len(global_resources)
|
||||
inventory_table = create_inventory_table(resources, total_resources_per_region)
|
||||
|
||||
resources.extend(global_resources)
|
||||
total_resources_per_region["global"] = len(global_resources)
|
||||
inventory_table = create_inventory_table(resources, total_resources_per_region)
|
||||
|
||||
print(
|
||||
f"\nQuick Inventory of AWS Account {Fore.YELLOW}{provider.identity.account}{Style.RESET_ALL}:"
|
||||
)
|
||||
|
||||
print(
|
||||
tabulate(
|
||||
inventory_table, headers="keys", tablefmt="rounded_grid", stralign="left"
|
||||
print(
|
||||
f"\nQuick Inventory of AWS Account {Fore.YELLOW}{provider.identity.account}{Style.RESET_ALL}:"
|
||||
)
|
||||
)
|
||||
print(f"\nTotal resources found: {Fore.GREEN}{len(resources)}{Style.RESET_ALL}")
|
||||
|
||||
create_output(resources, provider, args)
|
||||
print(
|
||||
tabulate(
|
||||
inventory_table,
|
||||
headers="keys",
|
||||
tablefmt="rounded_grid",
|
||||
stralign="left",
|
||||
)
|
||||
)
|
||||
print(f"\nTotal resources found: {Fore.GREEN}{len(resources)}{Style.RESET_ALL}")
|
||||
|
||||
create_output(resources, provider, args)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
|
||||
def create_inventory_table(resources: list, resources_in_region: dict) -> dict:
|
||||
regions_with_resources = list(resources_in_region.keys())
|
||||
services = {}
|
||||
# { "S3":
|
||||
# 123,
|
||||
# "IAM":
|
||||
# 239,
|
||||
# }
|
||||
resources_type = {}
|
||||
# { "S3":
|
||||
# "Buckets":
|
||||
# eu-west-1: 10,
|
||||
# eu-west-2: 3,
|
||||
# "IAM":
|
||||
# "Roles":
|
||||
# us-east-1: 143,
|
||||
# "Users":
|
||||
# us-west-2: 22,
|
||||
# }
|
||||
|
||||
inventory_table = {
|
||||
"Service": [],
|
||||
f"Total\n({Fore.GREEN}{str(len(resources))}{Style.RESET_ALL})": [],
|
||||
"Total per\nresource type": [],
|
||||
}
|
||||
|
||||
for region, count in resources_in_region.items():
|
||||
inventory_table[f"{region}\n({Fore.GREEN}{str(count)}{Style.RESET_ALL})"] = []
|
||||
|
||||
for resource in sorted(resources, key=lambda d: d["arn"]):
|
||||
service = resource["arn"].split(":")[2]
|
||||
region = resource["arn"].split(":")[3]
|
||||
if not region:
|
||||
region = "global"
|
||||
if service not in services:
|
||||
services[service] = 0
|
||||
services[service] += 1
|
||||
|
||||
resource_type = get_arn_resource_type(resource["arn"], service)
|
||||
|
||||
if service not in resources_type:
|
||||
resources_type[service] = {}
|
||||
if resource_type not in resources_type[service]:
|
||||
resources_type[service][resource_type] = {}
|
||||
if region not in resources_type[service][resource_type]:
|
||||
resources_type[service][resource_type][region] = 0
|
||||
resources_type[service][resource_type][region] += 1
|
||||
|
||||
# Add results to inventory table
|
||||
for service in services:
|
||||
pending_regions = deepcopy(regions_with_resources)
|
||||
aux = {}
|
||||
# {
|
||||
# "region": summary,
|
||||
try:
|
||||
regions_with_resources = list(resources_in_region.keys())
|
||||
services = {}
|
||||
# { "S3":
|
||||
# 123,
|
||||
# "IAM":
|
||||
# 239,
|
||||
# }
|
||||
resources_type = {}
|
||||
# { "S3":
|
||||
# "Buckets":
|
||||
# eu-west-1: 10,
|
||||
# eu-west-2: 3,
|
||||
# "IAM":
|
||||
# "Roles":
|
||||
# us-east-1: 143,
|
||||
# "Users":
|
||||
# us-west-2: 22,
|
||||
# }
|
||||
summary = ""
|
||||
inventory_table["Service"].append(f"{service}")
|
||||
inventory_table[
|
||||
f"Total\n({Fore.GREEN}{str(len(resources))}{Style.RESET_ALL})"
|
||||
].append(f"{Fore.GREEN}{services[service]}{Style.RESET_ALL}")
|
||||
for resource_type, regions in resources_type[service].items():
|
||||
summary += f"{resource_type} {Fore.GREEN}{str(sum(regions.values()))}{Style.RESET_ALL}\n"
|
||||
# Check if region does not have resource type
|
||||
for region in pending_regions:
|
||||
if region not in aux:
|
||||
aux[region] = ""
|
||||
if region not in regions:
|
||||
aux[region] += "-\n"
|
||||
for region, count in regions.items():
|
||||
aux[region] += f"{Fore.GREEN}{str(count)}{Style.RESET_ALL}\n"
|
||||
# Add Total per resource type
|
||||
inventory_table["Total per\nresource type"].append(summary)
|
||||
# Add Total per region
|
||||
for region, text in aux.items():
|
||||
inventory_table[
|
||||
f"{region}\n({Fore.GREEN}{str(resources_in_region[region])}{Style.RESET_ALL})"
|
||||
].append(text)
|
||||
if region in pending_regions:
|
||||
pending_regions.remove(region)
|
||||
for region_without_resource in pending_regions:
|
||||
inventory_table[
|
||||
f"{region_without_resource}\n ({Fore.GREEN}{str(resources_in_region[region_without_resource])}{Style.RESET_ALL})"
|
||||
].append("-")
|
||||
|
||||
return inventory_table
|
||||
inventory_table = {
|
||||
"Service": [],
|
||||
f"Total\n({Fore.GREEN}{str(len(resources))}{Style.RESET_ALL})": [],
|
||||
"Total per\nresource type": [],
|
||||
}
|
||||
|
||||
for region, count in resources_in_region.items():
|
||||
inventory_table[
|
||||
f"{region}\n({Fore.GREEN}{str(count)}{Style.RESET_ALL})"
|
||||
] = []
|
||||
|
||||
for resource in sorted(resources, key=lambda d: d["arn"]):
|
||||
service = resource["arn"].split(":")[2]
|
||||
region = resource["arn"].split(":")[3]
|
||||
if not region:
|
||||
region = "global"
|
||||
if service not in services:
|
||||
services[service] = 0
|
||||
services[service] += 1
|
||||
|
||||
resource_type = get_arn_resource_type(resource["arn"], service)
|
||||
|
||||
if service not in resources_type:
|
||||
resources_type[service] = {}
|
||||
if resource_type not in resources_type[service]:
|
||||
resources_type[service][resource_type] = {}
|
||||
if region not in resources_type[service][resource_type]:
|
||||
resources_type[service][resource_type][region] = 0
|
||||
resources_type[service][resource_type][region] += 1
|
||||
|
||||
# Add results to inventory table
|
||||
for service in services:
|
||||
pending_regions = deepcopy(regions_with_resources)
|
||||
aux = {}
|
||||
# {
|
||||
# "region": summary,
|
||||
# }
|
||||
summary = ""
|
||||
inventory_table["Service"].append(f"{service}")
|
||||
inventory_table[
|
||||
f"Total\n({Fore.GREEN}{str(len(resources))}{Style.RESET_ALL})"
|
||||
].append(f"{Fore.GREEN}{services[service]}{Style.RESET_ALL}")
|
||||
for resource_type, regions in resources_type[service].items():
|
||||
summary += f"{resource_type} {Fore.GREEN}{str(sum(regions.values()))}{Style.RESET_ALL}\n"
|
||||
# Check if region does not have resource type
|
||||
for region in pending_regions:
|
||||
if region not in aux:
|
||||
aux[region] = ""
|
||||
if region not in regions:
|
||||
aux[region] += "-\n"
|
||||
for region, count in regions.items():
|
||||
aux[region] += f"{Fore.GREEN}{str(count)}{Style.RESET_ALL}\n"
|
||||
# Add Total per resource type
|
||||
inventory_table["Total per\nresource type"].append(summary)
|
||||
# Add Total per region
|
||||
for region, text in aux.items():
|
||||
inventory_table[
|
||||
f"{region}\n({Fore.GREEN}{str(resources_in_region[region])}{Style.RESET_ALL})"
|
||||
].append(text)
|
||||
if region in pending_regions:
|
||||
pending_regions.remove(region)
|
||||
for region_without_resource in pending_regions:
|
||||
inventory_table[
|
||||
f"{region_without_resource}\n ({Fore.GREEN}{str(resources_in_region[region_without_resource])}{Style.RESET_ALL})"
|
||||
].append("-")
|
||||
|
||||
return inventory_table
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
|
||||
def create_output(resources: list, provider: AwsProvider, args):
|
||||
json_output = []
|
||||
# Check if custom output filename was input, if not, set the default
|
||||
if not hasattr(args, "output_filename") or args.output_filename is None:
|
||||
output_file = (
|
||||
f"prowler-inventory-{provider.identity.account}-{output_file_timestamp}"
|
||||
try:
|
||||
json_output = []
|
||||
# Check if custom output filename was input, if not, set the default
|
||||
if not hasattr(args, "output_filename") or args.output_filename is None:
|
||||
output_file = (
|
||||
f"prowler-inventory-{provider.identity.account}-{output_file_timestamp}"
|
||||
)
|
||||
else:
|
||||
output_file = args.output_filename
|
||||
|
||||
for item in sorted(resources, key=lambda d: d["arn"]):
|
||||
resource = {}
|
||||
resource["AWS_AccountID"] = provider.identity.account
|
||||
resource["AWS_Region"] = item["arn"].split(":")[3]
|
||||
resource["AWS_Partition"] = item["arn"].split(":")[1]
|
||||
resource["AWS_Service"] = item["arn"].split(":")[2]
|
||||
resource["AWS_ResourceType"] = item["arn"].split(":")[5].split("/")[0]
|
||||
resource["AWS_ResourceID"] = ""
|
||||
if len(item["arn"].split("/")) > 1:
|
||||
resource["AWS_ResourceID"] = item["arn"].split("/")[-1]
|
||||
elif len(item["arn"].split(":")) > 6:
|
||||
resource["AWS_ResourceID"] = item["arn"].split(":")[-1]
|
||||
resource["AWS_ResourceARN"] = item["arn"]
|
||||
# Cover S3 case
|
||||
if resource["AWS_Service"] == "s3":
|
||||
resource["AWS_ResourceType"] = "bucket"
|
||||
resource["AWS_ResourceID"] = item["arn"].split(":")[-1]
|
||||
# Cover WAFv2 case
|
||||
if resource["AWS_Service"] == "wafv2":
|
||||
resource["AWS_ResourceType"] = "/".join(
|
||||
item["arn"].split(":")[-1].split("/")[:-2]
|
||||
)
|
||||
resource["AWS_ResourceID"] = "/".join(
|
||||
item["arn"].split(":")[-1].split("/")[2:]
|
||||
)
|
||||
# Cover Config case
|
||||
if resource["AWS_Service"] == "config":
|
||||
resource["AWS_ResourceID"] = "/".join(
|
||||
item["arn"].split(":")[-1].split("/")[1:]
|
||||
)
|
||||
resource["AWS_Tags"] = item["tags"]
|
||||
json_output.append(resource)
|
||||
|
||||
# Serializing json
|
||||
json_object = json.dumps(json_output, indent=4)
|
||||
|
||||
# Writing to sample.json
|
||||
with open(
|
||||
args.output_directory + "/" + output_file + json_file_suffix, "w"
|
||||
) as outfile:
|
||||
outfile.write(json_object)
|
||||
|
||||
csv_file = open(
|
||||
args.output_directory + "/" + output_file + csv_file_suffix, "w", newline=""
|
||||
)
|
||||
else:
|
||||
output_file = args.output_filename
|
||||
csv_writer = csv.writer(csv_file)
|
||||
|
||||
for item in sorted(resources, key=lambda d: d["arn"]):
|
||||
resource = {}
|
||||
resource["AWS_AccountID"] = provider.identity.account
|
||||
resource["AWS_Region"] = item["arn"].split(":")[3]
|
||||
resource["AWS_Partition"] = item["arn"].split(":")[1]
|
||||
resource["AWS_Service"] = item["arn"].split(":")[2]
|
||||
resource["AWS_ResourceType"] = item["arn"].split(":")[5].split("/")[0]
|
||||
resource["AWS_ResourceID"] = ""
|
||||
if len(item["arn"].split("/")) > 1:
|
||||
resource["AWS_ResourceID"] = item["arn"].split("/")[-1]
|
||||
elif len(item["arn"].split(":")) > 6:
|
||||
resource["AWS_ResourceID"] = item["arn"].split(":")[-1]
|
||||
resource["AWS_ResourceARN"] = item["arn"]
|
||||
# Cover S3 case
|
||||
if resource["AWS_Service"] == "s3":
|
||||
resource["AWS_ResourceType"] = "bucket"
|
||||
resource["AWS_ResourceID"] = item["arn"].split(":")[-1]
|
||||
# Cover WAFv2 case
|
||||
if resource["AWS_Service"] == "wafv2":
|
||||
resource["AWS_ResourceType"] = "/".join(
|
||||
item["arn"].split(":")[-1].split("/")[:-2]
|
||||
)
|
||||
resource["AWS_ResourceID"] = "/".join(
|
||||
item["arn"].split(":")[-1].split("/")[2:]
|
||||
)
|
||||
# Cover Config case
|
||||
if resource["AWS_Service"] == "config":
|
||||
resource["AWS_ResourceID"] = "/".join(
|
||||
item["arn"].split(":")[-1].split("/")[1:]
|
||||
)
|
||||
resource["AWS_Tags"] = item["tags"]
|
||||
json_output.append(resource)
|
||||
count = 0
|
||||
for data in json_output:
|
||||
if count == 0:
|
||||
header = data.keys()
|
||||
csv_writer.writerow(header)
|
||||
count += 1
|
||||
csv_writer.writerow(data.values())
|
||||
|
||||
# Serializing json
|
||||
json_object = json.dumps(json_output, indent=4)
|
||||
csv_file.close()
|
||||
print(
|
||||
f"\n{Fore.YELLOW}WARNING: Only resources that have or have had tags will appear (except for IAM and S3).\nSee more in https://docs.prowler.cloud/en/latest/tutorials/quick-inventory/#objections{Style.RESET_ALL}"
|
||||
)
|
||||
print("\nMore details in files:")
|
||||
print(f" - CSV: {args.output_directory}/{output_file + csv_file_suffix}")
|
||||
print(f" - JSON: {args.output_directory}/{output_file + json_file_suffix}")
|
||||
|
||||
# Writing to sample.json
|
||||
with open(
|
||||
args.output_directory + "/" + output_file + json_file_suffix, "w"
|
||||
) as outfile:
|
||||
outfile.write(json_object)
|
||||
# Send output to S3 if needed (-B / -D)
|
||||
|
||||
csv_file = open(
|
||||
args.output_directory + "/" + output_file + csv_file_suffix, "w", newline=""
|
||||
)
|
||||
csv_writer = csv.writer(csv_file)
|
||||
|
||||
count = 0
|
||||
for data in json_output:
|
||||
if count == 0:
|
||||
header = data.keys()
|
||||
csv_writer.writerow(header)
|
||||
count += 1
|
||||
csv_writer.writerow(data.values())
|
||||
|
||||
csv_file.close()
|
||||
print(
|
||||
f"\n{Fore.YELLOW}WARNING: Only resources that have or have had tags will appear (except for IAM and S3).\nSee more in https://docs.prowler.cloud/en/latest/tutorials/quick-inventory/#objections{Style.RESET_ALL}"
|
||||
)
|
||||
print("\nMore details in files:")
|
||||
print(f" - CSV: {args.output_directory}/{output_file + csv_file_suffix}")
|
||||
print(f" - JSON: {args.output_directory}/{output_file + json_file_suffix}")
|
||||
|
||||
# Send output to S3 if needed (-B / -D)
|
||||
for mode in ["csv"]:
|
||||
if args.output_bucket or args.output_bucket_no_assume:
|
||||
# Check if -B was input
|
||||
if args.output_bucket:
|
||||
@@ -294,14 +310,29 @@ def create_output(resources: list, provider: AwsProvider, args):
|
||||
# Check if -D was input
|
||||
elif args.output_bucket_no_assume:
|
||||
output_bucket = args.output_bucket_no_assume
|
||||
bucket_session = provider.original_session
|
||||
send_to_s3_bucket(
|
||||
output_file,
|
||||
args.output_directory,
|
||||
mode,
|
||||
bucket_session = provider.session.original_session
|
||||
|
||||
s3_client = bucket_session.client("s3")
|
||||
# FIXME: Use get_object_path method from S3 class when quick inventory uses S3 class
|
||||
bucket_remote_dir = args.output_directory
|
||||
if "prowler/" in bucket_remote_dir: # Check if it is not a custom directory
|
||||
bucket_remote_dir = bucket_remote_dir.partition("prowler/")[-1]
|
||||
# CSV
|
||||
s3_client.upload_file(
|
||||
f"{args.output_directory}/{output_file + csv_file_suffix}",
|
||||
output_bucket,
|
||||
bucket_session,
|
||||
f"{bucket_remote_dir}/{output_file + csv_file_suffix}",
|
||||
)
|
||||
# JSON
|
||||
s3_client.upload_file(
|
||||
f"{args.output_directory}/{output_file + json_file_suffix}",
|
||||
output_bucket,
|
||||
f"{bucket_remote_dir}/{output_file + json_file_suffix}",
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
|
||||
def get_regional_buckets(provider: AwsProvider, region: str) -> list:
|
||||
|
||||
@@ -1,51 +1,149 @@
|
||||
from prowler.config.config import (
|
||||
available_output_formats,
|
||||
csv_file_suffix,
|
||||
html_file_suffix,
|
||||
json_asff_file_suffix,
|
||||
json_ocsf_file_suffix,
|
||||
)
|
||||
from os import path
|
||||
from tempfile import NamedTemporaryFile
|
||||
|
||||
from boto3 import Session
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.lib.outputs.output import Output
|
||||
|
||||
|
||||
def send_to_s3_bucket(
|
||||
output_filename, output_directory, output_mode, output_bucket_name, audit_session
|
||||
):
|
||||
try:
|
||||
# S3 Object name
|
||||
bucket_directory = get_s3_object_path(output_directory)
|
||||
filename = ""
|
||||
# Get only last part of the path
|
||||
if output_mode in available_output_formats:
|
||||
if output_mode == "csv":
|
||||
filename = f"{output_filename}{csv_file_suffix}"
|
||||
elif output_mode == "json-asff":
|
||||
filename = f"{output_filename}{json_asff_file_suffix}"
|
||||
elif output_mode == "json-ocsf":
|
||||
filename = f"{output_filename}{json_ocsf_file_suffix}"
|
||||
elif output_mode == "html":
|
||||
filename = f"{output_filename}{html_file_suffix}"
|
||||
file_name = output_directory + "/" + filename
|
||||
object_name = bucket_directory + "/" + output_mode + "/" + filename
|
||||
else: # Compliance output mode
|
||||
filename = f"{output_filename}_{output_mode}{csv_file_suffix}"
|
||||
file_name = output_directory + "/compliance/" + filename
|
||||
object_name = bucket_directory + "/compliance/" + filename
|
||||
class S3:
|
||||
"""
|
||||
A class representing an S3 bucket.
|
||||
|
||||
logger.info(f"Sending output file {filename} to S3 bucket {output_bucket_name}")
|
||||
Attributes:
|
||||
- _session: An instance of the `Session` class representing the AWS session.
|
||||
- _bucket_name: A string representing the name of the S3 bucket.
|
||||
- _output_directory: A string representing the output directory path.
|
||||
|
||||
s3_client = audit_session.client("s3")
|
||||
s3_client.upload_file(file_name, output_bucket_name, object_name)
|
||||
Methods:
|
||||
- __init__: Initializes a new instance of the `S3` class.
|
||||
- get_object_path: Returns the object path within the S3 bucket based on the provided output directory.
|
||||
- generate_subfolder_name_by_extension: Generates a subfolder name based on the provided file extension.
|
||||
- send_to_bucket: Sends the provided outputs to the S3 bucket.
|
||||
"""
|
||||
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
|
||||
)
|
||||
_session: Session
|
||||
_bucket_name: str
|
||||
_output_directory: str
|
||||
|
||||
def __init__(
|
||||
self, session: Session, bucket_name: str, output_directory: str
|
||||
) -> None:
|
||||
"""
|
||||
Initializes a new instance of the `S3` class.
|
||||
|
||||
def get_s3_object_path(output_directory: str) -> str:
|
||||
bucket_remote_dir = output_directory
|
||||
if "prowler/" in bucket_remote_dir: # Check if it is not a custom directory
|
||||
bucket_remote_dir = bucket_remote_dir.partition("prowler/")[-1]
|
||||
Parameters:
|
||||
- session: An instance of the `Session` class representing the AWS session.
|
||||
- bucket_name: A string representing the name of the S3 bucket.
|
||||
- output_directory: A string representing the output directory path.
|
||||
"""
|
||||
self._session = session.client(__class__.__name__.lower())
|
||||
self._bucket_name = bucket_name
|
||||
self._output_directory = output_directory
|
||||
|
||||
return bucket_remote_dir
|
||||
@staticmethod
|
||||
def get_object_path(output_directory: str) -> str:
|
||||
"""
|
||||
Return the object path within the S3 bucket based on the provided output directory.
|
||||
If the output directory contains "prowler/", it is removed to ensure the correct path is returned.
|
||||
|
||||
Parameters:
|
||||
- output_directory: A string representing the output directory path.
|
||||
|
||||
Returns:
|
||||
- A string representing the object path within the S3 bucket.
|
||||
"""
|
||||
bucket_remote_dir = output_directory
|
||||
if "prowler/" in bucket_remote_dir: # Check if it is not a custom directory
|
||||
bucket_remote_dir = bucket_remote_dir.partition("prowler/")[-1]
|
||||
|
||||
return bucket_remote_dir
|
||||
|
||||
@staticmethod
|
||||
def generate_subfolder_name_by_extension(extension: str) -> str:
|
||||
"""
|
||||
Generate a subfolder name based on the provided file extension.
|
||||
|
||||
Parameters:
|
||||
- extension: A string representing the file extension.
|
||||
|
||||
Returns:
|
||||
- A string representing the subfolder name based on the extension.
|
||||
"""
|
||||
subfolder_name = ""
|
||||
if extension == ".ocsf.json":
|
||||
subfolder_name = "json-ocsf"
|
||||
elif extension == ".asff.json":
|
||||
subfolder_name = "json-asff"
|
||||
else:
|
||||
subfolder_name = extension.lstrip(".")
|
||||
return subfolder_name
|
||||
|
||||
# TODO: Review the logic behind in Microsoft Windows
|
||||
def send_to_bucket(
|
||||
self, outputs: dict[str, list[Output]]
|
||||
) -> dict[str, dict[str, list[str]]]:
|
||||
"""
|
||||
Send the provided outputs to the S3 bucket.
|
||||
|
||||
Parameters:
|
||||
- outputs: A dictionary where keys are strings and values are lists of Output objects.
|
||||
|
||||
Returns:
|
||||
- A dictionary containing two keys: "success" and "failure", each holding a dictionary where keys are strings and values are lists of strings representing the uploaded object names or tuples of object names and errors respectively.
|
||||
"""
|
||||
try:
|
||||
uploaded_objects = {"success": {}, "failure": {}}
|
||||
# Keys are regular and/or compliance
|
||||
for key, output_list in outputs.items():
|
||||
for output in output_list:
|
||||
try:
|
||||
# Object is not written to file so we need to temporarily write it
|
||||
if not hasattr(output, "file_descriptor"):
|
||||
output.file_descriptor = NamedTemporaryFile(mode="a")
|
||||
|
||||
bucket_directory = self.get_object_path(self._output_directory)
|
||||
basename = path.basename(output.file_descriptor.name)
|
||||
|
||||
if key == "compliance":
|
||||
object_name = f"{bucket_directory}/{key}/{basename}"
|
||||
else:
|
||||
object_name = f"{bucket_directory}/{self.generate_subfolder_name_by_extension(output.file_extension)}/{basename}"
|
||||
logger.info(
|
||||
f"Sending output file {output.file_descriptor.name} to S3 bucket {self._bucket_name}"
|
||||
)
|
||||
|
||||
# TODO: This will need further optimization if some processes are calling this since the files are written
|
||||
# into the local filesystem because S3 upload file is the recommended way.
|
||||
# https://aws.amazon.com/blogs/developer/uploading-files-to-amazon-s3/
|
||||
self._session.upload_file(
|
||||
output.file_descriptor.name, self._bucket_name, object_name
|
||||
)
|
||||
|
||||
if output.file_extension in uploaded_objects["success"]:
|
||||
uploaded_objects["success"][output.file_extension].append(
|
||||
object_name
|
||||
)
|
||||
else:
|
||||
uploaded_objects["success"] = {
|
||||
output.file_extension: [object_name]
|
||||
}
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
|
||||
)
|
||||
if output.file_extension in uploaded_objects["failure"]:
|
||||
uploaded_objects["failure"][output.file_extension].append(
|
||||
(object_name, error)
|
||||
)
|
||||
else:
|
||||
uploaded_objects["failure"] = {
|
||||
output.file_extension: [(object_name, error)]
|
||||
}
|
||||
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
|
||||
)
|
||||
return uploaded_objects
|
||||
|
||||
|
|
@@ -1 +0,0 @@
|
||||
{}
|
||||
@@ -1 +0,0 @@
|
||||
{}
|
||||
@@ -1,236 +1,316 @@
|
||||
from os import path
|
||||
from os import path, remove
|
||||
from pathlib import Path
|
||||
|
||||
import boto3
|
||||
from mock import MagicMock
|
||||
from moto import mock_aws
|
||||
|
||||
from prowler.config.config import (
|
||||
csv_file_suffix,
|
||||
html_file_suffix,
|
||||
json_asff_file_suffix,
|
||||
json_ocsf_file_suffix,
|
||||
)
|
||||
from prowler.providers.aws.lib.s3.s3 import get_s3_object_path, send_to_s3_bucket
|
||||
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1
|
||||
from prowler.lib.outputs.compliance.iso27001.iso27001_aws import AWSISO27001
|
||||
from prowler.lib.outputs.csv.csv import CSV
|
||||
from prowler.lib.outputs.html.html import HTML
|
||||
from prowler.lib.outputs.ocsf.ocsf import OCSF
|
||||
from prowler.providers.aws.lib.s3.s3 import S3
|
||||
from tests.lib.outputs.compliance.fixtures import ISO27001_2013_AWS
|
||||
from tests.lib.outputs.fixtures.fixtures import generate_finding_output
|
||||
from tests.providers.aws.utils import AWS_REGION_US_EAST_1
|
||||
|
||||
ACTUAL_DIRECTORY = Path(path.dirname(path.realpath(__file__)))
|
||||
FIXTURES_DIR_NAME = "fixtures"
|
||||
CURRENT_DIRECTORY = str(Path(path.dirname(path.realpath(__file__))))
|
||||
S3_BUCKET_NAME = "test_bucket"
|
||||
OUTPUT_MODE_CSV = "csv"
|
||||
OUTPUT_MODE_JSON_OCSF = "json-ocsf"
|
||||
OUTPUT_MODE_JSON_ASFF = "json-asff"
|
||||
OUTPUT_MODE_HTML = "html"
|
||||
OUTPUT_MODE_CIS_1_4_AWS = "cis_1.4_aws"
|
||||
FINDING = generate_finding_output(
|
||||
status="PASS",
|
||||
status_extended="status-extended",
|
||||
resource_uid="resource-123",
|
||||
resource_name="Example Resource",
|
||||
resource_details="Detailed information about the resource",
|
||||
resource_tags="tag1,tag2",
|
||||
partition="aws",
|
||||
description="Description of the finding",
|
||||
risk="High",
|
||||
related_url="http://example.com",
|
||||
remediation_recommendation_text="Recommendation text",
|
||||
remediation_recommendation_url="http://example.com/remediation",
|
||||
remediation_code_nativeiac="native-iac-code",
|
||||
remediation_code_terraform="terraform-code",
|
||||
remediation_code_other="other-code",
|
||||
remediation_code_cli="cli-code",
|
||||
compliance={"compliance_key": "compliance_value"},
|
||||
categories="category1,category2",
|
||||
depends_on="dependency",
|
||||
related_to="related finding",
|
||||
notes="Notes about the finding",
|
||||
)
|
||||
|
||||
|
||||
class TestS3:
|
||||
@mock_aws
|
||||
def test_send_no_outputs(self):
|
||||
current_session = boto3.session.Session(region_name=AWS_REGION_US_EAST_1)
|
||||
s3 = S3(
|
||||
session=current_session,
|
||||
bucket_name=S3_BUCKET_NAME,
|
||||
output_directory=CURRENT_DIRECTORY,
|
||||
)
|
||||
assert s3.send_to_bucket({}) == {"success": {}, "failure": {}}
|
||||
|
||||
@mock_aws
|
||||
def test_send_to_s3_bucket_csv(self):
|
||||
# Mock Audit Info
|
||||
provider = MagicMock()
|
||||
|
||||
# Create mock session
|
||||
provider.current_session = boto3.session.Session(
|
||||
region_name=AWS_REGION_US_EAST_1
|
||||
)
|
||||
provider.identity.account = AWS_ACCOUNT_NUMBER
|
||||
|
||||
# Create mock bucket
|
||||
client = provider.current_session.client("s3")
|
||||
# Create bucket
|
||||
current_session = boto3.session.Session(region_name=AWS_REGION_US_EAST_1)
|
||||
client = current_session.client("s3")
|
||||
client.create_bucket(Bucket=S3_BUCKET_NAME)
|
||||
|
||||
# Mocked CSV output file
|
||||
output_directory = f"{ACTUAL_DIRECTORY}/{FIXTURES_DIR_NAME}"
|
||||
filename = f"prowler-output-{provider.identity.account}"
|
||||
|
||||
# Send mock CSV file to mock S3 Bucket
|
||||
send_to_s3_bucket(
|
||||
filename,
|
||||
output_directory,
|
||||
OUTPUT_MODE_CSV,
|
||||
S3_BUCKET_NAME,
|
||||
provider.current_session,
|
||||
s3 = S3(
|
||||
session=current_session,
|
||||
bucket_name=S3_BUCKET_NAME,
|
||||
output_directory=CURRENT_DIRECTORY,
|
||||
)
|
||||
|
||||
bucket_directory = get_s3_object_path(output_directory)
|
||||
object_name = (
|
||||
f"{bucket_directory}/{OUTPUT_MODE_CSV}/{filename}{csv_file_suffix}"
|
||||
extension = ".csv"
|
||||
csv = CSV(
|
||||
findings=[FINDING],
|
||||
file_extension=extension,
|
||||
)
|
||||
|
||||
s3_send_result = s3.send_to_bucket(outputs={"regular": [csv]})
|
||||
|
||||
assert "failure" in s3_send_result
|
||||
assert s3_send_result["failure"] == {}
|
||||
|
||||
assert "success" in s3_send_result
|
||||
assert extension in s3_send_result["success"]
|
||||
assert len(s3_send_result["success"][extension]) == 1
|
||||
|
||||
uploaded_object_name = s3_send_result["success"][extension][0]
|
||||
|
||||
assert (
|
||||
client.get_object(
|
||||
Bucket=S3_BUCKET_NAME,
|
||||
Key=object_name,
|
||||
Key=uploaded_object_name,
|
||||
)["ContentType"]
|
||||
== "binary/octet-stream"
|
||||
)
|
||||
|
||||
@mock_aws
|
||||
def test_send_to_s3_bucket_json_ocsf(self):
|
||||
# Mock Audit Info
|
||||
provider = MagicMock()
|
||||
|
||||
# Create mock session
|
||||
provider.current_session = boto3.session.Session(
|
||||
region_name=AWS_REGION_US_EAST_1
|
||||
)
|
||||
provider.identity.account = AWS_ACCOUNT_NUMBER
|
||||
|
||||
# Create mock bucket
|
||||
client = provider.current_session.client("s3")
|
||||
def test_send_to_s3_bucket_csv_with_file_descriptor(self):
|
||||
# Create bucket
|
||||
current_session = boto3.session.Session(region_name=AWS_REGION_US_EAST_1)
|
||||
client = current_session.client("s3")
|
||||
client.create_bucket(Bucket=S3_BUCKET_NAME)
|
||||
|
||||
# Mocked CSV output file
|
||||
output_directory = f"{ACTUAL_DIRECTORY}/{FIXTURES_DIR_NAME}"
|
||||
filename = f"prowler-output-{provider.identity.account}"
|
||||
|
||||
# Send mock CSV file to mock S3 Bucket
|
||||
send_to_s3_bucket(
|
||||
filename,
|
||||
output_directory,
|
||||
OUTPUT_MODE_JSON_OCSF,
|
||||
S3_BUCKET_NAME,
|
||||
provider.current_session,
|
||||
s3 = S3(
|
||||
session=current_session,
|
||||
bucket_name=S3_BUCKET_NAME,
|
||||
output_directory=CURRENT_DIRECTORY,
|
||||
)
|
||||
|
||||
bucket_directory = get_s3_object_path(output_directory)
|
||||
object_name = f"{bucket_directory}/{OUTPUT_MODE_JSON_OCSF}/{filename}{json_ocsf_file_suffix}"
|
||||
extension = ".csv"
|
||||
csv_file = f"test{extension}"
|
||||
csv = CSV(
|
||||
findings=[FINDING],
|
||||
create_file_descriptor=True,
|
||||
file_path=f"{CURRENT_DIRECTORY}/{csv_file}",
|
||||
)
|
||||
|
||||
s3_send_result = s3.send_to_bucket(outputs={"regular": [csv]})
|
||||
|
||||
assert "failure" in s3_send_result
|
||||
assert s3_send_result["failure"] == {}
|
||||
|
||||
assert "success" in s3_send_result
|
||||
assert extension in s3_send_result["success"]
|
||||
assert len(s3_send_result["success"][extension]) == 1
|
||||
|
||||
uploaded_object_name = s3_send_result["success"][extension][0]
|
||||
|
||||
assert (
|
||||
client.get_object(
|
||||
Bucket=S3_BUCKET_NAME,
|
||||
Key=object_name,
|
||||
Key=uploaded_object_name,
|
||||
)["ContentType"]
|
||||
== "binary/octet-stream"
|
||||
)
|
||||
|
||||
remove(f"{CURRENT_DIRECTORY}/{csv_file}")
|
||||
|
||||
@mock_aws
|
||||
def test_send_to_s3_bucket_json_asff(self):
|
||||
# Mock Audit Info
|
||||
provider = MagicMock()
|
||||
|
||||
# Create mock session
|
||||
provider.current_session = boto3.session.Session(
|
||||
region_name=AWS_REGION_US_EAST_1
|
||||
)
|
||||
provider.identity.account = AWS_ACCOUNT_NUMBER
|
||||
|
||||
# Create mock bucket
|
||||
client = provider.current_session.client("s3")
|
||||
def test_send_to_s3_bucket_ocsf(self):
|
||||
# Create bucket
|
||||
current_session = boto3.session.Session(region_name=AWS_REGION_US_EAST_1)
|
||||
client = current_session.client("s3")
|
||||
client.create_bucket(Bucket=S3_BUCKET_NAME)
|
||||
|
||||
# Mocked CSV output file
|
||||
output_directory = f"{ACTUAL_DIRECTORY}/{FIXTURES_DIR_NAME}"
|
||||
filename = f"prowler-output-{provider.identity.account}"
|
||||
|
||||
# Send mock CSV file to mock S3 Bucket
|
||||
send_to_s3_bucket(
|
||||
filename,
|
||||
output_directory,
|
||||
OUTPUT_MODE_JSON_ASFF,
|
||||
S3_BUCKET_NAME,
|
||||
provider.current_session,
|
||||
s3 = S3(
|
||||
session=current_session,
|
||||
bucket_name=S3_BUCKET_NAME,
|
||||
output_directory=CURRENT_DIRECTORY,
|
||||
)
|
||||
extension = ".ocsf.json"
|
||||
csv = OCSF(
|
||||
findings=[FINDING],
|
||||
file_extension=extension,
|
||||
)
|
||||
|
||||
bucket_directory = get_s3_object_path(output_directory)
|
||||
object_name = f"{bucket_directory}/{OUTPUT_MODE_JSON_ASFF}/{filename}{json_asff_file_suffix}"
|
||||
s3_send_result = s3.send_to_bucket(outputs={"regular": [csv]})
|
||||
|
||||
assert "failure" in s3_send_result
|
||||
assert s3_send_result["failure"] == {}
|
||||
|
||||
assert "success" in s3_send_result
|
||||
assert extension in s3_send_result["success"]
|
||||
assert len(s3_send_result["success"][extension]) == 1
|
||||
|
||||
uploaded_object_name = s3_send_result["success"][extension][0]
|
||||
|
||||
assert (
|
||||
client.get_object(
|
||||
Bucket=S3_BUCKET_NAME,
|
||||
Key=object_name,
|
||||
Key=uploaded_object_name,
|
||||
)["ContentType"]
|
||||
== "binary/octet-stream"
|
||||
)
|
||||
|
||||
@mock_aws
|
||||
def test_send_to_s3_bucket_html(self):
|
||||
# Mock Audit Info
|
||||
provider = MagicMock()
|
||||
|
||||
# Create mock session
|
||||
provider.current_session = boto3.session.Session(
|
||||
region_name=AWS_REGION_US_EAST_1
|
||||
)
|
||||
provider.identity.account = AWS_ACCOUNT_NUMBER
|
||||
|
||||
# Create mock bucket
|
||||
client = provider.current_session.client("s3")
|
||||
# Create bucket
|
||||
current_session = boto3.session.Session(region_name=AWS_REGION_US_EAST_1)
|
||||
client = current_session.client("s3")
|
||||
client.create_bucket(Bucket=S3_BUCKET_NAME)
|
||||
|
||||
# Mocked CSV output file
|
||||
output_directory = f"{ACTUAL_DIRECTORY}/{FIXTURES_DIR_NAME}"
|
||||
filename = f"prowler-output-{provider.identity.account}"
|
||||
|
||||
# Send mock CSV file to mock S3 Bucket
|
||||
send_to_s3_bucket(
|
||||
filename,
|
||||
output_directory,
|
||||
OUTPUT_MODE_HTML,
|
||||
S3_BUCKET_NAME,
|
||||
provider.current_session,
|
||||
s3 = S3(
|
||||
session=current_session,
|
||||
bucket_name=S3_BUCKET_NAME,
|
||||
output_directory=CURRENT_DIRECTORY,
|
||||
)
|
||||
|
||||
bucket_directory = get_s3_object_path(output_directory)
|
||||
object_name = (
|
||||
f"{bucket_directory}/{OUTPUT_MODE_HTML}/{filename}{html_file_suffix}"
|
||||
extension = ".html"
|
||||
csv = HTML(
|
||||
findings=[FINDING],
|
||||
file_extension=extension,
|
||||
)
|
||||
|
||||
s3_send_result = s3.send_to_bucket(outputs={"regular": [csv]})
|
||||
|
||||
assert "failure" in s3_send_result
|
||||
assert s3_send_result["failure"] == {}
|
||||
|
||||
assert "success" in s3_send_result
|
||||
assert extension in s3_send_result["success"]
|
||||
assert len(s3_send_result["success"][extension]) == 1
|
||||
|
||||
uploaded_object_name = s3_send_result["success"][extension][0]
|
||||
|
||||
assert (
|
||||
client.get_object(
|
||||
Bucket=S3_BUCKET_NAME,
|
||||
Key=object_name,
|
||||
Key=uploaded_object_name,
|
||||
)["ContentType"]
|
||||
== "binary/octet-stream"
|
||||
)
|
||||
|
||||
@mock_aws
|
||||
def test_send_to_s3_bucket_compliance(self):
|
||||
# Mock Audit Info
|
||||
provider = MagicMock()
|
||||
def test_send_to_s3_non_existent_bucket(self):
|
||||
# Create bucket
|
||||
current_session = boto3.session.Session(region_name=AWS_REGION_US_EAST_1)
|
||||
|
||||
# Create mock session
|
||||
provider.current_session = boto3.session.Session(
|
||||
region_name=AWS_REGION_US_EAST_1
|
||||
s3 = S3(
|
||||
session=current_session,
|
||||
bucket_name=S3_BUCKET_NAME,
|
||||
output_directory=CURRENT_DIRECTORY,
|
||||
)
|
||||
provider.identity.account = AWS_ACCOUNT_NUMBER
|
||||
|
||||
# Create mock bucket
|
||||
client = provider.current_session.client("s3")
|
||||
extension = ".csv"
|
||||
csv = CSV(
|
||||
findings=[FINDING],
|
||||
file_extension=extension,
|
||||
)
|
||||
|
||||
s3_send_result = s3.send_to_bucket(outputs={"regular": [csv]})
|
||||
|
||||
assert "success" in s3_send_result
|
||||
assert s3_send_result["success"] == {}
|
||||
|
||||
assert "failure" in s3_send_result
|
||||
assert extension in s3_send_result["failure"]
|
||||
assert len(s3_send_result["failure"][extension])
|
||||
|
||||
assert isinstance(s3_send_result["failure"][extension], list)
|
||||
assert len(s3_send_result["failure"][extension]) == 1
|
||||
|
||||
assert isinstance(s3_send_result["failure"][extension][0], tuple)
|
||||
|
||||
# Object name
|
||||
assert isinstance(s3_send_result["failure"][extension][0][0], str)
|
||||
assert (
|
||||
f"tests/providers/aws/lib/s3/csv/{path.basename(csv.file_descriptor.name)}"
|
||||
in s3_send_result["failure"][extension][0][0]
|
||||
)
|
||||
# Error
|
||||
assert isinstance(s3_send_result["failure"][extension][0][1], Exception)
|
||||
assert (
|
||||
"An error occurred (NoSuchBucket) when calling the PutObject operation: The specified bucket does not exist"
|
||||
in str(s3_send_result["failure"][extension][0][1])
|
||||
)
|
||||
|
||||
@mock_aws
|
||||
def test_send_to_s3_bucket_compliance_iso_27001(self):
|
||||
# Create bucket
|
||||
current_session = boto3.session.Session(region_name=AWS_REGION_US_EAST_1)
|
||||
client = current_session.client("s3")
|
||||
client.create_bucket(Bucket=S3_BUCKET_NAME)
|
||||
|
||||
# Mocked CSV output file
|
||||
output_directory = f"{ACTUAL_DIRECTORY}/{FIXTURES_DIR_NAME}"
|
||||
filename = f"prowler-output-{provider.identity.account}"
|
||||
|
||||
# Send mock CSV file to mock S3 Bucket
|
||||
send_to_s3_bucket(
|
||||
filename,
|
||||
output_directory,
|
||||
OUTPUT_MODE_CIS_1_4_AWS,
|
||||
S3_BUCKET_NAME,
|
||||
provider.current_session,
|
||||
s3 = S3(
|
||||
session=current_session,
|
||||
bucket_name=S3_BUCKET_NAME,
|
||||
output_directory=CURRENT_DIRECTORY,
|
||||
)
|
||||
|
||||
bucket_directory = get_s3_object_path(output_directory)
|
||||
object_name = f"{bucket_directory}/compliance/{filename}_{OUTPUT_MODE_CIS_1_4_AWS}{csv_file_suffix}"
|
||||
extension = ".csv"
|
||||
compliance = AWSISO27001(
|
||||
findings=[FINDING], compliance=ISO27001_2013_AWS, file_extension=extension
|
||||
)
|
||||
|
||||
s3_send_result = s3.send_to_bucket(outputs={"compliance": [compliance]})
|
||||
|
||||
assert "failure" in s3_send_result
|
||||
assert s3_send_result["failure"] == {}
|
||||
|
||||
assert "success" in s3_send_result
|
||||
assert extension in s3_send_result["success"]
|
||||
assert len(s3_send_result["success"][extension]) == 1
|
||||
|
||||
uploaded_object_name = s3_send_result["success"][extension][0]
|
||||
|
||||
assert (
|
||||
client.get_object(
|
||||
Bucket=S3_BUCKET_NAME,
|
||||
Key=object_name,
|
||||
Key=uploaded_object_name,
|
||||
)["ContentType"]
|
||||
== "binary/octet-stream"
|
||||
)
|
||||
|
||||
def test_get_s3_object_path_with_prowler(self):
|
||||
def test_get_get_object_path_with_prowler(self):
|
||||
output_directory = "/Users/admin/prowler/"
|
||||
assert (
|
||||
get_s3_object_path(output_directory)
|
||||
S3.get_object_path(output_directory)
|
||||
== output_directory.partition("prowler/")[-1]
|
||||
)
|
||||
|
||||
def test_get_s3_object_path_without_prowler(self):
|
||||
def test_get_get_object_path_without_prowler(self):
|
||||
output_directory = "/Users/admin/"
|
||||
assert get_s3_object_path(output_directory) == output_directory
|
||||
assert S3.get_object_path(output_directory) == output_directory
|
||||
|
||||
def test_generate_subfolder_name_by_extension_csv(self):
|
||||
assert S3.generate_subfolder_name_by_extension(".csv") == "csv"
|
||||
|
||||
def test_generate_subfolder_name_by_extension_html(self):
|
||||
assert S3.generate_subfolder_name_by_extension(".html") == "html"
|
||||
|
||||
def test_generate_subfolder_name_by_extension_json_asff(self):
|
||||
assert S3.generate_subfolder_name_by_extension(".asff.json") == "json-asff"
|
||||
|
||||
def test_generate_subfolder_name_by_extension_json_ocsf(self):
|
||||
assert S3.generate_subfolder_name_by_extension(".ocsf.json") == "json-ocsf"
|
||||
|
||||
Reference in New Issue
Block a user