mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 21:11:53 +00:00
chore(aws): Simplify provider (#3481)
Co-authored-by: Sergio Garcia <sergargar1@gmail.com>
This commit is contained in:
Pepe Fagoaga
+5
-10
@@ -44,11 +44,6 @@ 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.clean import clean_provider_local_output_directories
|
||||
from prowler.providers.common.common import (
|
||||
get_global_provider,
|
||||
@@ -151,14 +146,14 @@ def prowler():
|
||||
print_checks(provider, sorted(checks_to_execute), bulk_checks_metadata)
|
||||
sys.exit()
|
||||
|
||||
# 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__)
|
||||
|
||||
# Provider to scan
|
||||
set_global_provider_object(args)
|
||||
# TODO: rename global_provider to provider
|
||||
global_provider = get_global_provider()
|
||||
|
||||
# Print Provider Credentials
|
||||
if not args.only_logs:
|
||||
global_provider.print_credentials()
|
||||
|
||||
# Import custom checks from folder
|
||||
if checks_folder:
|
||||
parse_checks_from_folder(global_provider, checks_folder)
|
||||
|
||||
@@ -190,7 +190,7 @@ def remove_custom_checks_module(input_folder: str, provider: str):
|
||||
shutil.rmtree(input_folder)
|
||||
|
||||
|
||||
def list_services(provider: str) -> set():
|
||||
def list_services(provider: str) -> set:
|
||||
available_services = set()
|
||||
checks_tuple = recover_checks_from_provider(provider)
|
||||
for _, check_path in checks_tuple:
|
||||
@@ -203,7 +203,7 @@ def list_services(provider: str) -> set():
|
||||
return sorted(available_services)
|
||||
|
||||
|
||||
def list_categories(bulk_checks_metadata: dict) -> set():
|
||||
def list_categories(bulk_checks_metadata: dict) -> set:
|
||||
available_categories = set()
|
||||
for check in bulk_checks_metadata.values():
|
||||
for cat in check.Categories:
|
||||
@@ -431,9 +431,11 @@ def execute_checks(
|
||||
services_executed = set()
|
||||
checks_executed = set()
|
||||
|
||||
# TODO: why is this here?
|
||||
global_provider = get_global_provider()
|
||||
|
||||
# Initialize the Audit Metadata
|
||||
# TODO: this should be done in the provider class
|
||||
global_provider.audit_metadata = Audit_Metadata(
|
||||
services_scanned=0,
|
||||
expected_checks=checks_to_execute,
|
||||
@@ -466,7 +468,7 @@ def execute_checks(
|
||||
check_findings = execute(
|
||||
service,
|
||||
check_name,
|
||||
global_provider.provider,
|
||||
global_provider.type,
|
||||
audit_output_options,
|
||||
global_provider.identity,
|
||||
services_executed,
|
||||
@@ -478,7 +480,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 {global_provider.provider.upper()} provider"
|
||||
f"Check '{check_name}' was not found for the {global_provider.type.upper()} provider"
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
@@ -524,7 +526,7 @@ def execute_checks(
|
||||
except ModuleNotFoundError:
|
||||
# TODO: add more loggin here, we need the original exception -- traceback.print_last()
|
||||
logger.error(
|
||||
f"Check '{check_name}' was not found for the {global_provider.provider.upper()} provider"
|
||||
f"Check '{check_name}' was not found for the {global_provider.type.upper()} provider"
|
||||
)
|
||||
except Exception as error:
|
||||
# TODO: add more loggin here, we need the original exception -- traceback.print_last()
|
||||
@@ -546,7 +548,7 @@ def execute(
|
||||
custom_checks_metadata: Any,
|
||||
):
|
||||
# Import check module
|
||||
check_module_path = f"prowler.providers.{global_provider.provider}.services.{service}.{check_name}.{check_name}"
|
||||
check_module_path = f"prowler.providers.{global_provider.type}.services.{service}.{check_name}.{check_name}"
|
||||
lib = import_check(check_module_path)
|
||||
# Recover functions from check
|
||||
check_to_execute = getattr(lib, check_name)
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -6,6 +6,7 @@ from prowler.providers.aws.lib.arn.error import RoleArnParsingFailedMissingField
|
||||
|
||||
|
||||
class ARN(BaseModel):
|
||||
arn: str
|
||||
partition: str
|
||||
service: str
|
||||
region: Optional[str] # In IAM ARN's do not have region
|
||||
@@ -21,6 +22,7 @@ class ARN(BaseModel):
|
||||
## Retrieve fields
|
||||
arn_elements = arn.split(":", 5)
|
||||
data = {
|
||||
"arn": arn,
|
||||
"partition": arn_elements[1],
|
||||
"service": arn_elements[2],
|
||||
"region": arn_elements[3] if arn_elements[3] != "" else None,
|
||||
|
||||
@@ -1,70 +0,0 @@
|
||||
import sys
|
||||
|
||||
from boto3 import session
|
||||
from colorama import Fore, Style
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.providers.aws.config import AWS_STS_GLOBAL_ENDPOINT_REGION
|
||||
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
|
||||
|
||||
|
||||
def validate_AWSCredentials(
|
||||
session: session, input_regions: list, sts_endpoint_region: str = None
|
||||
) -> dict:
|
||||
try:
|
||||
# If there is no region passed with -f/--region/--filter-region
|
||||
if input_regions is None or len(input_regions) == 0:
|
||||
# If you have a region configured in your AWS config or credentials file
|
||||
if session.region_name is not None:
|
||||
aws_region = session.region_name
|
||||
else:
|
||||
# If there is no region set passed with -f/--region
|
||||
# we use the Global STS Endpoint Region, us-east-1
|
||||
aws_region = AWS_STS_GLOBAL_ENDPOINT_REGION
|
||||
else:
|
||||
# Get the first region passed to the -f/--region
|
||||
aws_region = input_regions[0]
|
||||
|
||||
validate_credentials_client = create_sts_session(session, aws_region)
|
||||
caller_identity = validate_credentials_client.get_caller_identity()
|
||||
# Include the region where the caller_identity has validated the credentials
|
||||
caller_identity["region"] = aws_region
|
||||
except Exception as error:
|
||||
logger.critical(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
sys.exit(1)
|
||||
else:
|
||||
return caller_identity
|
||||
|
||||
|
||||
def print_AWSCredentials(audit_info: AWS_Audit_Info):
|
||||
# Beautify audited regions, set "all" if there is no filter region
|
||||
regions = (
|
||||
", ".join(audit_info.audited_regions)
|
||||
if audit_info.audited_regions is not None
|
||||
else "all"
|
||||
)
|
||||
# Beautify audited profile, set "default" if there is no profile set
|
||||
profile = audit_info.profile if audit_info.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}[{audit_info.audited_account}]{Style.RESET_ALL} UserId: {Fore.YELLOW}[{audit_info.audited_user_id}]{Style.RESET_ALL}
|
||||
Caller Identity ARN: {Fore.YELLOW}[{audit_info.audited_identity_arn}]{Style.RESET_ALL}
|
||||
"""
|
||||
# If -A is set, print Assumed Role ARN
|
||||
if audit_info.assumed_role_info.role_arn is not None:
|
||||
report += f"""Assumed Role ARN: {Fore.YELLOW}[{audit_info.assumed_role_info.role_arn}]{Style.RESET_ALL}
|
||||
"""
|
||||
print(report)
|
||||
|
||||
|
||||
def create_sts_session(
|
||||
session: session.Session, aws_region: str
|
||||
) -> session.Session.client:
|
||||
return session.client(
|
||||
"sts", aws_region, endpoint_url=f"https://sts.{aws_region}.amazonaws.com"
|
||||
)
|
||||
@@ -1,28 +1,16 @@
|
||||
from boto3 import client, session
|
||||
from boto3 import session
|
||||
|
||||
# from boto3 import client
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.providers.aws.lib.audit_info.models import AWSOrganizationsInfo
|
||||
|
||||
|
||||
def get_organizations_metadata(
|
||||
aws_account_id: str,
|
||||
assumed_credentials: dict = None,
|
||||
session: session = None,
|
||||
session: session.Session,
|
||||
) -> tuple[dict, dict]:
|
||||
try:
|
||||
if assumed_credentials:
|
||||
organizations_client = client(
|
||||
"organizations",
|
||||
aws_access_key_id=assumed_credentials["Credentials"]["AccessKeyId"],
|
||||
aws_secret_access_key=assumed_credentials["Credentials"][
|
||||
"SecretAccessKey"
|
||||
],
|
||||
aws_session_token=assumed_credentials["Credentials"]["SessionToken"],
|
||||
)
|
||||
if session:
|
||||
organizations_client = session.client("organizations")
|
||||
else:
|
||||
organizations_client = client("organizations")
|
||||
organizations_client = session.client("organizations")
|
||||
|
||||
organizations_metadata = organizations_client.describe_account(
|
||||
AccountId=aws_account_id
|
||||
|
||||
@@ -23,6 +23,7 @@ class AWSService:
|
||||
|
||||
def __init__(self, service: str, provider: AwsProvider, global_service=False):
|
||||
# Audit Information
|
||||
# Do we need to store the whole provider?
|
||||
self.provider = provider
|
||||
self.audited_account = provider.identity.account
|
||||
self.audited_account_arn = provider.identity.account_arn
|
||||
@@ -32,7 +33,7 @@ class AWSService:
|
||||
self.audit_config = provider.audit_config
|
||||
|
||||
# AWS Session
|
||||
self.session = provider.session.session
|
||||
self.session = provider.session.current_session
|
||||
|
||||
# We receive the service using __class__.__name__ or the service name in lowercase
|
||||
# e.g.: AccessAnalyzer --> we need a lowercase string, so service.lower()
|
||||
@@ -40,9 +41,7 @@ class AWSService:
|
||||
|
||||
# Generate Regional Clients
|
||||
if not global_service:
|
||||
self.regional_clients = provider.generate_regional_clients(
|
||||
self.service, global_service
|
||||
)
|
||||
self.regional_clients = provider.generate_regional_clients(self.service)
|
||||
# TODO: review the following code
|
||||
# self.regional_clients = generate_regional_clients(self.service, audit_info)
|
||||
|
||||
|
||||
@@ -4,6 +4,8 @@ from datetime import datetime
|
||||
from boto3.session import Session
|
||||
from botocore.config import Config
|
||||
|
||||
from prowler.providers.aws.lib.arn.models import ARN
|
||||
|
||||
|
||||
@dataclass
|
||||
class AWSOrganizationsInfo:
|
||||
@@ -23,17 +25,18 @@ class AWSCredentials:
|
||||
|
||||
|
||||
@dataclass
|
||||
class AWSAssumeRole:
|
||||
role_arn: str
|
||||
class AWSAssumeRoleInfo:
|
||||
role_arn: ARN
|
||||
session_duration: int
|
||||
external_id: str
|
||||
mfa_enabled: bool
|
||||
role_session_name: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class AWSAssumeRoleConfiguration:
|
||||
assumed_role_info: AWSAssumeRole
|
||||
assumed_role_credentials: AWSCredentials
|
||||
info: AWSAssumeRoleInfo
|
||||
credentials: AWSCredentials
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -45,11 +48,30 @@ class AWSIdentityInfo:
|
||||
identity_arn: str
|
||||
profile: str
|
||||
profile_region: str
|
||||
audited_regions: list
|
||||
audited_regions: set
|
||||
|
||||
|
||||
@dataclass
|
||||
class AWSSession:
|
||||
session: Session
|
||||
"""
|
||||
AWSSession stores the AWS session's configuration. We store the original_session in the case we need to setup a new one with different credentials and the restore to the original one.
|
||||
|
||||
"""
|
||||
|
||||
current_session: Session
|
||||
original_session: Session
|
||||
session_config: Config
|
||||
original_session: None
|
||||
|
||||
|
||||
@dataclass
|
||||
class AWSCallerIdentity:
|
||||
user_id: str
|
||||
account: str
|
||||
arn: str
|
||||
region: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class AWSMFAInfo:
|
||||
arn: str
|
||||
totp: str
|
||||
|
||||
@@ -52,14 +52,13 @@ class AzureProvider(Provider):
|
||||
managed_entity_auth,
|
||||
subscription_ids,
|
||||
)
|
||||
if not arguments.only_logs:
|
||||
self.print_credentials()
|
||||
|
||||
# TODO: should we keep this here or within the identity?
|
||||
self.locations = self.get_locations(self.session, self.region_config)
|
||||
# TODO: move this to the providers, pending for AWS, GCP, AZURE and K8s
|
||||
self.audit_config = {}
|
||||
|
||||
# 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
|
||||
):
|
||||
|
||||
@@ -1,423 +0,0 @@
|
||||
import sys
|
||||
|
||||
from colorama import Fore, Style
|
||||
|
||||
from prowler.config.config import load_and_validate_config_file
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.providers.gcp.gcp_provider import GCP_Provider
|
||||
from prowler.providers.gcp.lib.audit_info.audit_info import gcp_audit_info
|
||||
from prowler.providers.gcp.lib.audit_info.models import GCP_Audit_Info
|
||||
from prowler.providers.kubernetes.kubernetes_provider import Kubernetes_Provider
|
||||
from prowler.providers.kubernetes.lib.audit_info.audit_info import kubernetes_audit_info
|
||||
from prowler.providers.kubernetes.lib.audit_info.models import Kubernetes_Audit_Info
|
||||
|
||||
|
||||
class Audit_Info:
|
||||
def __init__(self):
|
||||
logger.info("Setting Audit Info ...")
|
||||
|
||||
def print_gcp_credentials(self, audit_info: GCP_Audit_Info):
|
||||
# Beautify audited profile, set "default" if there is no profile set
|
||||
profile = getattr(audit_info.credentials, "_service_account_email", "default")
|
||||
|
||||
report = f"""
|
||||
This report is being generated using credentials below:
|
||||
|
||||
GCP Account: {Fore.YELLOW}[{profile}]{Style.RESET_ALL} GCP Project IDs: {Fore.YELLOW}[{", ".join(audit_info.project_ids)}]{Style.RESET_ALL}
|
||||
"""
|
||||
print(report)
|
||||
|
||||
def print_kubernetes_credentials(self, audit_info: Kubernetes_Audit_Info):
|
||||
# Get the current context
|
||||
cluster_name = self.context.get("context").get("cluster")
|
||||
user_name = self.context.get("context").get("user")
|
||||
roles = self.get_context_user_roles()
|
||||
roles_str = ", ".join(roles) if roles else "No associated Roles"
|
||||
|
||||
report = f"""
|
||||
This report is being generated using the Kubernetes configuration below:
|
||||
|
||||
Kubernetes Cluster: {Fore.YELLOW}[{cluster_name}]{Style.RESET_ALL} User: {Fore.YELLOW}[{user_name}]{Style.RESET_ALL} Namespaces: {Fore.YELLOW}[{', '.join(self.namespaces)}]{Style.RESET_ALL} Roles: {Fore.YELLOW}[{roles_str}]{Style.RESET_ALL}
|
||||
"""
|
||||
print(report)
|
||||
|
||||
# def print_azure_credentials(self, audit_info: Azure_Audit_Info):
|
||||
# printed_subscriptions = []
|
||||
# for key, value in audit_info.identity.subscriptions.items():
|
||||
# intermediate = f"{key} : {value}"
|
||||
# printed_subscriptions.append(intermediate)
|
||||
# report = f"""
|
||||
# This report is being generated using the identity below:
|
||||
|
||||
# Azure Tenant IDs: {Fore.YELLOW}[{" ".join(audit_info.identity.tenant_ids)}]{Style.RESET_ALL} Azure Tenant Domain: {Fore.YELLOW}[{audit_info.identity.domain}]{Style.RESET_ALL} Azure Region: {Fore.YELLOW}[{audit_info.azure_region_config.name}]{Style.RESET_ALL}
|
||||
# Azure Subscriptions: {Fore.YELLOW}{printed_subscriptions}{Style.RESET_ALL}
|
||||
# Azure Identity Type: {Fore.YELLOW}[{audit_info.identity.identity_type}]{Style.RESET_ALL} Azure Identity ID: {Fore.YELLOW}[{audit_info.identity.identity_id}]{Style.RESET_ALL}
|
||||
# """
|
||||
# print(report)
|
||||
|
||||
# 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")
|
||||
|
||||
# # 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
|
||||
|
||||
# input_profile = arguments.get("profile")
|
||||
# input_regions = arguments.get("region")
|
||||
# organizations_role_arn = arguments.get("organizations_role")
|
||||
|
||||
# # 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 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
|
||||
|
||||
# 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']}")
|
||||
|
||||
# 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
|
||||
# )
|
||||
|
||||
# # 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)
|
||||
|
||||
# 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
|
||||
|
||||
# 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
|
||||
# )
|
||||
|
||||
# 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
|
||||
# )
|
||||
# )
|
||||
|
||||
# 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"
|
||||
|
||||
# # 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")
|
||||
|
||||
# # Get Enabled Regions
|
||||
# current_audit_info.enabled_regions = get_aws_enabled_regions(current_audit_info)
|
||||
|
||||
# return current_audit_info
|
||||
|
||||
# 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)
|
||||
# TODO: remove if not needed, but not now :)
|
||||
# def set_azure_audit_info(self, arguments) -> Azure_Audit_Info:
|
||||
# """
|
||||
# set_azure_audit_info returns the Azure_Audit_Info
|
||||
# """
|
||||
# logger.info("Setting Azure session ...")
|
||||
# subscription_ids = arguments.get("subscription_ids")
|
||||
|
||||
# logger.info("Checking if any credentials mode is set ...")
|
||||
# az_cli_auth = arguments.get("az_cli_auth")
|
||||
# sp_env_auth = arguments.get("sp_env_auth")
|
||||
# browser_auth = arguments.get("browser_auth")
|
||||
# managed_entity_auth = arguments.get("managed_entity_auth")
|
||||
# tenant_id = arguments.get("tenant_id")
|
||||
|
||||
# logger.info("Checking if region is different than default one")
|
||||
# region = arguments.get("azure_region")
|
||||
|
||||
# if (
|
||||
# not az_cli_auth
|
||||
# and not sp_env_auth
|
||||
# and not browser_auth
|
||||
# and not managed_entity_auth
|
||||
# ):
|
||||
# raise AzureException(
|
||||
# "Azure provider requires at least one authentication method set: [--az-cli-auth | --sp-env-auth | --browser-auth | --managed-identity-auth]"
|
||||
# )
|
||||
# if (not browser_auth and tenant_id) or (browser_auth and not tenant_id):
|
||||
# raise AzureException(
|
||||
# "Azure Tenant ID (--tenant-id) is required only for browser authentication mode"
|
||||
# )
|
||||
|
||||
# azure_provider = Azure_Provider(
|
||||
# az_cli_auth,
|
||||
# sp_env_auth,
|
||||
# browser_auth,
|
||||
# managed_entity_auth,
|
||||
# subscription_ids,
|
||||
# tenant_id,
|
||||
# region,
|
||||
# )
|
||||
# azure_audit_info.credentials = azure_provider.get_credentials()
|
||||
# azure_audit_info.identity = azure_provider.get_identity()
|
||||
# region_config = azure_provider.get_region_config()
|
||||
# azure_audit_info.azure_region_config = AzureRegionConfig(
|
||||
# name=region,
|
||||
# authority=region_config["authority"],
|
||||
# base_url=region_config["base_url"],
|
||||
# credential_scopes=region_config["credential_scopes"],
|
||||
# )
|
||||
# azure_audit_info.locations = azure_provider.get_locations(
|
||||
# azure_audit_info.credentials, region_config
|
||||
# )
|
||||
|
||||
# if not arguments.get("only_logs"):
|
||||
# self.print_azure_credentials(get_global_provider())
|
||||
|
||||
# return azure_audit_info
|
||||
|
||||
def set_gcp_audit_info(self, arguments) -> GCP_Audit_Info:
|
||||
"""
|
||||
set_gcp_audit_info returns the GCP_Audit_Info
|
||||
"""
|
||||
logger.info("Setting GCP session ...")
|
||||
project_ids = arguments.get("project_ids")
|
||||
|
||||
logger.info("Checking if any credentials mode is set ...")
|
||||
credentials_file = arguments.get("credentials_file")
|
||||
|
||||
gcp_provider = GCP_Provider(
|
||||
credentials_file,
|
||||
project_ids,
|
||||
)
|
||||
|
||||
(
|
||||
gcp_audit_info.credentials,
|
||||
gcp_audit_info.default_project_id,
|
||||
gcp_audit_info.project_ids,
|
||||
) = gcp_provider.get_credentials()
|
||||
|
||||
return gcp_audit_info
|
||||
|
||||
def set_kubernetes_audit_info(self, arguments) -> Kubernetes_Audit_Info:
|
||||
"""
|
||||
set_kubernetes_audit_info returns the Kubernetes_Audit_Info
|
||||
"""
|
||||
logger.info("Setting Kubernetes session ...")
|
||||
kubeconfig_file = arguments.get("kubeconfig_file")
|
||||
|
||||
logger.info("Checking if any context is set ...")
|
||||
context = arguments.get("context")
|
||||
|
||||
logger.info("Checking if any namespace is set ...")
|
||||
namespaces = arguments.get("namespaces")
|
||||
|
||||
kubernetes_provider = Kubernetes_Provider(kubeconfig_file, context, namespaces)
|
||||
|
||||
(
|
||||
kubernetes_audit_info.api_client,
|
||||
kubernetes_audit_info.context,
|
||||
) = kubernetes_provider.get_credentials()
|
||||
|
||||
return kubernetes_audit_info
|
||||
|
||||
|
||||
def set_provider_audit_info(provider: str, arguments: dict):
|
||||
"""
|
||||
set_provider_audit_info configures automatically the audit session based on the selected provider and returns the audit_info object.
|
||||
"""
|
||||
try:
|
||||
provider_set_audit_info = f"set_{provider}_audit_info"
|
||||
provider_audit_info = getattr(Audit_Info(), provider_set_audit_info)(arguments)
|
||||
|
||||
# Set the audit configuration from the config file
|
||||
# TODO: move this to the providers
|
||||
provider_audit_info.audit_config = load_and_validate_config_file(
|
||||
provider, arguments["config_file"]
|
||||
)
|
||||
except Exception as error:
|
||||
logger.critical(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
sys.exit(1)
|
||||
else:
|
||||
return provider_audit_info
|
||||
@@ -1,6 +1,9 @@
|
||||
import sys
|
||||
from importlib import import_module
|
||||
from typing import Any
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
|
||||
providers_prowler_lib_path = "prowler.providers"
|
||||
|
||||
global_provider = None
|
||||
@@ -31,12 +34,18 @@ def get_available_providers() -> list[str]:
|
||||
|
||||
|
||||
def set_global_provider_object(arguments):
|
||||
global global_provider
|
||||
# make here dynamic import
|
||||
common_import_path = (
|
||||
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)(
|
||||
arguments
|
||||
)
|
||||
try:
|
||||
global global_provider
|
||||
# make here dynamic import
|
||||
common_import_path = (
|
||||
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)(
|
||||
arguments
|
||||
)
|
||||
except TypeError as error:
|
||||
logger.critical(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
# TODO: include this for all the providers
|
||||
class Audit_Metadata(BaseModel):
|
||||
services_scanned: int
|
||||
# We can't use a set in the expected
|
||||
|
||||
@@ -1,15 +1,68 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
# TODO: with this we can enforce that all classes ending with "Provider" needs to inherint from the Provider class
|
||||
# class ProviderMeta:
|
||||
# def __init__(cls, name, bases, dct):
|
||||
# # Check if the class name ends with 'Provider'
|
||||
# if name.endswith("Provider"):
|
||||
# # Check if any base class is a subclass of Provider (or is Provider itself)
|
||||
# if not any(issubclass(b, Provider) for b in bases if b is not object):
|
||||
# raise TypeError(f"{name} must inherit from Provider")
|
||||
# super().__init__(name, bases, dct)
|
||||
# class Provider(metaclass=ProviderMeta):
|
||||
|
||||
|
||||
class Provider(ABC):
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def type(self):
|
||||
"""
|
||||
type method stores the provider's type.
|
||||
|
||||
This method needs to be created in each provider.
|
||||
"""
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def identity(self):
|
||||
"""
|
||||
identity method stores the provider's identity to audit.
|
||||
|
||||
This method needs to be created in each provider.
|
||||
"""
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def session(self):
|
||||
"""
|
||||
session method stores the provider's session to audit.
|
||||
|
||||
This method needs to be created in each provider.
|
||||
"""
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def audit_config(self):
|
||||
"""
|
||||
audit_config method stores the provider's audit configuration.
|
||||
|
||||
This method needs to be created in each provider.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def print_credentials(self):
|
||||
"""
|
||||
print_credentials is used to display in the CLI the provider's credentials used to audit.
|
||||
|
||||
This method needs to be created in each provider.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def setup_session(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def print_credentials(self):
|
||||
pass
|
||||
|
||||
# TODO: probably this won't be here since we want to do the arguments validation during the parse()
|
||||
def validate_arguments(self):
|
||||
pass
|
||||
|
||||
|
||||
@@ -60,10 +60,6 @@ class GcpProvider(Provider):
|
||||
profile=getattr(self.session, "_service_account_email", "default")
|
||||
)
|
||||
|
||||
# TODO: move this to the parent class or the main
|
||||
if not arguments.only_logs:
|
||||
self.print_credentials()
|
||||
|
||||
def setup_session(self, credentials_file):
|
||||
try:
|
||||
if credentials_file:
|
||||
|
||||
@@ -18,6 +18,7 @@ class KubernetesIdentityInfo:
|
||||
|
||||
class KubernetesProvider(Provider):
|
||||
provider = "kubernetes"
|
||||
# TODO: api_client is the session
|
||||
api_client: Any
|
||||
context: dict
|
||||
namespaces: list
|
||||
@@ -49,8 +50,6 @@ class KubernetesProvider(Provider):
|
||||
self.identity = KubernetesIdentityInfo(
|
||||
active_context=self.context["name"].replace(":", "_").replace("/", "_")
|
||||
)
|
||||
if not arguments.only_logs:
|
||||
self.print_credentials()
|
||||
|
||||
def setup_session(self, kubeconfig_file, input_context):
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user