ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 21:11:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(providers): Store output options and mutelist (#3497)

Browse Source
This commit is contained in:
Pepe Fagoaga
2024-03-05 09:56:30 +01:00
committed by GitHub
parent 3385b630e7
commit da1084907e
17 changed files with 416 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/__main__.py
+23 -19
View File
@@ -45,7 +45,6 @@ from prowler.providers.aws.lib.security_hub.security_hub import (
verify_security_hub_integration_enabled_per_region,
)
from prowler.providers.common.common import set_global_provider_object
from prowler.providers.common.outputs import set_provider_output_options
def prowler():
@@ -174,16 +173,14 @@ def prowler():
# Sort final check list
checks_to_execute = sorted(checks_to_execute)
# Parse Mute List
mutelist_file = ""
# Setup Mute List
# TODO: this should be available for all the providers
# Move the argument to the Prowler level to be available for all
if hasattr(args, "mutelist_file"):
mutelist_file = global_provider.get_mutelist(args.mutelist_file)
global_provider.mutelist = args.mutelist_file
# Set output options based on the selected provider
# TODO: this is going to be removed an include in the Provider as a new common object
audit_output_options = set_provider_output_options(
provider, args, global_provider.identity, mutelist_file, bulk_checks_metadata
)
# Setup Output Options
global_provider.output_options = (args, bulk_checks_metadata)
# TODO: adapt the quick inventory for the new AWS provider
# Run the quick inventory for the provider if available
@@ -198,7 +195,6 @@ def prowler():
findings = execute_checks(
checks_to_execute,
global_provider,
audit_output_options,
custom_checks_metadata,
)
else:
@@ -230,14 +226,19 @@ def prowler():
# Close json file if exists
if "json" in mode:
close_json(
audit_output_options.output_filename, args.output_directory, mode
global_provider.output_options.output_filename,
args.output_directory,
mode,
)
if mode == "html":
add_html_footer(
audit_output_options.output_filename, args.output_directory
global_provider.output_options.output_filename,
args.output_directory,
)
fill_html_overview_statistics(
stats, audit_output_options.output_filename, args.output_directory
stats,
global_provider.output_options.output_filename,
args.output_directory,
)
# Send output to S3 if needed (-B / -D)
if provider == "aws" and (
@@ -250,7 +251,7 @@ def prowler():
output_bucket = args.output_bucket_no_assume
bucket_session = global_provider.session.original_session
send_to_s3_bucket(
audit_output_options.output_filename,
global_provider.output_options.output_filename,
args.output_directory,
mode,
output_bucket,
@@ -281,7 +282,10 @@ def prowler():
# Prepare the findings to be sent to Security Hub
security_hub_findings_per_region = prepare_security_hub_findings(
findings, provider, audit_output_options, aws_security_enabled_regions
findings,
provider,
global_provider.output_options,
aws_security_enabled_regions,
)
# Send the findings to Security Hub
@@ -311,7 +315,7 @@ def prowler():
display_summary_table(
findings,
global_provider,
audit_output_options,
global_provider.output_options,
)
if findings:
@@ -325,13 +329,13 @@ def prowler():
findings,
bulk_checks_metadata,
compliance,
audit_output_options.output_filename,
audit_output_options.output_directory,
global_provider.output_options.output_filename,
global_provider.output_options.output_directory,
compliance_overview,
)
if compliance_overview:
print(
f"\nDetailed compliance results are in {Fore.YELLOW}{audit_output_options.output_directory}/compliance/{Style.RESET_ALL}\n"
f"\nDetailed compliance results are in {Fore.YELLOW}{global_provider.output_options.output_directory}/compliance/{Style.RESET_ALL}\n"
)
# If custom checks were passed, remove the modules
prowler/lib/check/check.py
+9 -13
View File
@@ -24,7 +24,6 @@ from prowler.lib.utils.utils import open_file, parse_json_file
from prowler.providers.aws.lib.mutelist.mutelist import mutelist_findings
from prowler.providers.common.common import get_global_provider
from prowler.providers.common.models import Audit_Metadata
from prowler.providers.common.outputs import Provider_Output_Options
# Load all checks metadata
@@ -398,7 +397,7 @@ def import_check(check_path: str) -> ModuleType:
return lib
def run_check(check: Check, output_options: Provider_Output_Options) -> list:
def run_check(check: Check, output_options) -> list:
findings = []
if output_options.verbose:
print(
@@ -422,7 +421,6 @@ def run_check(check: Check, output_options: Provider_Output_Options) -> list:
def execute_checks(
checks_to_execute: list,
global_provider: Any,
audit_output_options: Provider_Output_Options,
custom_checks_metadata: Any,
) -> list:
# List to store all the check's findings
@@ -460,7 +458,7 @@ def execute_checks(
)
# Execution with the --only-logs flag
if audit_output_options.only_logs:
if global_provider.output_options.only_logs:
for check_name in checks_to_execute:
# Recover service from check name
service = check_name.split("_")[0]
@@ -469,7 +467,6 @@ def execute_checks(
service,
check_name,
global_provider.type,
audit_output_options,
global_provider.identity,
services_executed,
checks_executed,
@@ -514,7 +511,6 @@ def execute_checks(
check_findings = execute(
service,
check_name,
audit_output_options,
global_provider,
services_executed,
checks_executed,
@@ -541,7 +537,6 @@ def execute_checks(
def execute(
service: str,
check_name: str,
audit_output_options: Provider_Output_Options,
global_provider: Any,
services_executed: set,
checks_executed: set,
@@ -559,7 +554,7 @@ def execute(
c = update_check_metadata(c, custom_checks_metadata["Checks"][c.CheckID])
# Run check
check_findings = run_check(c, audit_output_options)
check_findings = run_check(c, global_provider.output_options)
# Update Audit Status
services_executed.add(service)
@@ -569,15 +564,15 @@ def execute(
)
# Mute List findings
if audit_output_options.mutelist_file:
if hasattr(global_provider, "mutelist") and global_provider.mutelist:
check_findings = mutelist_findings(
audit_output_options.mutelist_file,
global_provider.audited_account,
global_provider.mutelist,
global_provider.identity.account,
check_findings,
)
# Report the check's findings
report(check_findings, audit_output_options, global_provider)
report(check_findings, global_provider)
if os.environ.get("PROWLER_REPORT_LIB_PATH"):
try:
@@ -586,8 +581,9 @@ def execute(
outputs_module = importlib.import_module(lib)
custom_report_interface = getattr(outputs_module, "report")
# TODO: review this call and see if we can remove the global_provider.output_options since it is contained in the global_provider
custom_report_interface(
check_findings, audit_output_options, global_provider
check_findings, global_provider.output_options, global_provider
)
except Exception:
sys.exit(1)
prowler/lib/outputs/models.py
+1 -1
View File
@@ -48,7 +48,7 @@ def get_check_compliance(finding, provider_type, output_options) -> dict:
def generate_provider_output_csv(provider, finding, mode: str, fd, output_options):
"""
set_provider_output_options configures automatically the outputs based on the selected provider and returns the Provider_Output_Options object.
generate_provider_output_csv creates the provider's CSV output
"""
try:
# Dynamically load the Provider_Output_Options class
prowler/lib/outputs/outputs.py
+2 -1
View File
@@ -34,8 +34,9 @@ def stdout_report(finding, color, verbose, status):
)
def report(check_findings, output_options, provider):
def report(check_findings, provider):
try:
output_options = provider.output_options
file_descriptors = {}
if check_findings:
# TO-DO Generic Function
prowler/lib/outputs/summary_table.py
+1 -2
View File
@@ -11,13 +11,12 @@ from prowler.config.config import (
json_ocsf_file_suffix,
)
from prowler.lib.logger import logger
from prowler.providers.common.outputs import Provider_Output_Options
def display_summary_table(
findings: list,
provider,
output_options: Provider_Output_Options,
output_options,
):
output_directory = output_options.output_directory
output_filename = output_options.output_filename
prowler/providers/aws/aws_provider.py
+29 -11
View File
@@ -33,6 +33,7 @@ from prowler.providers.aws.models import (
AWSIdentityInfo,
AWSMFAInfo,
AWSOrganizationsInfo,
AWSOutputOptions,
AWSSession,
)
from prowler.providers.common.models import Audit_Metadata
@@ -48,6 +49,9 @@ class AwsProvider(Provider):
_audit_config: dict = {}
_ignore_unused_services: bool = False
_enabled_regions: set = set()
# TODO: enforce the mutelist for the Provider class
_mutelist: dict = {}
_output_options: AWSOutputOptions
# TODO: this is not optional, enforce for all providers
audit_metadata: Audit_Metadata
@@ -254,6 +258,31 @@ class AwsProvider(Provider):
def audit_config(self):
return self._audit_config
@property
def output_options(self):
return self._output_options
@output_options.setter
def output_options(self, options: tuple):
arguments, bulk_checks_metadata = options
self._output_options = AWSOutputOptions(
arguments, bulk_checks_metadata, self._identity
)
@property
def mutelist(self):
return self._mutelist
@mutelist.setter
def mutelist(self, mutelist_path):
if mutelist_path:
mutelist = parse_mutelist_file(
self._session.current_session, self._identity.account, mutelist_path
)
else:
mutelist = {}
self._mutelist = mutelist
# TODO: This can be moved to another class since it doesn't need self
def get_organizations_info(
self, organizations_session: Session, aws_account_id: str
@@ -774,17 +803,6 @@ Caller Identity ARN: {Fore.YELLOW}[{self._identity.identity_arn}]{Style.RESET_AL
)
sys.exit(1)
# TODO: maybe create a function in the provider with a default empty string
def get_mutelist(self, mutelist_file):
# Parse content from Mute List file and get it, if necessary, from S3
if mutelist_file:
mutelist_file = parse_mutelist_file(
self.session.session, self._identity.account, mutelist_file
)
else:
mutelist_file = None
return mutelist_file
def read_aws_regions_file() -> dict:
# Get JSON locally
prowler/providers/aws/lib/mutelist/mutelist.py
+12 -11
View File
@@ -3,6 +3,7 @@ import sys
from typing import Any
import yaml
from boto3 import Session
from boto3.dynamodb.conditions import Attr
from schema import Optional, Schema
@@ -32,34 +33,34 @@ mutelist_schema = Schema(
)
def parse_mutelist_file(session, aws_account, mutelist_file):
def parse_mutelist_file(session: Session, aws_account: str, mutelist_path: str):
try:
# Check if file is a S3 URI
if re.search("^s3://([^/]+)/(.*?([^/]+))$", mutelist_file):
bucket = mutelist_file.split("/")[2]
key = ("/").join(mutelist_file.split("/")[3:])
if re.search("^s3://([^/]+)/(.*?([^/]+))$", mutelist_path):
bucket = mutelist_path.split("/")[2]
key = ("/").join(mutelist_path.split("/")[3:])
s3_client = session.client("s3")
mutelist = yaml.safe_load(
s3_client.get_object(Bucket=bucket, Key=key)["Body"]
)["Mute List"]
# Check if file is a Lambda Function ARN
elif re.search(r"^arn:(\w+):lambda:", mutelist_file):
lambda_region = mutelist_file.split(":")[3]
elif re.search(r"^arn:(\w+):lambda:", mutelist_path):
lambda_region = mutelist_path.split(":")[3]
lambda_client = session.client("lambda", region_name=lambda_region)
lambda_response = lambda_client.invoke(
FunctionName=mutelist_file, InvocationType="RequestResponse"
FunctionName=mutelist_path, InvocationType="RequestResponse"
)
lambda_payload = lambda_response["Payload"].read()
mutelist = yaml.safe_load(lambda_payload)["Mute List"]
# Check if file is a DynamoDB ARN
elif re.search(
r"^arn:aws(-cn|-us-gov)?:dynamodb:[a-z]{2}-[a-z-]+-[1-9]{1}:[0-9]{12}:table\/[a-zA-Z0-9._-]+$",
mutelist_file,
mutelist_path,
):
mutelist = {"Accounts": {}}
table_region = mutelist_file.split(":")[3]
table_region = mutelist_path.split(":")[3]
dynamodb_resource = session.resource("dynamodb", region_name=table_region)
dynamo_table = dynamodb_resource.Table(mutelist_file.split("/")[1])
dynamo_table = dynamodb_resource.Table(mutelist_path.split("/")[1])
response = dynamo_table.scan(
FilterExpression=Attr("Accounts").is_in([aws_account, "*"])
)
@@ -90,7 +91,7 @@ def parse_mutelist_file(session, aws_account, mutelist_file):
"Exceptions"
] = item["Exceptions"]
else:
with open(mutelist_file) as f:
with open(mutelist_path) as f:
mutelist = yaml.safe_load(f)["Mute List"]
try:
mutelist_schema.validate(mutelist)
prowler/providers/aws/models.py
+37
View File
@@ -4,7 +4,9 @@ from datetime import datetime
from boto3.session import Session
from botocore.config import Config
from prowler.config.config import output_file_timestamp
from prowler.providers.aws.lib.arn.models import ARN
from prowler.providers.common.models import ProviderOutputOptions
@dataclass
@@ -75,3 +77,38 @@ class AWSCallerIdentity:
class AWSMFAInfo:
arn: str
totp: str
class AWSOutputOptions(ProviderOutputOptions):
security_hub_enabled: bool
def __init__(self, arguments, bulk_checks_metadata, identity):
# First call Provider_Output_Options init
super().__init__(arguments, bulk_checks_metadata)
# Confire Shodan API
# TODO: review shodan for the new AWS provider
# if arguments.shodan:
# audit_info = change_config_var(
# "shodan_api_key", arguments.shodan, audit_info
# )
# Check if custom output filename was input, if not, set the default
if (
not hasattr(arguments, "output_filename")
or arguments.output_filename is None
):
self.output_filename = (
f"prowler-output-{identity.account}-{output_file_timestamp}"
)
else:
self.output_filename = arguments.output_filename
# Security Hub Outputs
self.security_hub_enabled = arguments.security_hub
self.send_sh_only_fails = arguments.send_sh_only_fails
if arguments.security_hub:
if not self.output_modes:
self.output_modes = ["json-asff"]
else:
self.output_modes.append("json-asff")
prowler/providers/azure/azure_provider.py
+34 -1
View File
@@ -11,7 +11,11 @@ from msgraph import GraphServiceClient
from prowler.lib.logger import logger
from prowler.providers.azure.lib.regions.regions import get_regions_config
from prowler.providers.azure.models import AzureIdentityInfo, AzureRegionConfig
from prowler.providers.azure.models import (
AzureIdentityInfo,
AzureOutputOptions,
AzureRegionConfig,
)
from prowler.providers.common.models import Audit_Metadata
from prowler.providers.common.provider import Provider
@@ -23,6 +27,9 @@ class AzureProvider(Provider):
_audit_config: Optional[dict]
_region_config: AzureRegionConfig
_locations: dict
_output_options: AzureOutputOptions
# TODO: enforce the mutelist for the Provider class
# _mutelist: dict = {}
# TODO: this is not optional, enforce for all providers
audit_metadata: Audit_Metadata
@@ -83,6 +90,32 @@ class AzureProvider(Provider):
def audit_config(self):
return self._audit_config
@property
def output_options(self):
return self._output_options
@output_options.setter
def output_options(self, options: tuple):
arguments, bulk_checks_metadata = options
self._output_options = AzureOutputOptions(
arguments, bulk_checks_metadata, self._identity
)
# TODO: pending to implement
# @property
# def mutelist(self):
# return self._mutelist
# @mutelist.setter
# def mutelist(self, mutelist_path):
# if mutelist_path:
# mutelist = parse_mutelist_file(
# self._session.current_session, self._identity.account, mutelist_path
# )
# else:
# mutelist = {}
# self._mutelist = mutelist
# TODO: this should be moved to the argparse, if not we need to enforce it from the Provider
def validate_arguments(
self, az_cli_auth, sp_env_auth, browser_auth, managed_entity_auth, tenant_id
prowler/providers/azure/models.py
+30
View File
@@ -1,5 +1,8 @@
from pydantic import BaseModel
from prowler.config.config import output_file_timestamp
from prowler.providers.common.models import ProviderOutputOptions
class AzureIdentityInfo(BaseModel):
identity_id: str = ""
@@ -15,3 +18,30 @@ class AzureRegionConfig(BaseModel):
authority: str = None
base_url: str = ""
credential_scopes: list = []
class AzureOutputOptions(ProviderOutputOptions):
def __init__(self, arguments, bulk_checks_metadata, identity):
# First call Provider_Output_Options init
super().__init__(arguments, bulk_checks_metadata)
# Confire Shodan API
# TODO: review shodan for the new AWS provider
# if arguments.shodan:
# audit_info = change_config_var(
# "shodan_api_key", arguments.shodan, audit_info
# )
# Check if custom output filename was input, if not, set the default
if (
not hasattr(arguments, "output_filename")
or arguments.output_filename is None
):
if identity.domain != "Unknown tenant domain (missing AAD permissions)":
self.output_filename = (
f"prowler-output-{identity.domain}-{output_file_timestamp}"
)
else:
self.output_filename = f"prowler-output-{'-'.join(identity.tenant_ids)}-{output_file_timestamp}"
else:
self.output_filename = arguments.output_filename
prowler/providers/common/models.py
+31
View File
@@ -1,3 +1,6 @@
from os import makedirs
from os.path import isdir
from pydantic import BaseModel
@@ -9,3 +12,31 @@ class Audit_Metadata(BaseModel):
expected_checks: list
completed_checks: int
audit_progress: int
class ProviderOutputOptions:
status: bool
output_modes: list
output_directory: str
bulk_checks_metadata: dict
verbose: str
output_filename: str
only_logs: bool
unix_timestamp: bool
def __init__(self, arguments, bulk_checks_metadata):
self.status = arguments.status
self.output_modes = arguments.output_modes
self.output_directory = arguments.output_directory
self.verbose = arguments.verbose
self.bulk_checks_metadata = bulk_checks_metadata
self.only_logs = arguments.only_logs
self.unix_timestamp = arguments.unix_timestamp
# Check output directory, if it is not created -> create it
if arguments.output_directory:
if not isdir(arguments.output_directory):
if arguments.output_modes:
makedirs(arguments.output_directory, exist_ok=True)
if not isdir(arguments.output_directory + "/compliance"):
if arguments.output_modes:
makedirs(arguments.output_directory + "/compliance", exist_ok=True)
prowler/providers/common/outputs.py
+57 -149
View File
@@ -1,34 +1,37 @@
import importlib
import sys
from dataclasses import dataclass
from os import makedirs
from os.path import isdir
from prowler.config.config import output_file_timestamp
from prowler.lib.logger import logger
# TODO: remove after fixing tests
# import sys
# from dataclasses import dataclass
# from os import makedirs
# from os.path import isdir
# from prowler.lib.logger import logger
def set_provider_output_options(
provider: str, arguments, identity, mutelist_file, bulk_checks_metadata
):
"""
set_provider_output_options configures automatically the outputs based on the selected provider and returns the Provider_Output_Options object.
"""
try:
# Dynamically load the Provider_Output_Options class
provider_output_class = f"{provider.capitalize()}_Output_Options"
provider_output_options = getattr(
importlib.import_module(__name__), provider_output_class
)(arguments, identity, mutelist_file, bulk_checks_metadata)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
sys.exit(1)
else:
return provider_output_options
# TODO: remove after fixing tests
# def set_provider_output_options(
# provider: str, arguments, identity, mutelist_file, bulk_checks_metadata
# ):
# """
# set_provider_output_options configures automatically the outputs based on the selected provider and returns the Provider_Output_Options object.
# """
# try:
# # Dynamically load the Provider_Output_Options class
# provider_output_class = f"{provider.capitalize()}_Output_Options"
# provider_output_options = getattr(
# importlib.import_module(__name__), provider_output_class
# )(arguments, identity, mutelist_file, bulk_checks_metadata)
# except Exception as error:
# logger.critical(
# f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
# )
# sys.exit(1)
# else:
# return provider_output_options
# TODO: review this function, probably is not needed anymore
def get_provider_output_model(provider_type):
"""
get_provider_output_model returns the model <provider>_Check_Output_CSV for each provider
@@ -43,128 +46,33 @@ def get_provider_output_model(provider_type):
return output_provider_model
@dataclass
class Provider_Output_Options:
status: bool
output_modes: list
output_directory: str
mutelist_file: str
bulk_checks_metadata: dict
verbose: str
output_filename: str
only_logs: bool
unix_timestamp: bool
# TODO: remove after fixing tests
# @dataclass
# class Provider_Output_Options:
# status: bool
# output_modes: list
# output_directory: str
# mutelist_file: str
# bulk_checks_metadata: dict
# verbose: str
# output_filename: str
# only_logs: bool
# unix_timestamp: bool
def __init__(self, arguments, mutelist_file, bulk_checks_metadata):
self.status = arguments.status
self.output_modes = arguments.output_modes
self.output_directory = arguments.output_directory
self.verbose = arguments.verbose
self.bulk_checks_metadata = bulk_checks_metadata
self.mutelist_file = mutelist_file
self.only_logs = arguments.only_logs
self.unix_timestamp = arguments.unix_timestamp
# Check output directory, if it is not created -> create it
if arguments.output_directory:
if not isdir(arguments.output_directory):
if arguments.output_modes:
makedirs(arguments.output_directory, exist_ok=True)
if not isdir(arguments.output_directory + "/compliance"):
if arguments.output_modes:
makedirs(arguments.output_directory + "/compliance", exist_ok=True)
class Azure_Output_Options(Provider_Output_Options):
def __init__(self, arguments, identity, mutelist_file, bulk_checks_metadata):
# First call Provider_Output_Options init
super().__init__(arguments, mutelist_file, bulk_checks_metadata)
# Confire Shodan API
# TODO: review shodan for the new AWS provider
# if arguments.shodan:
# audit_info = change_config_var(
# "shodan_api_key", arguments.shodan, audit_info
# )
# Check if custom output filename was input, if not, set the default
if (
not hasattr(arguments, "output_filename")
or arguments.output_filename is None
):
if identity.domain != "Unknown tenant domain (missing AAD permissions)":
self.output_filename = (
f"prowler-output-{identity.domain}-{output_file_timestamp}"
)
else:
self.output_filename = f"prowler-output-{'-'.join(identity.tenant_ids)}-{output_file_timestamp}"
else:
self.output_filename = arguments.output_filename
class Gcp_Output_Options(Provider_Output_Options):
def __init__(self, arguments, identity, mutelist_file, bulk_checks_metadata):
# First call Provider_Output_Options init
super().__init__(arguments, mutelist_file, bulk_checks_metadata)
# Check if custom output filename was input, if not, set the default
if (
not hasattr(arguments, "output_filename")
or arguments.output_filename is None
):
self.output_filename = (
f"prowler-output-{identity.profile}-{output_file_timestamp}"
)
else:
self.output_filename = arguments.output_filename
class Kubernetes_Output_Options(Provider_Output_Options):
def __init__(self, arguments, identity, mutelist_file, bulk_checks_metadata):
# First call Provider_Output_Options init
super().__init__(arguments, mutelist_file, bulk_checks_metadata)
# TODO move the below if to Provider_Output_Options
# Check if custom output filename was input, if not, set the default
if (
not hasattr(arguments, "output_filename")
or arguments.output_filename is None
):
self.output_filename = (
f"prowler-output-{identity.context}-{output_file_timestamp}"
)
else:
self.output_filename = arguments.output_filename
class Aws_Output_Options(Provider_Output_Options):
security_hub_enabled: bool
def __init__(self, arguments, identity, mutelist_file, bulk_checks_metadata):
# First call Provider_Output_Options init
super().__init__(arguments, mutelist_file, bulk_checks_metadata)
# Confire Shodan API
# TODO: review shodan for the new AWS provider
# if arguments.shodan:
# audit_info = change_config_var(
# "shodan_api_key", arguments.shodan, audit_info
# )
# Check if custom output filename was input, if not, set the default
if (
not hasattr(arguments, "output_filename")
or arguments.output_filename is None
):
self.output_filename = (
f"prowler-output-{identity.account}-{output_file_timestamp}"
)
else:
self.output_filename = arguments.output_filename
# Security Hub Outputs
self.security_hub_enabled = arguments.security_hub
self.send_sh_only_fails = arguments.send_sh_only_fails
if arguments.security_hub:
if not self.output_modes:
self.output_modes = ["json-asff"]
else:
self.output_modes.append("json-asff")
# def __init__(self, arguments, mutelist_file, bulk_checks_metadata):
# self.status = arguments.status
# self.output_modes = arguments.output_modes
# self.output_directory = arguments.output_directory
# self.verbose = arguments.verbose
# self.bulk_checks_metadata = bulk_checks_metadata
# self.mutelist_file = mutelist_file
# self.only_logs = arguments.only_logs
# self.unix_timestamp = arguments.unix_timestamp
# # Check output directory, if it is not created -> create it
# if arguments.output_directory:
# if not isdir(arguments.output_directory):
# if arguments.output_modes:
# makedirs(arguments.output_directory, exist_ok=True)
# if not isdir(arguments.output_directory + "/compliance"):
# if arguments.output_modes:
# makedirs(arguments.output_directory + "/compliance", exist_ok=True)
prowler/providers/common/provider.py
+18
View File
@@ -62,6 +62,24 @@ class Provider(ABC):
def setup_session(self):
pass
@property
@abstractmethod
def output_options(self):
"""
output_options method returns the provider's audit output configuration.
This method needs to be created in each provider.
"""
@output_options.setter
@abstractmethod
def output_options(self):
"""
output_options.setter sets the provider's audit output configuration.
This method needs to be created in each provider.
"""
# TODO: probably this won't be here since we want to do the arguments validation during the parse()
def validate_arguments(self):
pass
prowler/providers/gcp/gcp_provider.py
+30 -7
View File
@@ -1,6 +1,5 @@
import os
import sys
from dataclasses import dataclass
from typing import Optional
from colorama import Fore, Style
@@ -11,12 +10,7 @@ from googleapiclient import discovery
from prowler.lib.logger import logger
from prowler.providers.common.models import Audit_Metadata
from prowler.providers.common.provider import Provider
@dataclass
class GCPIdentityInfo:
profile: str
default_project_id: str
from prowler.providers.gcp.models import GCPIdentityInfo, GCPOutputOptions
class GcpProvider(Provider):
@@ -25,6 +19,9 @@ class GcpProvider(Provider):
_project_ids: list
_identity: GCPIdentityInfo
_audit_config: Optional[dict]
_output_options: GCPOutputOptions
# TODO: enforce the mutelist for the Provider class
# _mutelist: dict = {}
# TODO: this is not optional, enforce for all providers
audit_metadata: Audit_Metadata
@@ -79,6 +76,32 @@ class GcpProvider(Provider):
def audit_config(self):
return self._audit_config
@property
def output_options(self):
return self._output_options
@output_options.setter
def output_options(self, options: tuple):
arguments, bulk_checks_metadata = options
self._output_options = GCPOutputOptions(
arguments, bulk_checks_metadata, self._identity
)
# TODO: pending to implement
# @property
# def mutelist(self):
# return self._mutelist
# @mutelist.setter
# def mutelist(self, mutelist_path):
# if mutelist_path:
# mutelist = parse_mutelist_file(
# self._session.current_session, self._identity.account, mutelist_path
# )
# else:
# mutelist = {}
# self._mutelist = mutelist
def setup_session(self, credentials_file):
try:
if credentials_file:
prowler/providers/gcp/models.py
+27
View File
@@ -0,0 +1,27 @@
from dataclasses import dataclass
from prowler.config.config import output_file_timestamp
from prowler.providers.common.models import ProviderOutputOptions
@dataclass
class GCPIdentityInfo:
profile: str
default_project_id: str
class GCPOutputOptions(ProviderOutputOptions):
def __init__(self, arguments, bulk_checks_metadata, identity):
# First call ProviderOutputOptions init
super().__init__(arguments, bulk_checks_metadata)
# Check if custom output filename was input, if not, set the default
if (
not hasattr(arguments, "output_filename")
or arguments.output_filename is None
):
self.output_filename = (
f"prowler-output-{identity.profile}-{output_file_timestamp}"
)
else:
self.output_filename = arguments.output_filename
prowler/providers/kubernetes/kubernetes_provider.py
+34 -19
View File
@@ -1,7 +1,6 @@
import os
import sys
from argparse import Namespace
from dataclasses import dataclass
from typing import Optional
from colorama import Fore, Style
@@ -10,24 +9,11 @@ from kubernetes import client, config
from prowler.lib.logger import logger
from prowler.providers.common.models import Audit_Metadata
from prowler.providers.common.provider import Provider
@dataclass
class KubernetesIdentityInfo:
context: str
cluster: str
user: str
@dataclass
class KubernetesSession:
"""
KubernetesSession stores the Kubernetes session's configuration.
"""
api_client: client.ApiClient
context: dict
from prowler.providers.kubernetes.models import (
KubernetesIdentityInfo,
KubernetesOutputOptions,
KubernetesSession,
)
class KubernetesProvider(Provider):
@@ -36,6 +22,9 @@ class KubernetesProvider(Provider):
_namespaces: list
_audit_config: Optional[dict]
_identity: KubernetesIdentityInfo
_output_options: KubernetesOutputOptions
# TODO: enforce the mutelist for the Provider class
# _mutelist: dict = {}
# TODO: this is not optional, enforce for all providers
audit_metadata: Audit_Metadata
@@ -83,6 +72,32 @@ class KubernetesProvider(Provider):
def audit_config(self):
return self._audit_config
@property
def output_options(self):
return self._output_options
@output_options.setter
def output_options(self, options: tuple):
arguments, bulk_checks_metadata = options
self._output_options = KubernetesOutputOptions(
arguments, bulk_checks_metadata, self._identity
)
# TODO: pending to implement
# @property
# def mutelist(self):
# return self._mutelist
# @mutelist.setter
# def mutelist(self, mutelist_path):
# if mutelist_path:
# mutelist = parse_mutelist_file(
# self._session.current_session, self._identity.account, mutelist_path
# )
# else:
# mutelist = {}
# self._mutelist = mutelist
def setup_session(self, kubeconfig_file, input_context) -> KubernetesSession:
"""
Sets up the Kubernetes session.
prowler/providers/kubernetes/models.py
+41
View File
@@ -0,0 +1,41 @@
from dataclasses import dataclass
from kubernetes import client
from prowler.config.config import output_file_timestamp
from prowler.providers.common.models import ProviderOutputOptions
@dataclass
class KubernetesIdentityInfo:
context: str
cluster: str
user: str
@dataclass
class KubernetesSession:
"""
KubernetesSession stores the Kubernetes session's configuration.
"""
api_client: client.ApiClient
context: dict
class KubernetesOutputOptions(ProviderOutputOptions):
def __init__(self, arguments, bulk_checks_metadata, identity):
# First call ProviderOutputOptions init
super().__init__(arguments, bulk_checks_metadata)
# TODO move the below if to ProviderOutputOptions
# Check if custom output filename was input, if not, set the default
if (
not hasattr(arguments, "output_filename")
or arguments.output_filename is None
):
self.output_filename = (
f"prowler-output-{identity.context}-{output_file_timestamp}"
)
else:
self.output_filename = arguments.output_filename