mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 13:01:56 +00:00
feat(aws): add custom exceptions class (#4847)
This commit is contained in:
@@ -0,0 +1,55 @@
|
||||
class ProwlerException(Exception):
|
||||
"""Base exception for all Prowler SDK errors."""
|
||||
|
||||
ERROR_CODES = {
|
||||
(1901, "UnexpectedError"): {
|
||||
"message": "Unexpected error occurred.",
|
||||
"remediation": "Please review the error message and try again.",
|
||||
"file": "{file}",
|
||||
"provider": "{provider}",
|
||||
}
|
||||
}
|
||||
|
||||
def __init__(
|
||||
self, code, provider=None, file=None, original_exception=None, error_info=None
|
||||
):
|
||||
"""
|
||||
Initialize the ProwlerException class.
|
||||
|
||||
Args:
|
||||
code (int): The error code.
|
||||
provider (str): The provider name.
|
||||
file (str): The file name.
|
||||
original_exception (Exception): The original exception.
|
||||
error_info (dict): The error information.
|
||||
|
||||
Example:
|
||||
A ProwlerException is raised with the following parameters and format:
|
||||
>>> original_exception = Exception("Error occurred.")
|
||||
ProwlerException(1901, "AWS", "file.txt", original_exception)
|
||||
>>> [1901] Unexpected error occurred. - Exception: Error occurred.
|
||||
"""
|
||||
self.code = code
|
||||
self.provider = provider
|
||||
self.file = file
|
||||
if error_info is None:
|
||||
error_info = self.ERROR_CODES.get((code, self.__class__.__name__))
|
||||
self.message = error_info.get("message")
|
||||
self.remediation = error_info.get("remediation")
|
||||
self.original_exception = original_exception
|
||||
# Format -> [code] message - original_exception
|
||||
if original_exception is None:
|
||||
super().__init__(f"[{self.code}] {self.message}")
|
||||
else:
|
||||
super().__init__(
|
||||
f"[{self.code}] {self.message} - {self.original_exception}"
|
||||
)
|
||||
|
||||
def __str__(self):
|
||||
"""Overriding the __str__ method"""
|
||||
return f"{self.__class__.__name__}[{self.code}]: {self.message} - {self.original_exception}"
|
||||
|
||||
|
||||
class UnexpectedError(ProwlerException):
|
||||
def __init__(self, provider, file, original_exception=None):
|
||||
super().__init__(1901, provider, file, original_exception)
|
||||
@@ -1,6 +1,5 @@
|
||||
import os
|
||||
import pathlib
|
||||
from argparse import ArgumentTypeError
|
||||
from datetime import datetime
|
||||
from re import fullmatch
|
||||
|
||||
@@ -29,6 +28,20 @@ from prowler.providers.aws.config import (
|
||||
BOTO3_USER_AGENT_EXTRA,
|
||||
ROLE_SESSION_NAME,
|
||||
)
|
||||
from prowler.providers.aws.exceptions.exceptions import (
|
||||
AWSArgumentTypeValidationError,
|
||||
AWSAssumeRoleError,
|
||||
AWSClientError,
|
||||
AWSIAMRoleARNEmptyResource,
|
||||
AWSIAMRoleARNInvalidAccountID,
|
||||
AWSIAMRoleARNInvalidResourceType,
|
||||
AWSIAMRoleARNPartitionEmpty,
|
||||
AWSIAMRoleARNRegionNotEmtpy,
|
||||
AWSIAMRoleARNServiceNotIAMnorSTS,
|
||||
AWSNoCredentialsError,
|
||||
AWSProfileNotFoundError,
|
||||
AWSSetUpSessionError,
|
||||
)
|
||||
from prowler.providers.aws.lib.arn.arn import parse_iam_credentials_arn
|
||||
from prowler.providers.aws.lib.arn.models import ARN
|
||||
from prowler.providers.aws.lib.mutelist.mutelist import AWSMutelist
|
||||
@@ -462,9 +475,12 @@ class AwsProvider(Provider):
|
||||
)
|
||||
except Exception as error:
|
||||
logger.critical(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
f"AWSSetUpSessionError[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
raise AWSSetUpSessionError(
|
||||
original_exception=error,
|
||||
file=pathlib.Path(__file__).name,
|
||||
)
|
||||
raise error
|
||||
|
||||
def setup_assumed_session(
|
||||
self,
|
||||
@@ -874,7 +890,10 @@ class AwsProvider(Provider):
|
||||
logger.critical(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
|
||||
)
|
||||
raise error
|
||||
raise AWSAssumeRoleError(
|
||||
original_exception=error,
|
||||
file=pathlib.Path(__file__).name,
|
||||
)
|
||||
|
||||
def get_aws_enabled_regions(self, current_session: Session) -> set:
|
||||
"""get_aws_enabled_regions returns a set of enabled AWS regions"""
|
||||
@@ -1030,24 +1049,92 @@ class AwsProvider(Provider):
|
||||
is_connected=True,
|
||||
)
|
||||
|
||||
except (ClientError, ProfileNotFound, NoCredentialsError) as credentials_error:
|
||||
logger.error(
|
||||
f"{credentials_error.__class__.__name__}[{credentials_error.__traceback__.tb_lineno}]: {credentials_error}"
|
||||
)
|
||||
except AWSSetUpSessionError as setup_session_error:
|
||||
logger.error(str(setup_session_error))
|
||||
if raise_on_exception:
|
||||
raise credentials_error
|
||||
raise setup_session_error
|
||||
return Connection(error=setup_session_error)
|
||||
|
||||
return Connection(error=credentials_error)
|
||||
|
||||
except ArgumentTypeError as validation_error:
|
||||
logger.error(
|
||||
f"{validation_error.__class__.__name__}[{validation_error.__traceback__.tb_lineno}]: {validation_error}"
|
||||
)
|
||||
except AWSArgumentTypeValidationError as validation_error:
|
||||
logger.error(str(validation_error))
|
||||
if raise_on_exception:
|
||||
raise validation_error
|
||||
|
||||
return Connection(error=validation_error)
|
||||
|
||||
except AWSIAMRoleARNRegionNotEmtpy as arn_region_not_empty_error:
|
||||
logger.error(str(arn_region_not_empty_error))
|
||||
if raise_on_exception:
|
||||
raise arn_region_not_empty_error
|
||||
return Connection(error=arn_region_not_empty_error)
|
||||
|
||||
except AWSIAMRoleARNPartitionEmpty as arn_partition_empty_error:
|
||||
logger.error(str(arn_partition_empty_error))
|
||||
if raise_on_exception:
|
||||
raise arn_partition_empty_error
|
||||
return Connection(error=arn_partition_empty_error)
|
||||
|
||||
except AWSIAMRoleARNServiceNotIAMnorSTS as arn_service_not_iam_sts_error:
|
||||
logger.error(str(arn_service_not_iam_sts_error))
|
||||
if raise_on_exception:
|
||||
raise arn_service_not_iam_sts_error
|
||||
return Connection(error=arn_service_not_iam_sts_error)
|
||||
|
||||
except AWSIAMRoleARNInvalidAccountID as arn_invalid_account_id_error:
|
||||
logger.error(str(arn_invalid_account_id_error))
|
||||
if raise_on_exception:
|
||||
raise arn_invalid_account_id_error
|
||||
return Connection(error=arn_invalid_account_id_error)
|
||||
|
||||
except AWSIAMRoleARNInvalidResourceType as arn_invalid_resource_type_error:
|
||||
logger.error(str(arn_invalid_resource_type_error))
|
||||
if raise_on_exception:
|
||||
raise arn_invalid_resource_type_error
|
||||
return Connection(error=arn_invalid_resource_type_error)
|
||||
|
||||
except AWSIAMRoleARNEmptyResource as arn_empty_resource_error:
|
||||
logger.error(str(arn_empty_resource_error))
|
||||
if raise_on_exception:
|
||||
raise arn_empty_resource_error
|
||||
return Connection(error=arn_empty_resource_error)
|
||||
|
||||
except AWSAssumeRoleError as assume_role_error:
|
||||
logger.error(str(assume_role_error))
|
||||
if raise_on_exception:
|
||||
raise assume_role_error
|
||||
return Connection(error=assume_role_error)
|
||||
|
||||
except ClientError as client_error:
|
||||
logger.error(
|
||||
f"AWSClientError[{client_error.__traceback__.tb_lineno}]: {client_error}"
|
||||
)
|
||||
if raise_on_exception:
|
||||
raise AWSClientError(
|
||||
file=os.path.basename(__file__), original_exception=client_error
|
||||
) from client_error
|
||||
return Connection(error=client_error)
|
||||
|
||||
except ProfileNotFound as profile_not_found_error:
|
||||
logger.error(
|
||||
f"AWSProfileNotFoundError[{profile_not_found_error.__traceback__.tb_lineno}]: {profile_not_found_error}"
|
||||
)
|
||||
if raise_on_exception:
|
||||
raise AWSProfileNotFoundError(
|
||||
file=os.path.basename(__file__),
|
||||
original_exception=profile_not_found_error,
|
||||
) from profile_not_found_error
|
||||
return Connection(error=profile_not_found_error)
|
||||
|
||||
except NoCredentialsError as no_credentials_error:
|
||||
logger.error(
|
||||
f"AWSNoCredentialsError[{no_credentials_error.__traceback__.tb_lineno}]: {no_credentials_error}"
|
||||
)
|
||||
if raise_on_exception:
|
||||
raise AWSNoCredentialsError(
|
||||
file=os.path.basename(__file__),
|
||||
original_exception=no_credentials_error,
|
||||
) from no_credentials_error
|
||||
return Connection(error=no_credentials_error)
|
||||
|
||||
except Exception as error:
|
||||
logger.critical(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
@@ -1157,8 +1244,12 @@ def validate_session_duration(duration: int) -> int:
|
||||
duration = int(duration)
|
||||
# Since the range(i,j) goes from i to j-1 we have to j+1
|
||||
if duration not in range(900, 43201):
|
||||
raise ArgumentTypeError("Session duration must be between 900 and 43200")
|
||||
return duration
|
||||
raise AWSArgumentTypeValidationError(
|
||||
original_exception="Session Duration must be between 900 and 43200 seconds.",
|
||||
file=os.path.basename(__file__),
|
||||
)
|
||||
else:
|
||||
return duration
|
||||
|
||||
|
||||
# TODO: this duplicates the provider arguments validation library
|
||||
@@ -1181,6 +1272,7 @@ def validate_role_session_name(session_name) -> str:
|
||||
if fullmatch(r"[\w+=,.@-]{2,64}", session_name):
|
||||
return session_name
|
||||
else:
|
||||
raise ArgumentTypeError(
|
||||
"Role Session Name must be 2-64 characters long and consist only of upper- and lower-case alphanumeric characters with no spaces. You can also include underscores or any of the following characters: =,.@-"
|
||||
raise AWSArgumentTypeValidationError(
|
||||
file=os.path.basename(__file__),
|
||||
original_exception="Role Session Name must be between 2 and 64 characters and may contain alphanumeric characters, periods, hyphens, and underscores.",
|
||||
)
|
||||
|
||||
@@ -0,0 +1,195 @@
|
||||
from prowler.exceptions.exceptions import ProwlerException
|
||||
|
||||
|
||||
class AWSBaseException(ProwlerException):
|
||||
"""Base class for AWS errors."""
|
||||
|
||||
AWS_ERROR_CODES = {
|
||||
(1902, "AWSClientError"): {
|
||||
"message": "AWS ClientError occurred",
|
||||
"remediation": "Check your AWS client configuration and permissions.",
|
||||
"file": "{file}",
|
||||
"provider": "AWS",
|
||||
},
|
||||
(1903, "AWSProfileNotFoundError"): {
|
||||
"message": "AWS Profile not found",
|
||||
"remediation": "Ensure the AWS profile is correctly configured, please visit https://docs.aws.amazon.com/cli/v1/userguide/cli-configure-files.html",
|
||||
"file": "{file}",
|
||||
"provider": "AWS",
|
||||
},
|
||||
(1904, "AWSNoCredentialsError"): {
|
||||
"message": "No AWS credentials found",
|
||||
"remediation": "Verify that AWS credentials are properly set up, please visit https://docs.prowler.com/projects/prowler-open-source/en/latest/tutorials/aws/authentication/ and https://docs.aws.amazon.com/cli/v1/userguide/cli-chap-configure.html",
|
||||
"file": "{file}",
|
||||
"provider": "AWS",
|
||||
},
|
||||
(1905, "AWSArgumentTypeValidationError"): {
|
||||
"message": "AWS argument type validation error",
|
||||
"remediation": "Check the provided argument types specific to AWS and ensure they meet the required format. For session duration check: https://docs.aws.amazon.com/singlesignon/latest/userguide/howtosessionduration.html and for role session name check: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_terms-and-concepts.html#iam-term-role-session-name",
|
||||
"file": "{file}",
|
||||
"provider": "AWS",
|
||||
},
|
||||
(1906, "AWSSetUpSessionError"): {
|
||||
"message": "AWS session setup error",
|
||||
"remediation": "Check the AWS session setup and ensure it is properly configured, please visit https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-files.html and check if the provided profile has the necessary permissions.",
|
||||
"file": "{file}",
|
||||
"provider": "AWS",
|
||||
},
|
||||
(1907, "AWSIAMRoleARNRegionNotEmtpy"): {
|
||||
"message": "AWS IAM Role ARN region is not empty",
|
||||
"remediation": "Check the AWS IAM Role ARN region and ensure it is empty, visit https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_identifiers.html#identifiers-arns for more information.",
|
||||
"file": "{file}",
|
||||
"provider": "AWS",
|
||||
},
|
||||
(1908, "AWSIAMRoleARNPartitionEmpty"): {
|
||||
"message": "AWS IAM Role ARN partition is empty",
|
||||
"remediation": "Check the AWS IAM Role ARN partition and ensure it is not empty, visit https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_identifiers.html#identifiers-arns for more information.",
|
||||
"file": "{file}",
|
||||
"provider": "AWS",
|
||||
},
|
||||
(1909, "AWSIAMRoleARNMissingFields"): {
|
||||
"message": "AWS IAM Role ARN missing fields",
|
||||
"remediation": "Check the AWS IAM Role ARN and ensure all required fields are present, visit https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_identifiers.html#identifiers-arns for more information.",
|
||||
"file": "{file}",
|
||||
"provider": "AWS",
|
||||
},
|
||||
(1910, "AWSIAMRoleARNServiceNotIAMnorSTS"): {
|
||||
"message": "AWS IAM Role ARN service is not IAM nor STS",
|
||||
"remediation": "Check the AWS IAM Role ARN service and ensure it is either IAM or STS, visit https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_identifiers.html#identifiers-arns for more information.",
|
||||
"file": "{file}",
|
||||
"provider": "AWS",
|
||||
},
|
||||
(1911, "AWSIAMRoleARNInvalidAccountID"): {
|
||||
"message": "AWS IAM Role ARN account ID is invalid",
|
||||
"remediation": "Check the AWS IAM Role ARN account ID and ensure it is a valid 12-digit number, visit https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_identifiers.html#identifiers-arns for more information.",
|
||||
"file": "{file}",
|
||||
"provider": "AWS",
|
||||
},
|
||||
(1912, "AWSIAMRoleARNInvalidResourceType"): {
|
||||
"message": "AWS IAM Role ARN resource type is invalid",
|
||||
"remediation": "Check the AWS IAM Role ARN resource type and ensure it is valid, resources types are: role, user, assumed-role, root, federated-user, visit https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_identifiers.html#identifiers-arns for more information.",
|
||||
"file": "{file}",
|
||||
"provider": "AWS",
|
||||
},
|
||||
(1913, "AWSIAMRoleARNEmptyResource"): {
|
||||
"message": "AWS IAM Role ARN resource is empty",
|
||||
"remediation": "Check the AWS IAM Role ARN resource and ensure it is not empty, visit https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_identifiers.html#identifiers-arns for more information.",
|
||||
"file": "{file}",
|
||||
"provider": "AWS",
|
||||
},
|
||||
(1914, "AWSAssumeRoleError"): {
|
||||
"message": "AWS assume role error",
|
||||
"remediation": "Check the AWS assume role configuration and ensure it is properly set up, please visithttps://docs.prowler.com/projects/prowler-open-source/en/latest/tutorials/aws/role-assumption/ and https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_terms-and-concepts.html#iam-term-role-session-name",
|
||||
"file": "{file}",
|
||||
"provider": "AWS",
|
||||
},
|
||||
}
|
||||
|
||||
def __init__(self, code, provider="AWS", file=None, original_exception=None):
|
||||
error_info = self.AWS_ERROR_CODES.get((code, self.__class__.__name__))
|
||||
super().__init__(code, provider, file, original_exception, error_info)
|
||||
|
||||
|
||||
class AWSCredentialsError(AWSBaseException):
|
||||
"""Base class for AWS credentials errors."""
|
||||
|
||||
def __init__(self, code, provider="AWS", file=None, original_exception=None):
|
||||
super().__init__(code, provider, file, original_exception)
|
||||
|
||||
|
||||
class AWSClientError(AWSCredentialsError):
|
||||
def __init__(self, file=None, original_exception=None):
|
||||
super().__init__(
|
||||
1902, provider="AWS", file=file, original_exception=original_exception
|
||||
)
|
||||
|
||||
|
||||
class AWSProfileNotFoundError(AWSCredentialsError):
|
||||
def __init__(self, file=None, original_exception=None):
|
||||
super().__init__(
|
||||
1903, provider="AWS", file=file, original_exception=original_exception
|
||||
)
|
||||
|
||||
|
||||
class AWSNoCredentialsError(AWSCredentialsError):
|
||||
def __init__(self, file=None, original_exception=None):
|
||||
super().__init__(
|
||||
1904, provider="AWS", file=file, original_exception=original_exception
|
||||
)
|
||||
|
||||
|
||||
class AWSArgumentTypeValidationError(AWSBaseException):
|
||||
def __init__(self, file=None, original_exception=None):
|
||||
super().__init__(
|
||||
1905, provider="AWS", file=file, original_exception=original_exception
|
||||
)
|
||||
|
||||
|
||||
class AWSSetUpSessionError(AWSBaseException):
|
||||
def __init__(self, file=None, original_exception=None):
|
||||
super().__init__(
|
||||
1906, provider="AWS", file=file, original_exception=original_exception
|
||||
)
|
||||
|
||||
|
||||
class AWSRoleArnError(AWSBaseException):
|
||||
"""Base class for AWS role ARN errors."""
|
||||
|
||||
def __init__(self, code, provider="AWS", file=None, original_exception=None):
|
||||
super().__init__(code, provider, file, original_exception)
|
||||
|
||||
|
||||
class AWSIAMRoleARNRegionNotEmtpy(AWSRoleArnError):
|
||||
def __init__(self, file=None, original_exception=None):
|
||||
super().__init__(
|
||||
1907, provider="AWS", file=file, original_exception=original_exception
|
||||
)
|
||||
|
||||
|
||||
class AWSIAMRoleARNPartitionEmpty(AWSRoleArnError):
|
||||
def __init__(self, file=None, original_exception=None):
|
||||
super().__init__(
|
||||
1908, provider="AWS", file=file, original_exception=original_exception
|
||||
)
|
||||
|
||||
|
||||
class AWSIAMRoleARNMissingFields(AWSRoleArnError):
|
||||
def __init__(self, file=None, original_exception=None):
|
||||
super().__init__(
|
||||
1909, provider="AWS", file=file, original_exception=original_exception
|
||||
)
|
||||
|
||||
|
||||
class AWSIAMRoleARNServiceNotIAMnorSTS(AWSRoleArnError):
|
||||
def __init__(self, file=None, original_exception=None):
|
||||
super().__init__(
|
||||
1910, provider="AWS", file=file, original_exception=original_exception
|
||||
)
|
||||
|
||||
|
||||
class AWSIAMRoleARNInvalidAccountID(AWSRoleArnError):
|
||||
def __init__(self, file=None, original_exception=None):
|
||||
super().__init__(
|
||||
1911, provider="AWS", file=file, original_exception=original_exception
|
||||
)
|
||||
|
||||
|
||||
class AWSIAMRoleARNInvalidResourceType(AWSRoleArnError):
|
||||
def __init__(self, file=None, original_exception=None):
|
||||
super().__init__(
|
||||
1912, provider="AWS", file=file, original_exception=original_exception
|
||||
)
|
||||
|
||||
|
||||
class AWSIAMRoleARNEmptyResource(AWSRoleArnError):
|
||||
def __init__(self, file=None, original_exception=None):
|
||||
super().__init__(
|
||||
1913, provider="AWS", file=file, original_exception=original_exception
|
||||
)
|
||||
|
||||
|
||||
class AWSAssumeRoleError(AWSBaseException):
|
||||
def __init__(self, file=None, original_exception=None):
|
||||
super().__init__(
|
||||
1914, provider="AWS", file=file, original_exception=original_exception
|
||||
)
|
||||
@@ -1,13 +1,14 @@
|
||||
import os
|
||||
import re
|
||||
from argparse import ArgumentTypeError
|
||||
|
||||
from prowler.providers.aws.lib.arn.error import (
|
||||
RoleArnParsingEmptyResource,
|
||||
RoleArnParsingIAMRegionNotEmpty,
|
||||
RoleArnParsingInvalidAccountID,
|
||||
RoleArnParsingInvalidResourceType,
|
||||
RoleArnParsingPartitionEmpty,
|
||||
RoleArnParsingServiceNotIAMnorSTS,
|
||||
from prowler.providers.aws.exceptions.exceptions import (
|
||||
AWSIAMRoleARNEmptyResource,
|
||||
AWSIAMRoleARNInvalidAccountID,
|
||||
AWSIAMRoleARNInvalidResourceType,
|
||||
AWSIAMRoleARNPartitionEmpty,
|
||||
AWSIAMRoleARNRegionNotEmtpy,
|
||||
AWSIAMRoleARNServiceNotIAMnorSTS,
|
||||
)
|
||||
from prowler.providers.aws.lib.arn.models import ARN
|
||||
|
||||
@@ -24,7 +25,7 @@ def parse_iam_credentials_arn(arn: str) -> ARN:
|
||||
arn_parsed = ARN(arn)
|
||||
# First check if region is empty (in IAM ARN's region is always empty)
|
||||
if arn_parsed.region:
|
||||
raise RoleArnParsingIAMRegionNotEmpty
|
||||
raise AWSIAMRoleARNRegionNotEmtpy(file=os.path.basename(__file__))
|
||||
else:
|
||||
# check if needed fields are filled:
|
||||
# - partition
|
||||
@@ -33,15 +34,15 @@ def parse_iam_credentials_arn(arn: str) -> ARN:
|
||||
# - resource_type
|
||||
# - resource
|
||||
if arn_parsed.partition is None or arn_parsed.partition == "":
|
||||
raise RoleArnParsingPartitionEmpty
|
||||
raise AWSIAMRoleARNPartitionEmpty(file=os.path.basename(__file__))
|
||||
elif arn_parsed.service != "iam" and arn_parsed.service != "sts":
|
||||
raise RoleArnParsingServiceNotIAMnorSTS
|
||||
raise AWSIAMRoleARNServiceNotIAMnorSTS(file=os.path.basename(__file__))
|
||||
elif (
|
||||
arn_parsed.account_id is None
|
||||
or len(arn_parsed.account_id) != 12
|
||||
or not arn_parsed.account_id.isnumeric()
|
||||
):
|
||||
raise RoleArnParsingInvalidAccountID
|
||||
raise AWSIAMRoleARNInvalidAccountID(file=os.path.basename(__file__))
|
||||
elif (
|
||||
arn_parsed.resource_type != "role"
|
||||
and arn_parsed.resource_type != "user"
|
||||
@@ -49,9 +50,9 @@ def parse_iam_credentials_arn(arn: str) -> ARN:
|
||||
and arn_parsed.resource_type != "root"
|
||||
and arn_parsed.resource_type != "federated-user"
|
||||
):
|
||||
raise RoleArnParsingInvalidResourceType
|
||||
raise AWSIAMRoleARNInvalidResourceType(file=os.path.basename(__file__))
|
||||
elif arn_parsed.resource == "":
|
||||
raise RoleArnParsingEmptyResource
|
||||
raise AWSIAMRoleARNEmptyResource(file=os.path.basename(__file__))
|
||||
else:
|
||||
return arn_parsed
|
||||
|
||||
|
||||
@@ -1,49 +0,0 @@
|
||||
class RoleArnParsingFailedMissingFields(Exception):
|
||||
# The ARN contains a number of fields different than six separated by :"
|
||||
def __init__(self):
|
||||
self.message = "The assumed role ARN contains an invalid number of fields separated by : or it does not start by arn, please input a valid ARN"
|
||||
super().__init__(self.message)
|
||||
|
||||
|
||||
class RoleArnParsingIAMRegionNotEmpty(Exception):
|
||||
# The ARN contains a non-empty value for region, since it is an IAM ARN is not valid
|
||||
def __init__(self):
|
||||
self.message = "The assumed role ARN contains a non-empty value for region, since it is an IAM ARN is not valid, please input a valid ARN"
|
||||
super().__init__(self.message)
|
||||
|
||||
|
||||
class RoleArnParsingPartitionEmpty(Exception):
|
||||
# The ARN contains an empty value for partition
|
||||
def __init__(self):
|
||||
self.message = "The assumed role ARN does not contain a value for partition, please input a valid ARN"
|
||||
super().__init__(self.message)
|
||||
|
||||
|
||||
class RoleArnParsingServiceNotIAMnorSTS(Exception):
|
||||
def __init__(self):
|
||||
self.message = "The assumed role ARN contains a value for service distinct than IAM or STS, please input a valid ARN"
|
||||
super().__init__(self.message)
|
||||
|
||||
|
||||
class RoleArnParsingServiceNotSTS(Exception):
|
||||
def __init__(self):
|
||||
self.message = "The assumed role ARN contains a value for service distinct than STS, please input a valid ARN"
|
||||
super().__init__(self.message)
|
||||
|
||||
|
||||
class RoleArnParsingInvalidAccountID(Exception):
|
||||
def __init__(self):
|
||||
self.message = "The assumed role ARN contains a value for account id empty or invalid, a valid account id must be composed of 12 numbers, please input a valid ARN"
|
||||
super().__init__(self.message)
|
||||
|
||||
|
||||
class RoleArnParsingInvalidResourceType(Exception):
|
||||
def __init__(self):
|
||||
self.message = "The assumed role ARN contains a value for resource type different than role, please input a valid ARN"
|
||||
super().__init__(self.message)
|
||||
|
||||
|
||||
class RoleArnParsingEmptyResource(Exception):
|
||||
def __init__(self):
|
||||
self.message = "The assumed role ARN does not contain a value for resource, please input a valid ARN"
|
||||
super().__init__(self.message)
|
||||
@@ -1,8 +1,9 @@
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from prowler.providers.aws.lib.arn.error import RoleArnParsingFailedMissingFields
|
||||
from prowler.providers.aws.exceptions.exceptions import AWSIAMRoleARNMissingFields
|
||||
|
||||
|
||||
class ARN(BaseModel):
|
||||
@@ -18,7 +19,7 @@ class ARN(BaseModel):
|
||||
# Validate the ARN
|
||||
## Check that arn starts with arn
|
||||
if not arn.startswith("arn:"):
|
||||
raise RoleArnParsingFailedMissingFields
|
||||
raise AWSIAMRoleARNMissingFields(file=os.path.basename(__file__))
|
||||
## Retrieve fields
|
||||
arn_elements = arn.split(":", 5)
|
||||
data = {
|
||||
|
||||
@@ -6,6 +6,7 @@ from mock import patch
|
||||
|
||||
from prowler.lib.cli.parser import ProwlerArgumentParser
|
||||
from prowler.providers.aws.config import ROLE_SESSION_NAME
|
||||
from prowler.providers.aws.exceptions.exceptions import AWSArgumentTypeValidationError
|
||||
from prowler.providers.aws.lib.arguments.arguments import (
|
||||
validate_bucket,
|
||||
validate_role_session_name,
|
||||
@@ -1325,13 +1326,13 @@ class Test_Parser:
|
||||
"role-name?",
|
||||
]
|
||||
for role_name in bad_role_names:
|
||||
with pytest.raises(ArgumentTypeError) as argument_error:
|
||||
with pytest.raises(AWSArgumentTypeValidationError) as argument_error:
|
||||
validate_role_session_name(role_name)
|
||||
|
||||
assert argument_error.type == ArgumentTypeError
|
||||
assert argument_error.type == AWSArgumentTypeValidationError
|
||||
assert (
|
||||
argument_error.value.args[0]
|
||||
== "Role Session Name must be 2-64 characters long and consist only of upper- and lower-case alphanumeric characters with no spaces. You can also include underscores or any of the following characters: =,.@-"
|
||||
== "[1905] AWS argument type validation error - Role Session Name must be between 2 and 64 characters and may contain alphanumeric characters, periods, hyphens, and underscores."
|
||||
)
|
||||
|
||||
def test_validate_role_session_name_valid_role_names(self):
|
||||
|
||||
@@ -2,7 +2,7 @@ import json
|
||||
import os
|
||||
import re
|
||||
import tempfile
|
||||
from argparse import ArgumentTypeError, Namespace
|
||||
from argparse import Namespace
|
||||
from datetime import datetime, timedelta
|
||||
from json import dumps
|
||||
from os import rmdir
|
||||
@@ -28,7 +28,11 @@ from prowler.providers.aws.config import (
|
||||
BOTO3_USER_AGENT_EXTRA,
|
||||
ROLE_SESSION_NAME,
|
||||
)
|
||||
from prowler.providers.aws.lib.arn.error import RoleArnParsingInvalidResourceType
|
||||
from prowler.providers.aws.exceptions.exceptions import (
|
||||
AWSArgumentTypeValidationError,
|
||||
AWSIAMRoleARNInvalidResourceType,
|
||||
AWSNoCredentialsError,
|
||||
)
|
||||
from prowler.providers.aws.lib.arn.models import ARN
|
||||
from prowler.providers.aws.lib.mutelist.mutelist import AWSMutelist
|
||||
from prowler.providers.aws.models import (
|
||||
@@ -1281,13 +1285,16 @@ aws:
|
||||
clear=True,
|
||||
):
|
||||
|
||||
with raises(botocore.exceptions.NoCredentialsError) as exception:
|
||||
with raises(AWSNoCredentialsError) as exception:
|
||||
AwsProvider.test_connection(
|
||||
profile=None
|
||||
) # No profile to avoid ProfileNotFound error
|
||||
|
||||
assert exception.type == botocore.exceptions.NoCredentialsError
|
||||
assert "Unable to locate credentials" in str(exception.value)
|
||||
assert exception.type == AWSNoCredentialsError
|
||||
assert (
|
||||
"AWSNoCredentialsError[1904]: No AWS credentials found - Unable to locate credentials"
|
||||
in str(exception.value)
|
||||
)
|
||||
|
||||
@mock_aws
|
||||
def test_test_connection_with_role_from_env(self, monkeypatch):
|
||||
@@ -1321,12 +1328,13 @@ aws:
|
||||
role_arn = (
|
||||
f"arn:{AWS_COMMERCIAL_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/{role_name}"
|
||||
)
|
||||
with raises(ArgumentTypeError) as exception:
|
||||
with raises(AWSArgumentTypeValidationError) as exception:
|
||||
AwsProvider.test_connection(role_arn=role_arn, session_duration=899)
|
||||
|
||||
assert exception.type == ArgumentTypeError
|
||||
assert exception.type == AWSArgumentTypeValidationError
|
||||
assert (
|
||||
exception.value.args[0] == "Session duration must be between 900 and 43200"
|
||||
exception.value.args[0]
|
||||
== "[1905] AWS argument type validation error - Session Duration must be between 900 and 43200 seconds."
|
||||
)
|
||||
|
||||
@mock_aws
|
||||
@@ -1343,9 +1351,10 @@ aws:
|
||||
|
||||
assert isinstance(connection, Connection)
|
||||
assert not connection.is_connected
|
||||
assert isinstance(connection.error, ArgumentTypeError)
|
||||
assert isinstance(connection.error, AWSArgumentTypeValidationError)
|
||||
assert (
|
||||
connection.error.args[0] == "Session duration must be between 900 and 43200"
|
||||
connection.error.args[0]
|
||||
== "[1905] AWS argument type validation error - Session Duration must be between 900 and 43200 seconds."
|
||||
)
|
||||
|
||||
@mock_aws
|
||||
@@ -1355,13 +1364,13 @@ aws:
|
||||
f"arn:{AWS_COMMERCIAL_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/{role_name}"
|
||||
)
|
||||
|
||||
with raises(ArgumentTypeError) as exception:
|
||||
with raises(AWSArgumentTypeValidationError) as exception:
|
||||
AwsProvider.test_connection(role_arn=role_arn, role_session_name="???")
|
||||
|
||||
assert exception.type == ArgumentTypeError
|
||||
assert exception.type == AWSArgumentTypeValidationError
|
||||
assert (
|
||||
exception.value.args[0]
|
||||
== "Role Session Name must be 2-64 characters long and consist only of upper- and lower-case alphanumeric characters with no spaces. You can also include underscores or any of the following characters: =,.@-"
|
||||
== "[1905] AWS argument type validation error - Role Session Name must be between 2 and 64 characters and may contain alphanumeric characters, periods, hyphens, and underscores."
|
||||
)
|
||||
|
||||
@mock_aws
|
||||
@@ -1369,13 +1378,13 @@ aws:
|
||||
role_name = "test-role"
|
||||
role_arn = f"arn:{AWS_COMMERCIAL_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:not-role/{role_name}"
|
||||
|
||||
with raises(RoleArnParsingInvalidResourceType) as exception:
|
||||
with raises(AWSIAMRoleARNInvalidResourceType) as exception:
|
||||
AwsProvider.test_connection(role_arn=role_arn)
|
||||
|
||||
assert exception.type == RoleArnParsingInvalidResourceType
|
||||
assert exception.type == AWSIAMRoleARNInvalidResourceType
|
||||
assert (
|
||||
exception.value.args[0]
|
||||
== "The assumed role ARN contains a value for resource type different than role, please input a valid ARN"
|
||||
== "[1912] AWS IAM Role ARN resource type is invalid"
|
||||
)
|
||||
|
||||
@mock_aws
|
||||
|
||||
@@ -1,15 +1,15 @@
|
||||
from pytest import raises
|
||||
|
||||
from prowler.providers.aws.lib.arn.arn import is_valid_arn, parse_iam_credentials_arn
|
||||
from prowler.providers.aws.lib.arn.error import (
|
||||
RoleArnParsingEmptyResource,
|
||||
RoleArnParsingFailedMissingFields,
|
||||
RoleArnParsingIAMRegionNotEmpty,
|
||||
RoleArnParsingInvalidAccountID,
|
||||
RoleArnParsingInvalidResourceType,
|
||||
RoleArnParsingPartitionEmpty,
|
||||
RoleArnParsingServiceNotIAMnorSTS,
|
||||
from prowler.providers.aws.exceptions.exceptions import (
|
||||
AWSIAMRoleARNEmptyResource,
|
||||
AWSIAMRoleARNInvalidAccountID,
|
||||
AWSIAMRoleARNInvalidResourceType,
|
||||
AWSIAMRoleARNMissingFields,
|
||||
AWSIAMRoleARNPartitionEmpty,
|
||||
AWSIAMRoleARNRegionNotEmtpy,
|
||||
AWSIAMRoleARNServiceNotIAMnorSTS,
|
||||
)
|
||||
from prowler.providers.aws.lib.arn.arn import is_valid_arn, parse_iam_credentials_arn
|
||||
from prowler.providers.aws.lib.arn.models import ARN
|
||||
|
||||
ACCOUNT_ID = "123456789012"
|
||||
@@ -323,58 +323,58 @@ class Test_ARN_Parsing:
|
||||
assert parsed_arn.resource_type == test["expected"]["resource_type"]
|
||||
assert parsed_arn.resource == test["expected"]["resource"]
|
||||
|
||||
def test_iam_credentials_arn_parsing_raising_RoleArnParsingFailedMissingFields(
|
||||
def test_iam_credentials_arn_parsing_raising_AWSIAMRoleARNMissingFields(
|
||||
self,
|
||||
):
|
||||
input_arn = ""
|
||||
with raises(RoleArnParsingFailedMissingFields) as error:
|
||||
with raises(AWSIAMRoleARNMissingFields) as error:
|
||||
parse_iam_credentials_arn(input_arn)
|
||||
|
||||
assert error._excinfo[0] == RoleArnParsingFailedMissingFields
|
||||
assert error._excinfo[0] == AWSIAMRoleARNMissingFields
|
||||
|
||||
def test_iam_credentials_arn_parsing_raising_RoleArnParsingIAMRegionNotEmpty(self):
|
||||
def test_iam_credentials_arn_parsing_raising_AWSIAMRoleARNRegionNotEmtpy(self):
|
||||
input_arn = "arn:aws:iam:eu-west-1:111111111111:user/prowler"
|
||||
with raises(RoleArnParsingIAMRegionNotEmpty) as error:
|
||||
with raises(AWSIAMRoleARNRegionNotEmtpy) as error:
|
||||
parse_iam_credentials_arn(input_arn)
|
||||
|
||||
assert error._excinfo[0] == RoleArnParsingIAMRegionNotEmpty
|
||||
assert error._excinfo[0] == AWSIAMRoleARNRegionNotEmtpy
|
||||
|
||||
def test_iam_credentials_arn_parsing_raising_RoleArnParsingPartitionEmpty(self):
|
||||
def test_iam_credentials_arn_parsing_raising_AWSIAMRoleARNPartitionEmpty(self):
|
||||
input_arn = "arn::iam::111111111111:user/prowler"
|
||||
with raises(RoleArnParsingPartitionEmpty) as error:
|
||||
with raises(AWSIAMRoleARNPartitionEmpty) as error:
|
||||
parse_iam_credentials_arn(input_arn)
|
||||
|
||||
assert error._excinfo[0] == RoleArnParsingPartitionEmpty
|
||||
assert error._excinfo[0] == AWSIAMRoleARNPartitionEmpty
|
||||
|
||||
def test_iam_credentials_arn_parsing_raising_RoleArnParsingServiceNotIAM(self):
|
||||
def test_iam_credentials_arn_parsing_raising_AWSIAMRoleARNServiceNotIAMnorSTS(self):
|
||||
input_arn = "arn:aws:s3::111111111111:user/prowler"
|
||||
with raises(RoleArnParsingServiceNotIAMnorSTS) as error:
|
||||
with raises(AWSIAMRoleARNServiceNotIAMnorSTS) as error:
|
||||
parse_iam_credentials_arn(input_arn)
|
||||
|
||||
assert error._excinfo[0] == RoleArnParsingServiceNotIAMnorSTS
|
||||
assert error._excinfo[0] == AWSIAMRoleARNServiceNotIAMnorSTS
|
||||
|
||||
def test_iam_credentials_arn_parsing_raising_RoleArnParsingInvalidAccountID(self):
|
||||
def test_iam_credentials_arn_parsing_raising_AWSIAMRoleARNInvalidAccountID(self):
|
||||
input_arn = "arn:aws:iam::AWS_ACCOUNT_ID:user/prowler"
|
||||
with raises(RoleArnParsingInvalidAccountID) as error:
|
||||
with raises(AWSIAMRoleARNInvalidAccountID) as error:
|
||||
parse_iam_credentials_arn(input_arn)
|
||||
|
||||
assert error._excinfo[0] == RoleArnParsingInvalidAccountID
|
||||
assert error._excinfo[0] == AWSIAMRoleARNInvalidAccountID
|
||||
|
||||
def test_iam_credentials_arn_parsing_raising_RoleArnParsingInvalidResourceType(
|
||||
def test_iam_credentials_arn_parsing_raising_AWSIAMRoleARNInvalidResourceType(
|
||||
self,
|
||||
):
|
||||
input_arn = "arn:aws:iam::111111111111:account/prowler"
|
||||
with raises(RoleArnParsingInvalidResourceType) as error:
|
||||
with raises(AWSIAMRoleARNInvalidResourceType) as error:
|
||||
parse_iam_credentials_arn(input_arn)
|
||||
|
||||
assert error._excinfo[0] == RoleArnParsingInvalidResourceType
|
||||
assert error._excinfo[0] == AWSIAMRoleARNInvalidResourceType
|
||||
|
||||
def test_iam_credentials_arn_parsing_raising_RoleArnParsingEmptyResource(self):
|
||||
def test_iam_credentials_arn_parsing_raising_AWSIAMRoleARNEmptyResource(self):
|
||||
input_arn = "arn:aws:iam::111111111111:role/"
|
||||
with raises(RoleArnParsingEmptyResource) as error:
|
||||
with raises(AWSIAMRoleARNEmptyResource) as error:
|
||||
parse_iam_credentials_arn(input_arn)
|
||||
|
||||
assert error._excinfo[0] == RoleArnParsingEmptyResource
|
||||
assert error._excinfo[0] == AWSIAMRoleARNEmptyResource
|
||||
|
||||
def test_is_valid_arn(self):
|
||||
assert is_valid_arn("arn:aws:iam::012345678910:user/test")
|
||||
|
||||
Reference in New Issue
Block a user