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

fix(aws): handle none type attributes (#5216)

Browse Source
This commit is contained in:
Sergio Garcia
2024-09-30 12:04:14 -04:00
committed by GitHub
parent c9ff96144d
commit 3d2f1a3aa7
9 changed files with 170 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/services/apigateway/apigateway_restapi_public/apigateway_restapi_public.py
+1 -1
View File
@@ -15,7 +15,7 @@ class apigateway_restapi_public(Check):
report.resource_tags = rest_api.tags
if rest_api.public_endpoint:
report.status = "FAIL"
report.status_extended = f"API Gateway {rest_api.name} ID {rest_api.id} is internet accesible."
report.status_extended = f"API Gateway {rest_api.name} ID {rest_api.id} is internet accessible."
else:
report.status = "PASS"
report.status_extended = (
prowler/providers/aws/services/ecr/ecr_repositories_not_publicly_accessible/ecr_repositories_not_publicly_accessible.py
+7 -4
View File
@@ -15,11 +15,14 @@ class ecr_repositories_not_publicly_accessible(Check):
report.resource_tags = repository.tags
report.status = "PASS"
report.status_extended = (
f"Repository {repository.name} is not publicly accesible."
f"Repository {repository.name} is not publicly accessible."
)
if is_policy_public(repository.policy):
report.status = "FAIL"
report.status_extended = f"Repository {repository.name} policy may allow anonymous users to perform actions (Principal: '*')."
if repository.policy:
if is_policy_public(repository.policy):
report.status = "FAIL"
report.status_extended = (
f"Repository {repository.name} is publicly accessible."
)
findings.append(report)
prowler/providers/aws/services/organizations/organizations_scp_check_deny_regions/organizations_scp_check_deny_regions.py
+72 -71
View File
@@ -12,86 +12,87 @@ class organizations_scp_check_deny_regions(Check):
)
for org in organizations_client.organizations:
report = Check_Report_AWS(self.metadata())
report.resource_id = org.id
report.resource_arn = org.arn
report.region = organizations_client.region
report.status = "FAIL"
report.status_extended = (
"AWS Organizations is not in-use for this AWS Account."
)
if org.status == "ACTIVE":
if org.policies is not None: # Access denied to list policies
report = Check_Report_AWS(self.metadata())
report.resource_id = org.id
report.resource_arn = org.arn
report.region = organizations_client.region
report.status = "FAIL"
report.status_extended = (
f"AWS Organizations {org.id} does not have SCP policies."
"AWS Organizations is not in-use for this AWS Account."
)
# We use this flag if we find a statement that is restricting regions but not all the configured ones:
is_region_restricted_statement = False
for policy in org.policies.get("SERVICE_CONTROL_POLICY", []):
if org.status == "ACTIVE":
report.status_extended = (
f"AWS Organizations {org.id} does not have SCP policies."
)
# We use this flag if we find a statement that is restricting regions but not all the configured ones:
is_region_restricted_statement = False
# Statements are not always list
statements = policy.content.get("Statement")
if type(policy.content["Statement"]) is not list:
statements = [policy.content.get("Statement")]
for policy in org.policies.get("SERVICE_CONTROL_POLICY", []):
for statement in statements:
# Deny if Condition = {"StringNotEquals": {"aws:RequestedRegion": [region1, region2]}}
if (
statement.get("Effect") == "Deny"
and "Condition" in statement
and "StringNotEquals" in statement["Condition"]
and "aws:RequestedRegion"
in statement["Condition"]["StringNotEquals"]
):
if all(
region
in statement["Condition"]["StringNotEquals"][
"aws:RequestedRegion"
]
for region in organizations_enabled_regions
# Statements are not always list
statements = policy.content.get("Statement")
if type(policy.content["Statement"]) is not list:
statements = [policy.content.get("Statement")]
for statement in statements:
# Deny if Condition = {"StringNotEquals": {"aws:RequestedRegion": [region1, region2]}}
if (
statement.get("Effect") == "Deny"
and "Condition" in statement
and "StringNotEquals" in statement["Condition"]
and "aws:RequestedRegion"
in statement["Condition"]["StringNotEquals"]
):
# All defined regions are restricted, we exit here, no need to continue.
report.status = "PASS"
report.status_extended = f"AWS Organization {org.id} has SCP policy {policy.id} restricting all configured regions found."
findings.append(report)
return findings
else:
# Regions are restricted, but not the ones defined, we keep this finding, but we continue analyzing:
is_region_restricted_statement = True
report.status = "FAIL"
report.status_extended = f"AWS Organization {org.id} has SCP policies {policy.id} restricting some AWS Regions, but not all the configured ones, please check config."
if all(
region
in statement["Condition"]["StringNotEquals"][
"aws:RequestedRegion"
]
for region in organizations_enabled_regions
):
# All defined regions are restricted, we exit here, no need to continue.
report.status = "PASS"
report.status_extended = f"AWS Organization {org.id} has SCP policy {policy.id} restricting all configured regions found."
findings.append(report)
return findings
else:
# Regions are restricted, but not the ones defined, we keep this finding, but we continue analyzing:
is_region_restricted_statement = True
report.status = "FAIL"
report.status_extended = f"AWS Organization {org.id} has SCP policies {policy.id} restricting some AWS Regions, but not all the configured ones, please check config."
# Allow if Condition = {"StringEquals": {"aws:RequestedRegion": [region1, region2]}}
if (
policy.content.get("Statement") == "Allow"
and "Condition" in statement
and "StringEquals" in statement["Condition"]
and "aws:RequestedRegion"
in statement["Condition"]["StringEquals"]
):
if all(
region
in statement["Condition"]["StringEquals"][
"aws:RequestedRegion"
]
for region in organizations_enabled_regions
# Allow if Condition = {"StringEquals": {"aws:RequestedRegion": [region1, region2]}}
if (
policy.content.get("Statement") == "Allow"
and "Condition" in statement
and "StringEquals" in statement["Condition"]
and "aws:RequestedRegion"
in statement["Condition"]["StringEquals"]
):
# All defined regions are restricted, we exit here, no need to continue.
report.status = "PASS"
report.status_extended = f"AWS Organization {org.id} has SCP policy {policy.id} restricting all configured regions found."
findings.append(report)
return findings
else:
# Regions are restricted, but not the ones defined, we keep this finding, but we continue analyzing:
is_region_restricted_statement = True
report.status = "FAIL"
report.status_extended = f"AWS Organization {org.id} has SCP policies {policy.id} restricting some AWS Regions, but not all the configured ones, please check config."
if all(
region
in statement["Condition"]["StringEquals"][
"aws:RequestedRegion"
]
for region in organizations_enabled_regions
):
# All defined regions are restricted, we exit here, no need to continue.
report.status = "PASS"
report.status_extended = f"AWS Organization {org.id} has SCP policy {policy.id} restricting all configured regions found."
findings.append(report)
return findings
else:
# Regions are restricted, but not the ones defined, we keep this finding, but we continue analyzing:
is_region_restricted_statement = True
report.status = "FAIL"
report.status_extended = f"AWS Organization {org.id} has SCP policies {policy.id} restricting some AWS Regions, but not all the configured ones, please check config."
if not is_region_restricted_statement:
report.status = "FAIL"
report.status_extended = f"AWS Organization {org.id} has SCP policies but don't restrict AWS Regions."
if not is_region_restricted_statement:
report.status = "FAIL"
report.status_extended = f"AWS Organization {org.id} has SCP policies but don't restrict AWS Regions."
findings.append(report)
findings.append(report)
return findings
prowler/providers/aws/services/organizations/organizations_service.py
+4 -1
View File
@@ -116,6 +116,9 @@ class Organizations(AWSService):
except ClientError as error:
if error.response["Error"]["Code"] == "AccessDeniedException":
policies = None
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
@@ -219,5 +222,5 @@ class Organization(BaseModel):
id: str
status: str
master_id: str
policies: dict[str, list[Policy]] = {}
policies: Optional[dict[str, list[Policy]]] = {}
delegated_administrators: list[DelegatedAdministrator] = None
prowler/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible.py
+1 -1
View File
@@ -17,7 +17,7 @@ class sns_topics_not_publicly_accessible(Check):
report.resource_tags = topic.tags
report.status = "PASS"
report.status_extended = (
f"SNS topic {topic.name} is not publicly accesible."
f"SNS topic {topic.name} is not publicly accessible."
)
if topic.policy:
for statement in topic.policy["Statement"]:
tests/providers/aws/services/apigateway/apigateway_endpoint_public/apigateway_endpoint_public_test.py
+1 -1
View File
@@ -128,7 +128,7 @@ class Test_apigateway_restapi_public:
assert len(result) == 1
assert (
result[0].status_extended
== f"API Gateway test-rest-api ID {rest_api['id']} is internet accesible."
== f"API Gateway test-rest-api ID {rest_api['id']} is internet accessible."
)
assert result[0].resource_id == "test-rest-api"
assert (
tests/providers/aws/services/ecr/ecr_repositories_not_publicly_accessible/ecr_repositories_not_publicly_accessible_test.py
+45 -2
View File
@@ -122,7 +122,50 @@ class Test_ecr_repositories_not_publicly_accessible:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Repository {repository_name} is not publicly accesible."
== f"Repository {repository_name} is not publicly accessible."
)
assert result[0].resource_id == repository_name
assert result[0].resource_arn == repository_arn
def test_repository_no_policy(self):
ecr_client = mock.MagicMock
ecr_client.registries = {}
ecr_client.registries[AWS_REGION_EU_WEST_1] = Registry(
id=AWS_ACCOUNT_NUMBER,
region=AWS_REGION_EU_WEST_1,
scan_type="BASIC",
repositories=[
Repository(
name=repository_name,
arn=repository_arn,
region=AWS_REGION_EU_WEST_1,
scan_on_push=True,
policy=None,
images_details=None,
lifecycle_policy=None,
)
],
rules=[],
)
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider(),
), mock.patch(
"prowler.providers.aws.services.ecr.ecr_repositories_not_publicly_accessible.ecr_repositories_not_publicly_accessible.ecr_client",
ecr_client,
):
from prowler.providers.aws.services.ecr.ecr_repositories_not_publicly_accessible.ecr_repositories_not_publicly_accessible import (
ecr_repositories_not_publicly_accessible,
)
check = ecr_repositories_not_publicly_accessible()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Repository {repository_name} is not publicly accessible."
)
assert result[0].resource_id == repository_name
assert result[0].resource_arn == repository_arn
@@ -165,7 +208,7 @@ class Test_ecr_repositories_not_publicly_accessible:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Repository {repository_name} policy may allow anonymous users to perform actions (Principal: '*')."
== f"Repository {repository_name} is publicly accessible."
)
assert result[0].resource_id == repository_name
assert result[0].resource_arn == repository_arn
tests/providers/aws/services/organizations/organizations_scp_check_deny_regions/organizations_scp_check_deny_regions_test.py
+37
View File
@@ -233,3 +233,40 @@ class Test_organizations_scp_check_deny_regions:
== f"AWS Organization {org_id} has SCP policy {policy_id} restricting all configured regions found."
)
assert result[0].region == AWS_REGION_EU_WEST_1
@mock_aws
def test_access_denied(self):
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
aws_provider._audit_config = {
"organizations_enabled_regions": [
AWS_REGION_EU_WEST_1,
AWS_REGION_EU_CENTRAL_1,
]
}
# Create Organization
conn = client("organizations", region_name=AWS_REGION_EU_WEST_1)
response = conn.create_organization()
org_arn = response["Organization"]["Arn"]
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_scp_check_deny_regions.organizations_scp_check_deny_regions.organizations_client",
new=Organizations(aws_provider),
) as organizations_client:
# Test Check
from prowler.providers.aws.services.organizations.organizations_scp_check_deny_regions.organizations_scp_check_deny_regions import (
organizations_scp_check_deny_regions,
)
for org in organizations_client.organizations:
if org.arn == org_arn:
org.policies = None
check = organizations_scp_check_deny_regions()
result = check.execute()
assert len(result) == 0
tests/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible_test.py
+2 -2
View File
@@ -139,7 +139,7 @@ class Test_sns_topics_not_publicly_accessible:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"SNS topic {topic_name} is not publicly accesible."
== f"SNS topic {topic_name} is not publicly accessible."
)
assert result[0].resource_id == topic_name
assert result[0].resource_arn == topic_arn
@@ -167,7 +167,7 @@ class Test_sns_topics_not_publicly_accessible:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"SNS topic {topic_name} is not publicly accesible."
== f"SNS topic {topic_name} is not publicly accessible."
)
assert result[0].resource_id == topic_name
assert result[0].resource_arn == topic_arn