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

chore(aws): Remove old provider (#3468)

Browse Source
This commit is contained in:
Pepe Fagoaga
2024-02-29 13:45:43 +01:00
committed by GitHub
parent f0c027f54e
commit 0b32a10bb8
11 changed files with 910 additions and 1243 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/__main__.py
+40 -35
View File
@@ -1,7 +1,6 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import sys
from colorama import Fore, Style
@@ -37,7 +36,6 @@ from prowler.lib.outputs.compliance.compliance import display_compliance_table
from prowler.lib.outputs.html import add_html_footer, fill_html_overview_statistics
from prowler.lib.outputs.json import close_json
from prowler.lib.outputs.outputs import extract_findings_statistics
from prowler.lib.outputs.slack import send_slack_message
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.security_hub.security_hub import (
@@ -46,18 +44,17 @@ from prowler.providers.aws.lib.security_hub.security_hub import (
resolve_security_hub_previous_findings,
verify_security_hub_integration_enabled_per_region,
)
from prowler.providers.common.audit_info import (
set_provider_audit_info,
set_provider_execution_parameters,
)
# from prowler.providers.common.audit_info import (
# set_provider_audit_info,
# set_provider_execution_parameters,
# )
from prowler.providers.common.clean import clean_provider_local_output_directories
from prowler.providers.common.common import (
get_global_provider,
set_global_provider_object,
)
from prowler.providers.common.mutelist import set_provider_mutelist
from prowler.providers.common.outputs import set_provider_output_options
from prowler.providers.common.quick_inventory import run_provider_quick_inventory
def prowler():
@@ -154,13 +151,17 @@ def prowler():
print_checks(provider, sorted(checks_to_execute), bulk_checks_metadata)
sys.exit()
# Set the audit info based on the selected provider
audit_info = set_provider_audit_info(provider, args.__dict__)
# Set the audit info based on the selected provider
# TODO: remove the following line with the audit_info
# audit_info = set_provider_audit_info(provider, args.__dict__)
set_global_provider_object(args)
# TODO: rename global_provider to provider
global_provider = get_global_provider()
# Import custom checks from folder
if checks_folder:
parse_checks_from_folder(audit_info, checks_folder, provider)
parse_checks_from_folder(global_provider, checks_folder)
# Exclude checks if -e/--excluded-checks
if excluded_checks:
@@ -174,33 +175,37 @@ def prowler():
# Once the audit_info is set and we have the eventual checks based on the resource identifier,
# it is time to check what Prowler's checks are going to be executed
if audit_info.audit_resources:
checks_from_resources = set_provider_execution_parameters(provider, audit_info)
# TODO: the following if is done within the function
# if global_provider.audit_resources:
checks_from_resources = global_provider.get_checks_to_execute_by_audit_resources()
if checks_from_resources:
checks_to_execute = checks_to_execute.intersection(checks_from_resources)
# Sort final check list
checks_to_execute = sorted(checks_to_execute)
# Parse Mute List
mutelist_file = set_provider_mutelist(provider, audit_info, args)
if hasattr(args, "mutelist_file"):
mutelist_file = global_provider.get_mutelist(args.mutelist_file)
# Set output options based on the selected provider
audit_output_options = set_provider_output_options(
provider, args, audit_info, mutelist_file, bulk_checks_metadata
provider, args, global_provider.identity, mutelist_file, bulk_checks_metadata
)
# TODO: adapt the quick inventory for the new AWS provider
# Run the quick inventory for the provider if available
if hasattr(args, "quick_inventory") and args.quick_inventory:
run_provider_quick_inventory(provider, audit_info, args)
sys.exit()
# if hasattr(args, "quick_inventory") and args.quick_inventory:
# run_provider_quick_inventory(provider, global_provider.identity, args)
# sys.exit()
# Execute checks
findings = []
if len(checks_to_execute):
findings = execute_checks(
checks_to_execute,
provider,
audit_info,
global_provider,
audit_output_options,
custom_checks_metadata,
)
@@ -212,20 +217,21 @@ def prowler():
# Extract findings stats
stats = extract_findings_statistics(findings)
if args.slack:
if "SLACK_API_TOKEN" in os.environ and "SLACK_CHANNEL_ID" in os.environ:
_ = send_slack_message(
os.environ["SLACK_API_TOKEN"],
os.environ["SLACK_CHANNEL_ID"],
stats,
provider,
audit_info,
)
else:
logger.critical(
"Slack integration needs SLACK_API_TOKEN and SLACK_CHANNEL_ID environment variables (see more in https://docs.prowler.cloud/en/latest/tutorials/integrations/#slack)."
)
sys.exit(1)
# TODO: adapt the slack integration for the new AWS provider
# if args.slack:
# if "SLACK_API_TOKEN" in os.environ and "SLACK_CHANNEL_ID" in os.environ:
# _ = send_slack_message(
# os.environ["SLACK_API_TOKEN"],
# os.environ["SLACK_CHANNEL_ID"],
# stats,
# provider,
# audit_info,
# )
# else:
# logger.critical(
# "Slack integration needs SLACK_API_TOKEN and SLACK_CHANNEL_ID environment variables (see more in https://docs.prowler.cloud/en/latest/tutorials/integrations/#slack)."
# )
# sys.exit(1)
if args.output_modes:
for mode in args.output_modes:
@@ -265,7 +271,6 @@ def prowler():
f"{Style.BRIGHT}\nSending findings to AWS Security Hub, please wait...{Style.RESET_ALL}"
)
# Verify where AWS Security Hub is enabled
global_provider = get_global_provider()
aws_security_enabled_regions = []
security_hub_regions = (
global_provider.get_available_aws_service_regions("securityhub")
prowler/lib/check/check.py
+16 -20
View File
@@ -125,16 +125,16 @@ def parse_checks_from_file(input_file: str, provider: str) -> set:
# Load checks from custom folder
def parse_checks_from_folder(audit_info, input_folder: str, provider: str) -> int:
def parse_checks_from_folder(provider, input_folder: str) -> int:
try:
imported_checks = 0
# Check if input folder is a S3 URI
if provider == "aws" and re.search(
if provider.provider == "aws" and re.search(
"^s3://([^/]+)/(.*?([^/]+))/$", input_folder
):
bucket = input_folder.split("/")[2]
key = ("/").join(input_folder.split("/")[3:])
s3_resource = audit_info.audit_session.resource("s3")
s3_resource = provider.session.session.resource("s3")
bucket = s3_resource.Bucket(bucket)
for obj in bucket.objects.filter(Prefix=key):
if not os.path.exists(os.path.dirname(obj.key)):
@@ -151,7 +151,7 @@ def parse_checks_from_folder(audit_info, input_folder: str, provider: str) -> in
# Copy checks to specific provider/service folder
check_service = check.name.split("_")[0]
prowler_dir = prowler.__path__
prowler_module = f"{prowler_dir[0]}/providers/{provider}/services/{check_service}/{check.name}"
prowler_module = f"{prowler_dir[0]}/providers/{provider.provider}/services/{check_service}/{check.name}"
if os.path.exists(prowler_module):
shutil.rmtree(prowler_module)
shutil.copytree(check_module, prowler_module)
@@ -421,8 +421,7 @@ def run_check(check: Check, output_options: Provider_Output_Options) -> list:
def execute_checks(
checks_to_execute: list,
provider: str,
audit_info: Any,
global_provider: Any,
audit_output_options: Provider_Output_Options,
custom_checks_metadata: Any,
) -> list:
@@ -467,9 +466,9 @@ def execute_checks(
check_findings = execute(
service,
check_name,
provider,
global_provider.provider,
audit_output_options,
audit_info,
global_provider.identity,
services_executed,
checks_executed,
custom_checks_metadata,
@@ -479,7 +478,7 @@ def execute_checks(
# If check does not exists in the provider or is from another provider
except ModuleNotFoundError:
logger.error(
f"Check '{check_name}' was not found for the {provider.upper()} provider"
f"Check '{check_name}' was not found for the {global_provider.provider.upper()} provider"
)
except Exception as error:
logger.error(
@@ -513,9 +512,8 @@ def execute_checks(
check_findings = execute(
service,
check_name,
provider,
audit_output_options,
audit_info,
global_provider,
services_executed,
checks_executed,
custom_checks_metadata,
@@ -525,7 +523,7 @@ def execute_checks(
# If check does not exists in the provider or is from another provider
except ModuleNotFoundError:
logger.error(
f"Check '{check_name}' was not found for the {provider.upper()} provider"
f"Check '{check_name}' was not found for the {global_provider.provider.upper()} provider"
)
except Exception as error:
logger.error(
@@ -539,18 +537,14 @@ def execute_checks(
def execute(
service: str,
check_name: str,
provider: str,
audit_output_options: Provider_Output_Options,
audit_info: Any,
global_provider: Any,
services_executed: set,
checks_executed: set,
custom_checks_metadata: Any,
):
global_provider = get_global_provider()
# Import check module
check_module_path = (
f"prowler.providers.{provider}.services.{service}.{check_name}.{check_name}"
)
check_module_path = f"prowler.providers.{global_provider.provider}.services.{service}.{check_name}.{check_name}"
lib = import_check(check_module_path)
# Recover functions from check
check_to_execute = getattr(lib, check_name)
@@ -579,7 +573,7 @@ def execute(
)
# Report the check's findings
report(check_findings, audit_output_options, audit_info)
report(check_findings, audit_output_options, global_provider.identity)
if os.environ.get("PROWLER_REPORT_LIB_PATH"):
try:
@@ -588,7 +582,9 @@ def execute(
outputs_module = importlib.import_module(lib)
custom_report_interface = getattr(outputs_module, "report")
custom_report_interface(check_findings, audit_output_options, audit_info)
custom_report_interface(
check_findings, audit_output_options, global_provider.identity
)
except Exception:
sys.exit(1)
prowler/providers/aws/aws_provider.py
+568 -282
View File
@@ -1,102 +1,300 @@
import os
import pathlib
import sys
from argparse import Namespace
from typing import Any, Optional
from boto3 import client, session
from botocore.config import Config
from botocore.credentials import RefreshableCredentials
from botocore.session import get_session
from colorama import Fore, Style
from prowler.config.config import aws_services_json_file
from prowler.lib.check.check import list_modules, recover_checks_from_service
from prowler.lib.logger import logger
from prowler.lib.utils.utils import open_file, parse_json_file
# TODO: Add ROLE_SESSION_NAME feature
from prowler.providers.aws.config import (
AWS_STS_GLOBAL_ENDPOINT_REGION,
ROLE_SESSION_NAME,
BOTO3_USER_AGENT_EXTRA,
)
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info, AWSAssumeRole
from prowler.providers.aws.lib.credentials.credentials import create_sts_session
from prowler.providers.aws.lib.arn.arn import parse_iam_credentials_arn
from prowler.providers.aws.lib.credentials.credentials import (
create_sts_session,
validate_AWSCredentials,
)
from prowler.providers.aws.lib.mutelist.mutelist import parse_mutelist_file
from prowler.providers.aws.lib.organizations.organizations import (
get_organizations_metadata,
)
from prowler.providers.aws.models import (
AWSAssumeRole,
AWSAssumeRoleConfiguration,
AWSCredentials,
AWSIdentityInfo,
AWSOrganizationsInfo,
AWSSession,
)
from prowler.providers.common.provider import Provider
################## AWS PROVIDER
class AWS_Provider:
def __init__(self, audit_info):
logger.info("Instantiating aws provider ...")
self.aws_session = self.set_session(audit_info)
self.role_info = audit_info.assumed_role_info
class AwsProvider(Provider):
provider: str = "aws"
session: AWSSession = AWSSession(
session=None, session_config=None, original_session=None
)
identity: AWSIdentityInfo = AWSIdentityInfo(
account=None,
account_arn=None,
user_id=None,
partition=None,
identity_arn=None,
profile=None,
profile_region=None,
audited_regions=[],
)
assumed_role: AWSAssumeRoleConfiguration = AWSAssumeRoleConfiguration(
assumed_role_info=AWSAssumeRole(
role_arn=None,
session_duration=None,
external_id=None,
mfa_enabled=False,
),
assumed_role_credentials=AWSCredentials(
aws_access_key_id=None,
aws_session_token=None,
aws_secret_access_key=None,
expiration=None,
),
)
organizations_metadata: AWSOrganizationsInfo = AWSOrganizationsInfo(
account_details_email=None,
account_details_name=None,
account_details_arn=None,
account_details_org=None,
account_details_tags=None,
)
audit_resources: Optional[Any]
audit_metadata: Optional[Any]
audit_config: dict = {}
mfa_enabled: bool = False
ignore_unused_services: bool = False
enabled_regions: set = set()
def get_session(self):
return self.aws_session
def __init__(self, arguments: Namespace):
logger.info("Setting AWS provider ...")
# Parse input arguments
# Assume Role Options
input_role = getattr(arguments, "role", None)
input_session_duration = getattr(arguments, "session_duration", None)
input_external_id = getattr(arguments, "external_id", None)
def set_session(self, audit_info):
try:
# If we receive a credentials object filled is coming form an assumed role, so renewal is needed
if audit_info.credentials:
logger.info("Creating session for assumed role ...")
# From botocore we can use RefreshableCredentials class, which has an attribute (refresh_using)
# that needs to be a method without arguments that retrieves a new set of fresh credentials
# asuming the role again. -> https://github.com/boto/botocore/blob/098cc255f81a25b852e1ecdeb7adebd94c7b1b73/botocore/credentials.py#L395
assumed_refreshable_credentials = RefreshableCredentials(
access_key=audit_info.credentials.aws_access_key_id,
secret_key=audit_info.credentials.aws_secret_access_key,
token=audit_info.credentials.aws_session_token,
expiry_time=audit_info.credentials.expiration,
refresh_using=self.refresh_credentials,
method="sts-assume-role",
)
# Here we need the botocore session since it needs to use refreshable credentials
assumed_botocore_session = get_session()
assumed_botocore_session._credentials = assumed_refreshable_credentials
assumed_botocore_session.set_config_variable(
"region", audit_info.profile_region
)
return session.Session(
profile_name=audit_info.profile,
botocore_session=assumed_botocore_session,
)
# If we do not receive credentials start the session using the profile
else:
logger.info("Creating session for not assumed identity ...")
# Input MFA only if a role is not going to be assumed
if audit_info.mfa_enabled and not audit_info.assumed_role_info.role_arn:
mfa_ARN, mfa_TOTP = input_role_mfa_token_and_code()
get_session_token_arguments = {
"SerialNumber": mfa_ARN,
"TokenCode": mfa_TOTP,
}
sts_client = client("sts")
session_credentials = sts_client.get_session_token(
**get_session_token_arguments
)
return session.Session(
aws_access_key_id=session_credentials["Credentials"][
"AccessKeyId"
],
aws_secret_access_key=session_credentials["Credentials"][
"SecretAccessKey"
],
aws_session_token=session_credentials["Credentials"][
"SessionToken"
],
profile_name=audit_info.profile,
)
else:
return session.Session(
profile_name=audit_info.profile,
)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
# STS Endpoint Region
sts_endpoint_region = getattr(arguments, "sts_endpoint_region", None)
# MFA Configuration (false by default)
input_mfa = getattr(arguments, "mfa", None)
input_profile = getattr(arguments, "profile", None)
input_regions = getattr(arguments, "region", None)
organizations_role_arn = getattr(arguments, "organizations_role", None)
# Set the maximum retries for the standard retrier config
aws_retries_max_attempts = getattr(arguments, "aws_retries_max_attempts", None)
# Set if unused services must be ignored
ignore_unused_services = getattr(arguments, "ignore_unused_services", None)
# Set the maximum retries for the standard retrier config
self.session.session_config = self.__set_session_config__(
aws_retries_max_attempts
)
# Set ignore unused services
self.ignore_unused_services = ignore_unused_services
# Start populating AWS identity object
self.identity.profile = input_profile
self.identity.audited_regions = input_regions
# We need to create an original sessions using regular auth path (creds, profile, etc)
logger.info("Generating original session ...")
self.session.session = self.setup_session(input_mfa)
# After the session is created, validate it
logger.info("Validating credentials ...")
caller_identity = validate_AWSCredentials(
self.session.session, input_regions, sts_endpoint_region
)
logger.info("Credentials validated")
logger.info(f"Original caller identity UserId: {caller_identity['UserId']}")
logger.info(f"Original caller identity ARN: {caller_identity['Arn']}")
# Set values of AWS identity object
self.identity.account = caller_identity["Account"]
self.identity.identity_arn = caller_identity["Arn"]
self.identity.user_id = caller_identity["UserId"]
self.identity.partition = parse_iam_credentials_arn(
caller_identity["Arn"]
).partition
self.identity.account_arn = (
f"arn:{self.identity.partition}:iam::{self.identity.account}:root"
)
# save original session
self.session.original_session = self.session.session
# time for checking role assumption
if input_role:
# session will be the assumed one
self.session.session = self.setup_assumed_session(
input_role,
input_external_id,
input_mfa,
input_session_duration,
sts_endpoint_region,
)
logger.info("Audit session is the new session created assuming role")
# check if organizations info is gonna be retrieved
if organizations_role_arn:
logger.info(
f"Getting organizations metadata for account {organizations_role_arn}"
)
# session will be the assumed one with organizations permissions
self.session.session = self.setup_assumed_session(
organizations_role_arn,
input_external_id,
input_mfa,
input_session_duration,
sts_endpoint_region,
)
self.organizations_metadata = get_organizations_metadata(
self.identity.account, self.assumed_role.assumed_role_credentials
)
logger.info("Organizations metadata retrieved")
if self.session.session.region_name:
self.identity.profile_region = self.session.session.region_name
else:
self.identity.profile_region = "us-east-1"
if not getattr(arguments, "only_logs", None):
self.print_credentials()
# Parse Scan Tags
if getattr(arguments, "resource_tags", None):
input_resource_tags = arguments.resource_tags
self.audit_resources = self.get_tagged_resources(input_resource_tags)
# Parse Input Resource ARNs
self.audit_resources = getattr(arguments, "resource_arn", None)
# Get Enabled Regions
self.enabled_regions = self.get_aws_enabled_regions()
def setup_session(self, input_mfa: bool):
logger.info("Creating regular session ...")
# Input MFA only if a role is not going to be assumed
if input_mfa and not self.assumed_role.assumed_role_info.role_arn:
mfa_ARN, mfa_TOTP = self.__input_role_mfa_token_and_code__()
get_session_token_arguments = {
"SerialNumber": mfa_ARN,
"TokenCode": mfa_TOTP,
}
sts_client = client("sts")
session_credentials = sts_client.get_session_token(
**get_session_token_arguments
)
return session.Session(
aws_access_key_id=session_credentials["Credentials"]["AccessKeyId"],
aws_secret_access_key=session_credentials["Credentials"][
"SecretAccessKey"
],
aws_session_token=session_credentials["Credentials"]["SessionToken"],
profile_name=self.identity.profile,
)
else:
return session.Session(
profile_name=self.identity.profile,
)
def setup_assumed_session(
self,
input_role: str,
input_external_id: str,
input_mfa: str,
session_duration: int,
sts_endpoint_region: str,
):
logger.info("Creating assumed session ...")
# store information about the role is gonna be assumed
self.assumed_role.assumed_role_info.role_arn = input_role
self.assumed_role.assumed_role_info.session_duration = session_duration
self.assumed_role.assumed_role_info.external_id = input_external_id
self.assumed_role.assumed_role_info.mfa_enabled = input_mfa
# Check if role arn is valid
try:
# this returns the arn already parsed into a dict to be used when it is needed to access its fields
role_arn_parsed = parse_iam_credentials_arn(
self.assumed_role.assumed_role_info.role_arn
)
except Exception as error:
logger.critical(f"{error.__class__.__name__} -- {error}")
sys.exit(1)
else:
logger.info(f"Assuming role {self.assumed_role.assumed_role_info.role_arn}")
# Assume the role
assumed_role_response = self.__assume_role__(
self.session.session,
sts_endpoint_region,
)
logger.info("Role assumed")
# Set the info needed to create a session with an assumed role
self.assumed_role.assumed_role_credentials = AWSCredentials(
aws_access_key_id=assumed_role_response["Credentials"]["AccessKeyId"],
aws_session_token=assumed_role_response["Credentials"]["SessionToken"],
aws_secret_access_key=assumed_role_response["Credentials"][
"SecretAccessKey"
],
expiration=assumed_role_response["Credentials"]["Expiration"],
)
# Set identity parameters
self.identity.account = role_arn_parsed.account_id
self.identity.partition = role_arn_parsed.partition
self.identity.account_arn = (
f"arn:{self.identity.partition}:iam::{self.identity.account}:root"
)
# From botocore we can use RefreshableCredentials class, which has an attribute (refresh_using)
# that needs to be a method without arguments that retrieves a new set of fresh credentials
# asuming the role again. -> https://github.com/boto/botocore/blob/098cc255f81a25b852e1ecdeb7adebd94c7b1b73/botocore/credentials.py#L395
assumed_refreshable_credentials = RefreshableCredentials(
access_key=self.assumed_role.assumed_role_credentials.aws_access_key_id,
secret_key=self.assumed_role.assumed_role_credentials.aws_secret_access_key,
token=self.assumed_role.assumed_role_credentials.aws_session_token,
expiry_time=self.assumed_role.assumed_role_credentials.expiration,
refresh_using=self.refresh_credentials,
method="sts-assume-role",
)
# Here we need the botocore session since it needs to use refreshable credentials
assumed_botocore_session = get_session()
assumed_botocore_session._credentials = assumed_refreshable_credentials
assumed_botocore_session.set_config_variable(
"region", self.identity.profile_region
)
return session.Session(
profile_name=self.identity.profile,
botocore_session=assumed_botocore_session,
)
# Refresh credentials method using assume role
# This method is called "adding ()" to the name, so it cannot accept arguments
# https://github.com/boto/botocore/blob/098cc255f81a25b852e1ecdeb7adebd94c7b1b73/botocore/credentials.py#L570
def refresh_credentials(self):
logger.info("Refreshing assumed credentials...")
response = assume_role(self.aws_session, self.role_info)
response = self.__assume_role__(self.aws_session, self.role_info)
refreshed_credentials = dict(
# Keys of the dict has to be the same as those that are being searched in the parent class
# https://github.com/boto/botocore/blob/098cc255f81a25b852e1ecdeb7adebd94c7b1b73/botocore/credentials.py#L609
@@ -109,108 +307,316 @@ class AWS_Provider:
logger.info(refreshed_credentials)
return refreshed_credentials
def assume_role(
session: session.Session,
assumed_role_info: AWSAssumeRole,
sts_endpoint_region: str = None,
) -> dict:
try:
role_session_name = (
assumed_role_info.role_session_name
if assumed_role_info.role_session_name
else ROLE_SESSION_NAME
def print_credentials(self):
# Beautify audited regions, set "all" if there is no filter region
regions = (
", ".join(self.identity.audited_regions)
if self.identity.audited_regions is not None
else "all"
)
# Beautify audited profile, set "default" if there is no profile set
profile = (
self.identity.profile if self.identity.profile is not None else "default"
)
assume_role_arguments = {
"RoleArn": assumed_role_info.role_arn,
"RoleSessionName": role_session_name,
"DurationSeconds": assumed_role_info.session_duration,
}
report = f"""
This report is being generated using credentials below:
# Set the info to assume the role from the partition, account and role name
if assumed_role_info.external_id:
assume_role_arguments["ExternalId"] = assumed_role_info.external_id
AWS-CLI Profile: {Fore.YELLOW}[{profile}]{Style.RESET_ALL} AWS Filter Region: {Fore.YELLOW}[{regions}]{Style.RESET_ALL}
AWS Account: {Fore.YELLOW}[{self.identity.account}]{Style.RESET_ALL} UserId: {Fore.YELLOW}[{self.identity.user_id}]{Style.RESET_ALL}
Caller Identity ARN: {Fore.YELLOW}[{self.identity.identity_arn}]{Style.RESET_ALL}
"""
# If -A is set, print Assumed Role ARN
if self.assumed_role.assumed_role_info.role_arn is not None:
report += f"""Assumed Role ARN: {Fore.YELLOW}[{self.assumed_role.assumed_role_info.role_arn}]{Style.RESET_ALL}
"""
print(report)
if assumed_role_info.mfa_enabled:
mfa_ARN, mfa_TOTP = input_role_mfa_token_and_code()
assume_role_arguments["SerialNumber"] = mfa_ARN
assume_role_arguments["TokenCode"] = mfa_TOTP
# Set the STS Endpoint Region
if sts_endpoint_region is None:
sts_endpoint_region = AWS_STS_GLOBAL_ENDPOINT_REGION
sts_client = create_sts_session(session, sts_endpoint_region)
assumed_credentials = sts_client.assume_role(**assume_role_arguments)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
sys.exit(1)
else:
return assumed_credentials
def input_role_mfa_token_and_code() -> tuple[str]:
"""input_role_mfa_token_and_code ask for the AWS MFA ARN and TOTP and returns it."""
mfa_ARN = input("Enter ARN of MFA: ")
mfa_TOTP = input("Enter MFA code: ")
return (mfa_ARN.strip(), mfa_TOTP.strip())
def generate_regional_clients(
service: str,
audit_info: AWS_Audit_Info,
) -> dict:
"""generate_regional_clients returns a dict with the following format for the given service:
Example:
{"eu-west-1": boto3_service_client}
"""
try:
regional_clients = {}
service_regions = get_available_aws_service_regions(service, audit_info)
# Get the regions enabled for the account and get the intersection with the service available regions
if audit_info.enabled_regions:
enabled_regions = service_regions.intersection(audit_info.enabled_regions)
else:
enabled_regions = service_regions
for region in enabled_regions:
regional_client = audit_info.audit_session.client(
service, region_name=region, config=audit_info.session_config
def generate_regional_clients(
self, service: str, global_service: bool = False
) -> dict:
try:
regional_clients = {}
service_regions = self.get_available_aws_service_regions(service)
# Check if it is global service to gather only one region
if global_service:
if service_regions:
if self.identity.profile_region in service_regions:
service_regions = [self.identity.profile_region]
service_regions = service_regions[:1]
for region in service_regions:
regional_client = self.session.session.client(
service, region_name=region, config=self.session.session_config
)
regional_client.region = region
regional_clients[region] = regional_client
return regional_clients
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
regional_client.region = region
regional_clients[region] = regional_client
return regional_clients
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
def get_available_aws_service_regions(self, service: str) -> list:
# Get json locally
actual_directory = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
with open_file(f"{actual_directory}/{aws_services_json_file}") as f:
data = parse_json_file(f)
# Check if it is a subservice
json_regions = data["services"][service]["regions"][self.identity.partition]
if (
self.identity.audited_regions
): # Check for input aws audit_info.audited_regions
regions = list(
set(json_regions).intersection(self.identity.audited_regions)
) # Get common regions between input and json
else: # Get all regions from json of the service and partition
regions = json_regions
return regions
# TODO: why do we need the provider as an argument if this is only for AWS?
# This should be moved to the checks folder, maybe????
# Remove if not needed
def get_checks_from_input_arn(self) -> set:
"""get_checks_from_input_arn gets the list of checks from the input arns"""
checks_from_arn = set()
is_subservice_in_checks = False
# Handle if there are audit resources so only their services are executed
if self.audit_resources:
services_without_subservices = ["guardduty", "kms", "s3", "elb", "efs"]
service_list = set()
sub_service_list = set()
for resource in self.audit_resources:
service = resource.split(":")[2]
sub_service = resource.split(":")[5].split("/")[0].replace("-", "_")
# WAF Services does not have checks
if service != "wafv2" and service != "waf":
# Parse services when they are different in the ARNs
if service == "lambda":
service = "awslambda"
elif service == "elasticloadbalancing":
service = "elb"
elif service == "elasticfilesystem":
service = "efs"
elif service == "logs":
service = "cloudwatch"
# Check if Prowler has checks in service
try:
list_modules(self.provider, service)
except ModuleNotFoundError:
# Service is not supported
pass
else:
service_list.add(service)
# Get subservices to execute only applicable checks
if service not in services_without_subservices:
# Parse some specific subservices
if service == "ec2":
if sub_service == "security_group":
sub_service = "securitygroup"
if sub_service == "network_acl":
sub_service = "networkacl"
if sub_service == "image":
sub_service = "ami"
if service == "rds":
if sub_service == "cluster_snapshot":
sub_service = "snapshot"
sub_service_list.add(sub_service)
else:
sub_service_list.add(service)
checks = recover_checks_from_service(service_list, self.provider)
# Filter only checks with audited subservices
for check in checks:
if any(sub_service in check for sub_service in sub_service_list):
if not (sub_service == "policy" and "password_policy" in check):
checks_from_arn.add(check)
is_subservice_in_checks = True
if not is_subservice_in_checks:
checks_from_arn = checks
# Return final checks list
return sorted(checks_from_arn)
def get_regions_from_audit_resources(audit_resources: list) -> set:
"""get_regions_from_audit_resources gets the regions from the audit resources arns"""
audited_regions = set()
for resource in audit_resources:
region = resource.split(":")[3]
if region:
audited_regions.add(region)
return audited_regions
def get_tagged_resources(self, input_resource_tags: list):
"""
get_tagged_resources returns a list of the resources that are going to be scanned based on the given input tags
"""
try:
resource_tags = []
tagged_resources = []
for tag in input_resource_tags:
key = tag.split("=")[0]
value = tag.split("=")[1]
resource_tags.append({"Key": key, "Values": [value]})
# Get Resources with resource_tags for all regions
for regional_client in self.generate_regional_clients(
"resourcegroupstaggingapi"
).values():
try:
get_resources_paginator = regional_client.get_paginator(
"get_resources"
)
for page in get_resources_paginator.paginate(
TagFilters=resource_tags
):
for resource in page["ResourceTagMappingList"]:
tagged_resources.append(resource["ResourceARN"])
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
sys.exit(1)
else:
return tagged_resources
def get_default_region(self, service: str) -> str:
"""get_default_region gets the default region based on the profile and audited service regions"""
service_regions = self.get_available_aws_service_regions(service)
default_region = (
self.get_global_region()
) # global region of the partition when all regions are audited and there is no profile region
if self.identity.profile_region in service_regions:
# return profile region only if it is audited
default_region = self.identity.profile_region
# return first audited region if specific regions are audited
elif self.identity.audited_regions:
default_region = self.identity.audited_regions[0]
return default_region
def get_global_region(self) -> str:
"""get_global_region gets the global region based on the audited partition"""
global_region = "us-east-1"
if self.identity.partition == "aws-cn":
global_region = "cn-north-1"
elif self.identity.partition == "aws-us-gov":
global_region = "us-gov-east-1"
elif "aws-iso" in self.identity.partition:
global_region = "aws-iso-global"
return global_region
def __input_role_mfa_token_and_code__() -> tuple[str]:
"""input_role_mfa_token_and_code ask for the AWS MFA ARN and TOTP and returns it."""
mfa_ARN = input("Enter ARN of MFA: ")
mfa_TOTP = input("Enter MFA code: ")
return (mfa_ARN.strip(), mfa_TOTP.strip())
def __set_session_config__(self, aws_retries_max_attempts: bool):
session_config = Config(
retries={"max_attempts": 3, "mode": "standard"},
user_agent_extra=BOTO3_USER_AGENT_EXTRA,
)
if aws_retries_max_attempts:
# Create the new config
config = Config(
retries={
"max_attempts": aws_retries_max_attempts,
"mode": "standard",
},
)
# Merge the new configuration
session_config = self.session.session_config.merge(config)
return session_config
def get_aws_enabled_regions(audit_info: AWS_Audit_Info) -> set:
"""get_aws_enabled_regions returns a set of enabled AWS regions"""
def __assume_role__(
self,
session,
sts_endpoint_region: str,
) -> dict:
try:
assume_role_arguments = {
"RoleArn": self.assumed_role.assumed_role_info.role_arn,
"RoleSessionName": "ProwlerAsessmentSession",
"DurationSeconds": self.assumed_role.assumed_role_info.session_duration,
}
# EC2 Client to check enabled regions
service = "ec2"
default_region = get_default_region(service, audit_info)
ec2_client = audit_info.audit_session.client(service, region_name=default_region)
# Set the info to assume the role from the partition, account and role name
if self.assumed_role.assumed_role_info.external_id:
assume_role_arguments["ExternalId"] = (
self.assumed_role.assumed_role_info.external_id
)
enabled_regions = set()
try:
# With AllRegions=False we only get the enabled regions for the account
for region in ec2_client.describe_regions(AllRegions=False).get("Regions", []):
enabled_regions.add(region.get("RegionName"))
except Exception as error:
logger.warning(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return enabled_regions
if self.assumed_role.assumed_role_info.mfa_enabled:
mfa_ARN, mfa_TOTP = self.__input_role_mfa_token_and_code__()
assume_role_arguments["SerialNumber"] = mfa_ARN
assume_role_arguments["TokenCode"] = mfa_TOTP
# Set the STS Endpoint Region
if sts_endpoint_region is None:
sts_endpoint_region = AWS_STS_GLOBAL_ENDPOINT_REGION
sts_client = create_sts_session(session, sts_endpoint_region)
assumed_credentials = sts_client.assume_role(**assume_role_arguments)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
sys.exit(1)
else:
return assumed_credentials
def get_aws_enabled_regions(self) -> set:
"""get_aws_enabled_regions returns a set of enabled AWS regions"""
# EC2 Client to check enabled regions
service = "ec2"
default_region = self.get_default_region(service)
ec2_client = self.session.session.client(service, region_name=default_region)
enabled_regions = set()
try:
# With AllRegions=False we only get the enabled regions for the account
for region in ec2_client.describe_regions(AllRegions=False).get(
"Regions", []
):
enabled_regions.add(region.get("RegionName"))
except Exception as error:
logger.warning(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return enabled_regions
def get_checks_to_execute_by_audit_resources(self) -> set[str]:
# Once the audit_info is set and we have the eventual checks from arn, it is time to exclude the others
try:
checks = set()
if self.audit_resources:
self.identity.audited_regions = self.get_regions_from_audit_resources(
self.audit_resources
)
checks = self.get_checks_from_input_arn()
return checks
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
sys.exit(1)
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 get_aws_available_regions():
@@ -228,123 +634,3 @@ def get_aws_available_regions():
except Exception as error:
logger.error(f"{error.__class__.__name__}: {error}")
return []
def get_checks_from_input_arn(audit_resources: list, provider: str) -> set:
"""get_checks_from_input_arn gets the list of checks from the input arns"""
checks_from_arn = set()
is_subservice_in_checks = False
# Handle if there are audit resources so only their services are executed
if audit_resources:
services_without_subservices = ["guardduty", "kms", "s3", "elb", "efs"]
service_list = set()
sub_service_list = set()
for resource in audit_resources:
service = resource.split(":")[2]
sub_service = resource.split(":")[5].split("/")[0].replace("-", "_")
# WAF Services does not have checks
if service != "wafv2" and service != "waf":
# Parse services when they are different in the ARNs
if service == "lambda":
service = "awslambda"
elif service == "elasticloadbalancing":
service = "elb"
elif service == "elasticfilesystem":
service = "efs"
elif service == "logs":
service = "cloudwatch"
elif service == "cognito":
service = "cognito-idp"
# Check if Prowler has checks in service
try:
list_modules(provider, service)
except ModuleNotFoundError:
# Service is not supported
pass
else:
service_list.add(service)
# Get subservices to execute only applicable checks
if service not in services_without_subservices:
# Parse some specific subservices
if service == "ec2":
if sub_service == "security_group":
sub_service = "securitygroup"
if sub_service == "network_acl":
sub_service = "networkacl"
if sub_service == "image":
sub_service = "ami"
if service == "rds":
if sub_service == "cluster_snapshot":
sub_service = "snapshot"
sub_service_list.add(sub_service)
else:
sub_service_list.add(service)
checks = recover_checks_from_service(service_list, provider)
# Filter only checks with audited subservices
for check in checks:
if any(sub_service in check for sub_service in sub_service_list):
if not (sub_service == "policy" and "password_policy" in check):
checks_from_arn.add(check)
is_subservice_in_checks = True
if not is_subservice_in_checks:
checks_from_arn = checks
# Return final checks list
return sorted(checks_from_arn)
def get_regions_from_audit_resources(audit_resources: list) -> set:
"""get_regions_from_audit_resources gets the regions from the audit resources arns"""
audited_regions = set()
for resource in audit_resources:
region = resource.split(":")[3]
if region:
audited_regions.add(region)
return audited_regions
def get_available_aws_service_regions(service: str, audit_info: AWS_Audit_Info) -> set:
# Get json locally
actual_directory = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
with open_file(f"{actual_directory}/{aws_services_json_file}") as f:
data = parse_json_file(f)
json_regions = set(
data["services"][service]["regions"][audit_info.audited_partition]
)
# Check for input aws audit_info.audited_regions
if audit_info.audited_regions:
# Get common regions between input and json
regions = json_regions.intersection(audit_info.audited_regions)
else: # Get all regions from json of the service and partition
regions = json_regions
return regions
def get_default_region(service: str, audit_info: AWS_Audit_Info) -> str:
"""get_default_region gets the default region based on the profile and audited service regions"""
service_regions = get_available_aws_service_regions(service, audit_info)
default_region = get_global_region(
audit_info
) # global region of the partition when all regions are audited and there is no profile region
if audit_info.profile_region in service_regions:
# return profile region only if it is audited
default_region = audit_info.profile_region
# return first audited region if specific regions are audited
elif audit_info.audited_regions:
default_region = audit_info.audited_regions[0]
return default_region
def get_global_region(audit_info: AWS_Audit_Info) -> str:
"""get_global_region gets the global region based on the audited partition"""
global_region = "us-east-1"
if audit_info.audited_partition == "aws-cn":
global_region = "cn-north-1"
elif audit_info.audited_partition == "aws-us-gov":
global_region = "us-gov-east-1"
elif "aws-iso" in audit_info.audited_partition:
global_region = "aws-iso-global"
return global_region
prowler/providers/aws/aws_provider_new.py
-621
View File
@@ -1,621 +0,0 @@
import os
import pathlib
import sys
from argparse import Namespace
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Optional
from boto3 import client, session
from botocore.config import Config
from botocore.credentials import RefreshableCredentials
from botocore.session import get_session
from colorama import Fore, Style
from prowler.config.config import aws_services_json_file
from prowler.lib.check.check import list_modules, recover_checks_from_service
from prowler.lib.logger import logger
from prowler.lib.utils.utils import open_file, parse_json_file
from prowler.providers.aws.config import (
AWS_STS_GLOBAL_ENDPOINT_REGION,
BOTO3_USER_AGENT_EXTRA,
)
from prowler.providers.aws.lib.arn.arn import parse_iam_credentials_arn
from prowler.providers.aws.lib.credentials.credentials import (
create_sts_session,
validate_AWSCredentials,
)
from prowler.providers.aws.lib.organizations.organizations import (
get_organizations_metadata,
)
from prowler.providers.common.provider import Provider
@dataclass
class AWSOrganizationsInfo:
account_details_email: str
account_details_name: str
account_details_arn: str
account_details_org: str
account_details_tags: str
@dataclass
class AWSCredentials:
aws_access_key_id: str
aws_session_token: str
aws_secret_access_key: str
expiration: datetime
@dataclass
class AWSAssumeRole:
role_arn: str
session_duration: int
external_id: str
mfa_enabled: bool
@dataclass
class AWSAssumeRoleConfiguration:
assumed_role_info: AWSAssumeRole
assumed_role_credentials: AWSCredentials
@dataclass
class AWSIdentityInfo:
account: str
account_arn: str
user_id: str
partition: str
identity_arn: str
profile: str
profile_region: str
audited_regions: list
@dataclass
class AWSSession:
session: session.Session
session_config: Config
original_session: None
class AwsProvider(Provider):
session: AWSSession = AWSSession(
session=None, session_config=None, original_session=None
)
identity: AWSIdentityInfo = AWSIdentityInfo(
account=None,
account_arn=None,
user_id=None,
partition=None,
identity_arn=None,
profile=None,
profile_region=None,
audited_regions=[],
)
assumed_role: AWSAssumeRoleConfiguration = AWSAssumeRoleConfiguration(
assumed_role_info=AWSAssumeRole(
role_arn=None,
session_duration=None,
external_id=None,
mfa_enabled=False,
),
assumed_role_credentials=AWSCredentials(
aws_access_key_id=None,
aws_session_token=None,
aws_secret_access_key=None,
expiration=None,
),
)
organizations_metadata: AWSOrganizationsInfo = AWSOrganizationsInfo(
account_details_email=None,
account_details_name=None,
account_details_arn=None,
account_details_org=None,
account_details_tags=None,
)
audit_resources: Optional[Any]
audit_metadata: Optional[Any]
audit_config: dict = {}
mfa_enabled: bool = False
ignore_unused_services: bool = False
def __init__(self, arguments: Namespace):
logger.info("Setting AWS provider ...")
# Parse input arguments
# Assume Role Options
input_role = getattr(arguments, "role", None)
input_session_duration = getattr(arguments, "session_duration", None)
input_external_id = getattr(arguments, "external_id", None)
# STS Endpoint Region
sts_endpoint_region = getattr(arguments, "sts_endpoint_region", None)
# MFA Configuration (false by default)
input_mfa = getattr(arguments, "mfa", None)
input_profile = getattr(arguments, "profile", None)
input_regions = getattr(arguments, "region", None)
organizations_role_arn = getattr(arguments, "organizations_role", None)
# Set the maximum retries for the standard retrier config
aws_retries_max_attempts = getattr(arguments, "aws_retries_max_attempts", None)
# Set if unused services must be ignored
ignore_unused_services = getattr(arguments, "ignore_unused_services", None)
# Set the maximum retries for the standard retrier config
self.session.session_config = self.__set_session_config__(
aws_retries_max_attempts
)
# Set ignore unused services
self.ignore_unused_services = ignore_unused_services
# Start populating AWS identity object
self.identity.profile = input_profile
self.identity.audited_regions = input_regions
# We need to create an original sessions using regular auth path (creds, profile, etc)
logger.info("Generating original session ...")
self.session.session = self.setup_session(input_mfa)
# After the session is created, validate it
logger.info("Validating credentials ...")
caller_identity = validate_AWSCredentials(
self.session.session, input_regions, sts_endpoint_region
)
logger.info("Credentials validated")
logger.info(f"Original caller identity UserId: {caller_identity['UserId']}")
logger.info(f"Original caller identity ARN: {caller_identity['Arn']}")
# Set values of AWS identity object
self.identity.account = caller_identity["Account"]
self.identity.identity_arn = caller_identity["Arn"]
self.identity.user_id = caller_identity["UserId"]
self.identity.partition = parse_iam_credentials_arn(
caller_identity["Arn"]
).partition
self.identity.account_arn = (
f"arn:{self.identity.partition}:iam::{self.identity.account}:root"
)
# save original session
self.session.original_session = self.session.session
# time for checking role assumption
if input_role:
# session will be the assumed one
self.session.session = self.setup_assumed_session(
input_role,
input_external_id,
input_mfa,
input_session_duration,
sts_endpoint_region,
)
logger.info("Audit session is the new session created assuming role")
# check if organizations info is gonna be retrieved
if organizations_role_arn:
logger.info(
f"Getting organizations metadata for account {organizations_role_arn}"
)
# session will be the assumed one with organizations permissions
self.session.session = self.setup_assumed_session(
organizations_role_arn,
input_external_id,
input_mfa,
input_session_duration,
sts_endpoint_region,
)
self.organizations_metadata = get_organizations_metadata(
self.identity.account, self.assumed_role.assumed_role_credentials
)
logger.info("Organizations metadata retrieved")
if self.session.session.region_name:
self.identity.profile_region = self.session.session.region_name
else:
self.identity.profile_region = "us-east-1"
if not getattr(arguments, "only_logs", None):
self.print_credentials()
# Parse Scan Tags
if getattr(arguments, "resource_tags", None):
input_resource_tags = arguments.resource_tags
self.audit_resources = self.get_tagged_resources(input_resource_tags)
# Parse Input Resource ARNs
self.audit_resources = getattr(arguments, "resource_arn", None)
def setup_session(self, input_mfa: bool):
logger.info("Creating regular session ...")
# Input MFA only if a role is not going to be assumed
if input_mfa and not self.assumed_role.assumed_role_info.role_arn:
mfa_ARN, mfa_TOTP = self.__input_role_mfa_token_and_code__()
get_session_token_arguments = {
"SerialNumber": mfa_ARN,
"TokenCode": mfa_TOTP,
}
sts_client = client("sts")
session_credentials = sts_client.get_session_token(
**get_session_token_arguments
)
return session.Session(
aws_access_key_id=session_credentials["Credentials"]["AccessKeyId"],
aws_secret_access_key=session_credentials["Credentials"][
"SecretAccessKey"
],
aws_session_token=session_credentials["Credentials"]["SessionToken"],
profile_name=self.identity.profile,
)
else:
return session.Session(
profile_name=self.identity.profile,
)
def setup_assumed_session(
self,
input_role: str,
input_external_id: str,
input_mfa: str,
session_duration: int,
sts_endpoint_region: str,
):
logger.info("Creating assumed session ...")
# store information about the role is gonna be assumed
self.assumed_role.assumed_role_info.role_arn = input_role
self.assumed_role.assumed_role_info.session_duration = session_duration
self.assumed_role.assumed_role_info.external_id = input_external_id
self.assumed_role.assumed_role_info.mfa_enabled = input_mfa
# Check if role arn is valid
try:
# this returns the arn already parsed into a dict to be used when it is needed to access its fields
role_arn_parsed = parse_iam_credentials_arn(
self.assumed_role.assumed_role_info.role_arn
)
except Exception as error:
logger.critical(f"{error.__class__.__name__} -- {error}")
sys.exit(1)
else:
logger.info(f"Assuming role {self.assumed_role.assumed_role_info.role_arn}")
# Assume the role
assumed_role_response = self.__assume_role__(
self.session.session,
sts_endpoint_region,
)
logger.info("Role assumed")
# Set the info needed to create a session with an assumed role
self.assumed_role.assumed_role_credentials = AWSCredentials(
aws_access_key_id=assumed_role_response["Credentials"]["AccessKeyId"],
aws_session_token=assumed_role_response["Credentials"]["SessionToken"],
aws_secret_access_key=assumed_role_response["Credentials"][
"SecretAccessKey"
],
expiration=assumed_role_response["Credentials"]["Expiration"],
)
# Set identity parameters
self.identity.account = role_arn_parsed.account_id
self.identity.partition = role_arn_parsed.partition
self.identity.account_arn = (
f"arn:{self.identity.partition}:iam::{self.identity.account}:root"
)
# From botocore we can use RefreshableCredentials class, which has an attribute (refresh_using)
# that needs to be a method without arguments that retrieves a new set of fresh credentials
# asuming the role again. -> https://github.com/boto/botocore/blob/098cc255f81a25b852e1ecdeb7adebd94c7b1b73/botocore/credentials.py#L395
assumed_refreshable_credentials = RefreshableCredentials(
access_key=self.assumed_role.assumed_role_credentials.aws_access_key_id,
secret_key=self.assumed_role.assumed_role_credentials.aws_secret_access_key,
token=self.assumed_role.assumed_role_credentials.aws_session_token,
expiry_time=self.assumed_role.assumed_role_credentials.expiration,
refresh_using=self.refresh_credentials,
method="sts-assume-role",
)
# Here we need the botocore session since it needs to use refreshable credentials
assumed_botocore_session = get_session()
assumed_botocore_session._credentials = assumed_refreshable_credentials
assumed_botocore_session.set_config_variable(
"region", self.identity.profile_region
)
return session.Session(
profile_name=self.identity.profile,
botocore_session=assumed_botocore_session,
)
# Refresh credentials method using assume role
# This method is called "adding ()" to the name, so it cannot accept arguments
# https://github.com/boto/botocore/blob/098cc255f81a25b852e1ecdeb7adebd94c7b1b73/botocore/credentials.py#L570
def refresh_credentials(self):
logger.info("Refreshing assumed credentials...")
response = self.__assume_role__(self.aws_session, self.role_info)
refreshed_credentials = dict(
# Keys of the dict has to be the same as those that are being searched in the parent class
# https://github.com/boto/botocore/blob/098cc255f81a25b852e1ecdeb7adebd94c7b1b73/botocore/credentials.py#L609
access_key=response["Credentials"]["AccessKeyId"],
secret_key=response["Credentials"]["SecretAccessKey"],
token=response["Credentials"]["SessionToken"],
expiry_time=response["Credentials"]["Expiration"].isoformat(),
)
logger.info("Refreshed Credentials:")
logger.info(refreshed_credentials)
return refreshed_credentials
def print_credentials(self):
# Beautify audited regions, set "all" if there is no filter region
regions = (
", ".join(self.identity.audited_regions)
if self.identity.audited_regions is not None
else "all"
)
# Beautify audited profile, set "default" if there is no profile set
profile = (
self.identity.profile if self.identity.profile is not None else "default"
)
report = f"""
This report is being generated using credentials below:
AWS-CLI Profile: {Fore.YELLOW}[{profile}]{Style.RESET_ALL} AWS Filter Region: {Fore.YELLOW}[{regions}]{Style.RESET_ALL}
AWS Account: {Fore.YELLOW}[{self.identity.account}]{Style.RESET_ALL} UserId: {Fore.YELLOW}[{self.identity.user_id}]{Style.RESET_ALL}
Caller Identity ARN: {Fore.YELLOW}[{self.identity.identity_arn}]{Style.RESET_ALL}
"""
# If -A is set, print Assumed Role ARN
if self.assumed_role.assumed_role_info.role_arn is not None:
report += f"""Assumed Role ARN: {Fore.YELLOW}[{self.assumed_role.assumed_role_info.role_arn}]{Style.RESET_ALL}
"""
print(report)
def generate_regional_clients(
self, service: str, global_service: bool = False
) -> dict:
try:
regional_clients = {}
service_regions = self.get_available_aws_service_regions(service)
# Check if it is global service to gather only one region
if global_service:
if service_regions:
if self.identity.profile_region in service_regions:
service_regions = [self.identity.profile_region]
service_regions = service_regions[:1]
for region in service_regions:
regional_client = self.session.session.client(
service, region_name=region, config=self.session.session_config
)
regional_client.region = region
regional_clients[region] = regional_client
return regional_clients
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def get_available_aws_service_regions(self, service: str) -> list:
# Get json locally
actual_directory = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
with open_file(f"{actual_directory}/{aws_services_json_file}") as f:
data = parse_json_file(f)
# Check if it is a subservice
json_regions = data["services"][service]["regions"][self.identity.partition]
if (
self.identity.audited_regions
): # Check for input aws audit_info.audited_regions
regions = list(
set(json_regions).intersection(self.identity.audited_regions)
) # Get common regions between input and json
else: # Get all regions from json of the service and partition
regions = json_regions
return regions
def get_aws_available_regions():
try:
actual_directory = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
with open_file(f"{actual_directory}/{aws_services_json_file}") as f:
data = parse_json_file(f)
regions = set()
for service in data["services"].values():
for partition in service["regions"]:
for item in service["regions"][partition]:
regions.add(item)
return list(regions)
except Exception as error:
logger.error(f"{error.__class__.__name__}: {error}")
return []
def get_checks_from_input_arn(audit_resources: list, provider: str) -> set:
"""get_checks_from_input_arn gets the list of checks from the input arns"""
checks_from_arn = set()
is_subservice_in_checks = False
# Handle if there are audit resources so only their services are executed
if audit_resources:
services_without_subservices = ["guardduty", "kms", "s3", "elb", "efs"]
service_list = set()
sub_service_list = set()
for resource in audit_resources:
service = resource.split(":")[2]
sub_service = resource.split(":")[5].split("/")[0].replace("-", "_")
# WAF Services does not have checks
if service != "wafv2" and service != "waf":
# Parse services when they are different in the ARNs
if service == "lambda":
service = "awslambda"
elif service == "elasticloadbalancing":
service = "elb"
elif service == "elasticfilesystem":
service = "efs"
elif service == "logs":
service = "cloudwatch"
# Check if Prowler has checks in service
try:
list_modules(provider, service)
except ModuleNotFoundError:
# Service is not supported
pass
else:
service_list.add(service)
# Get subservices to execute only applicable checks
if service not in services_without_subservices:
# Parse some specific subservices
if service == "ec2":
if sub_service == "security_group":
sub_service = "securitygroup"
if sub_service == "network_acl":
sub_service = "networkacl"
if sub_service == "image":
sub_service = "ami"
if service == "rds":
if sub_service == "cluster_snapshot":
sub_service = "snapshot"
sub_service_list.add(sub_service)
else:
sub_service_list.add(service)
checks = recover_checks_from_service(service_list, provider)
# Filter only checks with audited subservices
for check in checks:
if any(sub_service in check for sub_service in sub_service_list):
if not (sub_service == "policy" and "password_policy" in check):
checks_from_arn.add(check)
is_subservice_in_checks = True
if not is_subservice_in_checks:
checks_from_arn = checks
# Return final checks list
return sorted(checks_from_arn)
def get_regions_from_audit_resources(audit_resources: list) -> set:
"""get_regions_from_audit_resources gets the regions from the audit resources arns"""
audited_regions = set()
for resource in audit_resources:
region = resource.split(":")[3]
if region:
audited_regions.add(region)
return audited_regions
def get_tagged_resources(self, input_resource_tags: list):
"""
get_tagged_resources returns a list of the resources that are going to be scanned based on the given input tags
"""
try:
resource_tags = []
tagged_resources = []
for tag in input_resource_tags:
key = tag.split("=")[0]
value = tag.split("=")[1]
resource_tags.append({"Key": key, "Values": [value]})
# Get Resources with resource_tags for all regions
for regional_client in self.generate_regional_clients(
"resourcegroupstaggingapi"
).values():
try:
get_resources_paginator = regional_client.get_paginator(
"get_resources"
)
for page in get_resources_paginator.paginate(
TagFilters=resource_tags
):
for resource in page["ResourceTagMappingList"]:
tagged_resources.append(resource["ResourceARN"])
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
sys.exit(1)
else:
return tagged_resources
def get_default_region(self, service: str) -> str:
"""get_default_region gets the default region based on the profile and audited service regions"""
service_regions = self.get_available_aws_service_regions(service)
default_region = (
self.get_global_region()
) # global region of the partition when all regions are audited and there is no profile region
if self.identity.profile_region in service_regions:
# return profile region only if it is audited
default_region = self.identity.profile_region
# return first audited region if specific regions are audited
elif self.identity.audited_regions:
default_region = self.identity.audited_regions[0]
return default_region
def get_global_region(self) -> str:
"""get_global_region gets the global region based on the audited partition"""
global_region = "us-east-1"
if self.identity.partition == "aws-cn":
global_region = "cn-north-1"
elif self.identity.partition == "aws-us-gov":
global_region = "us-gov-east-1"
elif "aws-iso" in self.identity.partition:
global_region = "aws-iso-global"
return global_region
def __input_role_mfa_token_and_code__() -> tuple[str]:
"""input_role_mfa_token_and_code ask for the AWS MFA ARN and TOTP and returns it."""
mfa_ARN = input("Enter ARN of MFA: ")
mfa_TOTP = input("Enter MFA code: ")
return (mfa_ARN.strip(), mfa_TOTP.strip())
def __set_session_config__(self, aws_retries_max_attempts: bool):
session_config = Config(
retries={"max_attempts": 3, "mode": "standard"},
user_agent_extra=BOTO3_USER_AGENT_EXTRA,
)
if aws_retries_max_attempts:
# Create the new config
config = Config(
retries={
"max_attempts": aws_retries_max_attempts,
"mode": "standard",
},
)
# Merge the new configuration
session_config = self.session.session_config.merge(config)
return session_config
def __assume_role__(
self,
session,
sts_endpoint_region: str,
) -> dict:
try:
assume_role_arguments = {
"RoleArn": self.assumed_role.assumed_role_info.role_arn,
"RoleSessionName": "ProwlerAsessmentSession",
"DurationSeconds": self.assumed_role.assumed_role_info.session_duration,
}
# Set the info to assume the role from the partition, account and role name
if self.assumed_role.assumed_role_info.external_id:
assume_role_arguments["ExternalId"] = (
self.assumed_role.assumed_role_info.external_id
)
if self.assumed_role.assumed_role_info.mfa_enabled:
mfa_ARN, mfa_TOTP = self.__input_role_mfa_token_and_code__()
assume_role_arguments["SerialNumber"] = mfa_ARN
assume_role_arguments["TokenCode"] = mfa_TOTP
# Set the STS Endpoint Region
if sts_endpoint_region is None:
sts_endpoint_region = AWS_STS_GLOBAL_ENDPOINT_REGION
sts_client = create_sts_session(session, sts_endpoint_region)
assumed_credentials = sts_client.assume_role(**assume_role_arguments)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
sys.exit(1)
else:
return assumed_credentials
prowler/providers/aws/lib/mutelist/mutelist.py
+6 -14
View File
@@ -32,22 +32,20 @@ mutelist_schema = Schema(
)
def parse_mutelist_file(audit_info, mutelist_file):
def parse_mutelist_file(session, aws_account, mutelist_file):
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:])
s3_client = audit_info.audit_session.client("s3")
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]
lambda_client = audit_info.audit_session.client(
"lambda", region_name=lambda_region
)
lambda_client = session.client("lambda", region_name=lambda_region)
lambda_response = lambda_client.invoke(
FunctionName=mutelist_file, InvocationType="RequestResponse"
)
@@ -60,23 +58,17 @@ def parse_mutelist_file(audit_info, mutelist_file):
):
mutelist = {"Accounts": {}}
table_region = mutelist_file.split(":")[3]
dynamodb_resource = audit_info.audit_session.resource(
"dynamodb", region_name=table_region
)
dynamodb_resource = session.resource("dynamodb", region_name=table_region)
dynamo_table = dynamodb_resource.Table(mutelist_file.split("/")[1])
response = dynamo_table.scan(
FilterExpression=Attr("Accounts").is_in(
[audit_info.audited_account, "*"]
)
FilterExpression=Attr("Accounts").is_in([aws_account, "*"])
)
dynamodb_items = response["Items"]
# Paginate through all results
while "LastEvaluatedKey" in dynamodb_items:
response = dynamo_table.scan(
ExclusiveStartKey=response["LastEvaluatedKey"],
FilterExpression=Attr("Accounts").is_in(
[audit_info.audited_account, "*"]
),
FilterExpression=Attr("Accounts").is_in([aws_account, "*"]),
)
dynamodb_items.update(response["Items"])
for item in dynamodb_items:
prowler/providers/aws/lib/service/service.py
+1 -1
View File
@@ -1,7 +1,7 @@
from concurrent.futures import ThreadPoolExecutor, as_completed
from prowler.lib.logger import logger
from prowler.providers.aws.aws_provider_new import AwsProvider
from prowler.providers.aws.aws_provider import AwsProvider
# TODO: review the following code
# from prowler.providers.aws.aws_provider import (
prowler/providers/aws/models.py
+55
View File
@@ -0,0 +1,55 @@
from dataclasses import dataclass
from datetime import datetime
from boto3.session import Session
from botocore.config import Config
@dataclass
class AWSOrganizationsInfo:
account_details_email: str
account_details_name: str
account_details_arn: str
account_details_org: str
account_details_tags: str
@dataclass
class AWSCredentials:
aws_access_key_id: str
aws_session_token: str
aws_secret_access_key: str
expiration: datetime
@dataclass
class AWSAssumeRole:
role_arn: str
session_duration: int
external_id: str
mfa_enabled: bool
@dataclass
class AWSAssumeRoleConfiguration:
assumed_role_info: AWSAssumeRole
assumed_role_credentials: AWSCredentials
@dataclass
class AWSIdentityInfo:
account: str
account_arn: str
user_id: str
partition: str
identity_arn: str
profile: str
profile_region: str
audited_regions: list
@dataclass
class AWSSession:
session: Session
session_config: Config
original_session: None
prowler/providers/common/audit_info.py
+214 -251
View File
@@ -1,28 +1,9 @@
import sys
from botocore.config import Config
from colorama import Fore, Style
from prowler.config.config import load_and_validate_config_file
from prowler.lib.logger import logger
from prowler.providers.aws.aws_provider import (
AWS_Provider,
assume_role,
get_aws_enabled_regions,
get_checks_from_input_arn,
get_regions_from_audit_resources,
)
from prowler.providers.aws.lib.arn.arn import parse_iam_credentials_arn
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info, AWSCredentials
from prowler.providers.aws.lib.credentials.credentials import validate_AWSCredentials
from prowler.providers.aws.lib.organizations.organizations import (
get_organizations_metadata,
parse_organizations_metadata,
)
from prowler.providers.aws.lib.resource_api_tagging.resource_api_tagging import (
get_tagged_resources,
)
from prowler.providers.azure.azure_provider import Azure_Provider
from prowler.providers.azure.lib.audit_info.audit_info import azure_audit_info
from prowler.providers.azure.lib.audit_info.models import (
@@ -81,245 +62,247 @@ Azure Identity Type: {Fore.YELLOW}[{audit_info.identity.identity_type}]{Style.RE
"""
print(report)
def set_aws_audit_info(self, arguments) -> AWS_Audit_Info:
"""
set_aws_audit_info returns the AWS_Audit_Info
"""
logger.info("Setting AWS session ...")
# TODO: remove if not needed, but not now :)
# def set_aws_audit_info(self, arguments) -> AWS_Audit_Info:
# """
# set_aws_audit_info returns the AWS_Audit_Info
# """
# logger.info("Setting AWS session ...")
# Assume Role Options
input_role = arguments.get("role")
current_audit_info.assumed_role_info.role_arn = input_role
input_session_duration = arguments.get("session_duration")
input_external_id = arguments.get("external_id")
input_role_session_name = arguments.get("role_session_name")
# # Assume Role Options
# input_role = arguments.get("role")
# current_audit_info.assumed_role_info.role_arn = input_role
# input_session_duration = arguments.get("session_duration")
# input_external_id = arguments.get("external_id")
# input_role_session_name = arguments.get("role_session_name")
# STS Endpoint Region
sts_endpoint_region = arguments.get("sts_endpoint_region")
# # STS Endpoint Region
# sts_endpoint_region = arguments.get("sts_endpoint_region")
# MFA Configuration (false by default)
input_mfa = arguments.get("mfa")
current_audit_info.mfa_enabled = input_mfa
# # MFA Configuration (false by default)
# input_mfa = arguments.get("mfa")
# current_audit_info.mfa_enabled = input_mfa
input_profile = arguments.get("profile")
input_regions = arguments.get("region")
organizations_role_arn = arguments.get("organizations_role")
# input_profile = arguments.get("profile")
# input_regions = arguments.get("region")
# organizations_role_arn = arguments.get("organizations_role")
# Assumed AWS session
assumed_session = None
# # Assumed AWS session
# assumed_session = None
# Set the maximum retries for the standard retrier config
aws_retries_max_attempts = arguments.get("aws_retries_max_attempts")
if aws_retries_max_attempts:
# Create the new config
config = Config(
retries={
"max_attempts": aws_retries_max_attempts,
"mode": "standard",
},
)
# Merge the new configuration
new_boto3_config = current_audit_info.session_config.merge(config)
current_audit_info.session_config = new_boto3_config
# # Set the maximum retries for the standard retrier config
# aws_retries_max_attempts = arguments.get("aws_retries_max_attempts")
# if aws_retries_max_attempts:
# # Create the new config
# config = Config(
# retries={
# "max_attempts": aws_retries_max_attempts,
# "mode": "standard",
# },
# )
# # Merge the new configuration
# new_boto3_config = current_audit_info.session_config.merge(config)
# current_audit_info.session_config = new_boto3_config
# Set ignore unused services argument
current_audit_info.ignore_unused_services = arguments.get(
"ignore_unused_services"
)
# # Set ignore unused services argument
# current_audit_info.ignore_unused_services = arguments.get(
# "ignore_unused_services"
# )
# Setting session
current_audit_info.profile = input_profile
current_audit_info.audited_regions = input_regions
# # Setting session
# current_audit_info.profile = input_profile
# current_audit_info.audited_regions = input_regions
logger.info("Generating original session ...")
# Create an global original session using only profile/basic credentials info
aws_provider = AWS_Provider(current_audit_info)
current_audit_info.original_session = aws_provider.aws_session
logger.info("Validating credentials ...")
# Verificate if we have valid credentials
caller_identity = validate_AWSCredentials(
current_audit_info.original_session, input_regions, sts_endpoint_region
)
# logger.info("Generating original session ...")
# # Create an global original session using only profile/basic credentials info
# aws_provider = AWS_Provider(current_audit_info)
# current_audit_info.original_session = aws_provider.aws_session
# logger.info("Validating credentials ...")
# # Verificate if we have valid credentials
# caller_identity = validate_AWSCredentials(
# current_audit_info.original_session, input_regions, sts_endpoint_region
# )
logger.info("Credentials validated")
logger.info(f"Original caller identity UserId: {caller_identity['UserId']}")
logger.info(f"Original caller identity ARN: {caller_identity['Arn']}")
# logger.info("Credentials validated")
# logger.info(f"Original caller identity UserId: {caller_identity['UserId']}")
# logger.info(f"Original caller identity ARN: {caller_identity['Arn']}")
current_audit_info.audited_account = caller_identity["Account"]
current_audit_info.audited_identity_arn = caller_identity["Arn"]
current_audit_info.audited_user_id = caller_identity["UserId"]
current_audit_info.audited_partition = parse_iam_credentials_arn(
caller_identity["Arn"]
).partition
current_audit_info.audited_account_arn = f"arn:{current_audit_info.audited_partition}:iam::{current_audit_info.audited_account}:root"
# current_audit_info.audited_account = caller_identity["Account"]
# current_audit_info.audited_identity_arn = caller_identity["Arn"]
# current_audit_info.audited_user_id = caller_identity["UserId"]
# current_audit_info.audited_partition = parse_iam_credentials_arn(
# caller_identity["Arn"]
# ).partition
# current_audit_info.audited_account_arn = f"arn:{current_audit_info.audited_partition}:iam::{current_audit_info.audited_account}:root"
logger.info("Checking if role assumption is needed ...")
if input_role:
current_audit_info.assumed_role_info.role_arn = input_role
current_audit_info.assumed_role_info.session_duration = (
input_session_duration
)
current_audit_info.assumed_role_info.external_id = input_external_id
current_audit_info.assumed_role_info.mfa_enabled = input_mfa
current_audit_info.assumed_role_info.role_session_name = (
input_role_session_name
)
# logger.info("Checking if role assumption is needed ...")
# if input_role:
# current_audit_info.assumed_role_info.role_arn = input_role
# current_audit_info.assumed_role_info.session_duration = (
# input_session_duration
# )
# current_audit_info.assumed_role_info.external_id = input_external_id
# current_audit_info.assumed_role_info.mfa_enabled = input_mfa
# current_audit_info.assumed_role_info.role_session_name = (
# input_role_session_name
# )
# Check if role arn is valid
try:
# this returns the arn already parsed into a dict to be used when it is needed to access its fields
role_arn_parsed = parse_iam_credentials_arn(
current_audit_info.assumed_role_info.role_arn
)
# # Check if role arn is valid
# try:
# # this returns the arn already parsed into a dict to be used when it is needed to access its fields
# role_arn_parsed = parse_iam_credentials_arn(
# current_audit_info.assumed_role_info.role_arn
# )
except Exception as error:
logger.critical(f"{error.__class__.__name__} -- {error}")
sys.exit(1)
# except Exception as error:
# logger.critical(f"{error.__class__.__name__} -- {error}")
# sys.exit(1)
else:
logger.info(
f"Assuming role {current_audit_info.assumed_role_info.role_arn}"
)
# Assume the role
assumed_role_response = assume_role(
aws_provider.aws_session,
aws_provider.role_info,
sts_endpoint_region,
)
logger.info("Role assumed")
# Set the info needed to create a session with an assumed role
current_audit_info.credentials = AWSCredentials(
aws_access_key_id=assumed_role_response["Credentials"][
"AccessKeyId"
],
aws_session_token=assumed_role_response["Credentials"][
"SessionToken"
],
aws_secret_access_key=assumed_role_response["Credentials"][
"SecretAccessKey"
],
expiration=assumed_role_response["Credentials"]["Expiration"],
)
# new session is needed
assumed_session = aws_provider.set_session(current_audit_info)
# else:
# logger.info(
# f"Assuming role {current_audit_info.assumed_role_info.role_arn}"
# )
# # Assume the role
# assumed_role_response = assume_role(
# aws_provider.aws_session,
# aws_provider.role_info,
# sts_endpoint_region,
# )
# logger.info("Role assumed")
# # Set the info needed to create a session with an assumed role
# current_audit_info.credentials = AWSCredentials(
# aws_access_key_id=assumed_role_response["Credentials"][
# "AccessKeyId"
# ],
# aws_session_token=assumed_role_response["Credentials"][
# "SessionToken"
# ],
# aws_secret_access_key=assumed_role_response["Credentials"][
# "SecretAccessKey"
# ],
# expiration=assumed_role_response["Credentials"]["Expiration"],
# )
# # new session is needed
# assumed_session = aws_provider.set_session(current_audit_info)
if assumed_session:
logger.info("Audit session is the new session created assuming role")
current_audit_info.audit_session = assumed_session
current_audit_info.audited_account = role_arn_parsed.account_id
current_audit_info.audited_partition = role_arn_parsed.partition
current_audit_info.audited_account_arn = f"arn:{current_audit_info.audited_partition}:iam::{current_audit_info.audited_account}:root"
else:
logger.info("Audit session is the original one")
current_audit_info.audit_session = current_audit_info.original_session
# if assumed_session:
# logger.info("Audit session is the new session created assuming role")
# current_audit_info.audit_session = assumed_session
# current_audit_info.audited_account = role_arn_parsed.account_id
# current_audit_info.audited_partition = role_arn_parsed.partition
# current_audit_info.audited_account_arn = f"arn:{current_audit_info.audited_partition}:iam::{current_audit_info.audited_account}:root"
# else:
# logger.info("Audit session is the original one")
# current_audit_info.audit_session = current_audit_info.original_session
logger.info("Checking if organizations role assumption is needed ...")
if organizations_role_arn:
current_audit_info.assumed_role_info.role_arn = organizations_role_arn
current_audit_info.assumed_role_info.session_duration = (
input_session_duration
)
current_audit_info.assumed_role_info.external_id = input_external_id
current_audit_info.assumed_role_info.mfa_enabled = input_mfa
# logger.info("Checking if organizations role assumption is needed ...")
# if organizations_role_arn:
# current_audit_info.assumed_role_info.role_arn = organizations_role_arn
# current_audit_info.assumed_role_info.session_duration = (
# input_session_duration
# )
# current_audit_info.assumed_role_info.external_id = input_external_id
# current_audit_info.assumed_role_info.mfa_enabled = input_mfa
# Check if role arn is valid
try:
# this returns the arn already parsed into a dict to be used when it is needed to access its fields
role_arn_parsed = parse_iam_credentials_arn(
current_audit_info.assumed_role_info.role_arn
)
# # Check if role arn is valid
# try:
# # this returns the arn already parsed into a dict to be used when it is needed to access its fields
# role_arn_parsed = parse_iam_credentials_arn(
# current_audit_info.assumed_role_info.role_arn
# )
except Exception as error:
logger.critical(f"{error.__class__.__name__} -- {error}")
sys.exit(1)
# except Exception as error:
# logger.critical(f"{error.__class__.__name__} -- {error}")
# sys.exit(1)
else:
logger.info(
f"Getting organizations metadata for account with IAM Role ARN {organizations_role_arn}"
)
assumed_credentials = assume_role(
aws_provider.aws_session,
aws_provider.role_info,
sts_endpoint_region,
)
organizations_metadata, list_tags_for_resource = (
get_organizations_metadata(
current_audit_info.audited_account, assumed_credentials
)
)
current_audit_info.organizations_metadata = (
parse_organizations_metadata(
organizations_metadata, list_tags_for_resource
)
)
logger.info(
f"Organizations metadata retrieved with IAM Role ARN {organizations_role_arn}"
)
else:
try:
logger.info(
"Getting organizations metadata for account if it is a delegated administrator"
)
organizations_metadata, list_tags_for_resource = (
get_organizations_metadata(
aws_account_id=current_audit_info.audited_account,
session=current_audit_info.audit_session,
)
)
if organizations_metadata:
current_audit_info.organizations_metadata = (
parse_organizations_metadata(
organizations_metadata, list_tags_for_resource
)
)
# else:
# logger.info(
# f"Getting organizations metadata for account with IAM Role ARN {organizations_role_arn}"
# )
# assumed_credentials = assume_role(
# aws_provider.aws_session,
# aws_provider.role_info,
# sts_endpoint_region,
# )
# organizations_metadata, list_tags_for_resource = (
# get_organizations_metadata(
# current_audit_info.audited_account, assumed_credentials
# )
# )
# current_audit_info.organizations_metadata = (
# parse_organizations_metadata(
# organizations_metadata, list_tags_for_resource
# )
# )
# logger.info(
# f"Organizations metadata retrieved with IAM Role ARN {organizations_role_arn}"
# )
# else:
# try:
# logger.info(
# "Getting organizations metadata for account if it is a delegated administrator"
# )
# organizations_metadata, list_tags_for_resource = (
# get_organizations_metadata(
# aws_account_id=current_audit_info.audited_account,
# session=current_audit_info.audit_session,
# )
# )
# if organizations_metadata:
# current_audit_info.organizations_metadata = (
# parse_organizations_metadata(
# organizations_metadata, list_tags_for_resource
# )
# )
logger.info(
"Organizations metadata retrieved as a delegated administrator"
)
except Exception as error:
# If the account is not a delegated administrator for AWS Organizations a credentials error will be thrown
# Since it is a permission issue for an optional we'll raise a warning
logger.warning(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
# logger.info(
# "Organizations metadata retrieved as a delegated administrator"
# )
# except Exception as error:
# # If the account is not a delegated administrator for AWS Organizations a credentials error will be thrown
# # Since it is a permission issue for an optional we'll raise a warning
# logger.warning(
# f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
# )
# Setting default region of session
if current_audit_info.audit_session.region_name:
current_audit_info.profile_region = (
current_audit_info.audit_session.region_name
)
else:
current_audit_info.profile_region = "us-east-1"
# # Setting default region of session
# if current_audit_info.audit_session.region_name:
# current_audit_info.profile_region = (
# current_audit_info.audit_session.region_name
# )
# else:
# current_audit_info.profile_region = "us-east-1"
# Parse Scan Tags
if arguments.get("resource_tags"):
input_resource_tags = arguments.get("resource_tags")
current_audit_info.audit_resources = get_tagged_resources(
input_resource_tags, current_audit_info
)
# # Parse Scan Tags
# if arguments.get("resource_tags"):
# input_resource_tags = arguments.get("resource_tags")
# current_audit_info.audit_resources = get_tagged_resources(
# input_resource_tags, current_audit_info
# )
# Parse Input Resource ARNs
if arguments.get("resource_arn"):
current_audit_info.audit_resources = arguments.get("resource_arn")
# # Parse Input Resource ARNs
# if arguments.get("resource_arn"):
# current_audit_info.audit_resources = arguments.get("resource_arn")
# Get Enabled Regions
current_audit_info.enabled_regions = get_aws_enabled_regions(current_audit_info)
# # Get Enabled Regions
# current_audit_info.enabled_regions = get_aws_enabled_regions(current_audit_info)
return current_audit_info
# return current_audit_info
def set_aws_execution_parameters(self, provider, audit_info) -> list[str]:
# Once the audit_info is set and we have the eventual checks from arn, it is time to exclude the others
try:
if audit_info.audit_resources:
audit_info.audited_regions = get_regions_from_audit_resources(
audit_info.audit_resources
)
return get_checks_from_input_arn(audit_info.audit_resources, provider)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
sys.exit(1)
# TODO: remove if not needed, but not now :)
# def set_aws_execution_parameters(self, provider, audit_info) -> list[str]:
# # Once the audit_info is set and we have the eventual checks from arn, it is time to exclude the others
# try:
# if audit_info.audit_resources:
# audit_info.audited_regions = get_regions_from_audit_resources(
# audit_info.audit_resources
# )
# return get_checks_from_input_arn(audit_info.audit_resources, provider)
# except Exception as error:
# logger.critical(
# f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
# )
# sys.exit(1)
def set_azure_audit_info(self, arguments) -> Azure_Audit_Info:
"""
@@ -444,23 +427,3 @@ def set_provider_audit_info(provider: str, arguments: dict):
sys.exit(1)
else:
return provider_audit_info
def set_provider_execution_parameters(provider: str, audit_info):
"""
set_provider_execution_parameters" configures automatically the audit execution based on the selected provider and returns the checks that are going to be executed.
"""
try:
set_provider_execution_parameters_function = (
f"set_{provider}_execution_parameters"
)
checks_to_execute = getattr(
Audit_Info(), set_provider_execution_parameters_function
)(provider, audit_info)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
sys.exit(1)
else:
return checks_to_execute
prowler/providers/common/common.py
+1 -1
View File
@@ -34,7 +34,7 @@ def set_global_provider_object(arguments):
global global_provider
# make here dynamic import
common_import_path = (
f"prowler.providers.{arguments.provider}.{arguments.provider}_provider_new"
f"prowler.providers.{arguments.provider}.{arguments.provider}_provider"
)
provider_class = f"{arguments.provider.capitalize()}Provider"
global_provider = getattr(import_module(common_import_path), provider_class)(
prowler/providers/common/mutelist.py
-10
View File
@@ -2,7 +2,6 @@ import importlib
import sys
from prowler.lib.logger import logger
from prowler.providers.aws.lib.mutelist.mutelist import parse_mutelist_file
def set_provider_mutelist(provider, audit_info, args):
@@ -24,12 +23,3 @@ def set_provider_mutelist(provider, audit_info, args):
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
sys.exit(1)
def set_aws_mutelist(audit_info, mutelist_file):
# Parse content from Mute List file and get it, if necessary, from S3
if mutelist_file:
mutelist_file = parse_mutelist_file(audit_info, mutelist_file)
else:
mutelist_file = None
return mutelist_file
prowler/providers/common/outputs.py
+9 -8
View File
@@ -9,7 +9,7 @@ from prowler.lib.logger import logger
def set_provider_output_options(
provider: str, arguments, audit_info, mutelist_file, bulk_checks_metadata
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.
@@ -19,7 +19,7 @@ def set_provider_output_options(
provider_output_class = f"{provider.capitalize()}_Output_Options"
provider_output_options = getattr(
importlib.import_module(__name__), provider_output_class
)(arguments, audit_info, mutelist_file, bulk_checks_metadata)
)(arguments, identity, mutelist_file, bulk_checks_metadata)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -135,15 +135,16 @@ class Kubernetes_Output_Options(Provider_Output_Options):
class Aws_Output_Options(Provider_Output_Options):
security_hub_enabled: bool
def __init__(self, arguments, audit_info, mutelist_file, bulk_checks_metadata):
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
if arguments.shodan:
audit_info = change_config_var(
"shodan_api_key", arguments.shodan, audit_info
)
# 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 (
@@ -151,7 +152,7 @@ class Aws_Output_Options(Provider_Output_Options):
or arguments.output_filename is None
):
self.output_filename = (
f"prowler-output-{audit_info.audited_account}-{output_file_timestamp}"
f"prowler-output-{identity.account}-{output_file_timestamp}"
)
else:
self.output_filename = arguments.output_filename