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

chore(v3): include latest v4 changes (#4350)

Browse Source
This commit is contained in:
Sergio Garcia
2024-07-02 09:15:47 -04:00
committed by GitHub
parent 9e7ffa6529
commit bd474ef474
41 changed files with 1682 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/tutorials/configuration_file.md
+15 -8
View File
@@ -34,6 +34,8 @@ The following list includes all the AWS checks with configurable variables that
| `guardduty_is_enabled` | `allowlist_non_default_regions` | Boolean |
| `securityhub_enabled` | `allowlist_non_default_regions` | Boolean |
| `rds_instance_backup_enabled` | `check_rds_instance_replicas` | Boolean |
| `acm_certificates_expiration_check` | `days_to_expire_threshold` | Integer |
## Azure
### Configurable Checks
@@ -59,7 +61,6 @@ The following list includes all the Azure checks with configurable variables tha
```yaml title="config.yaml"
# AWS Configuration
aws:
# AWS Global Configuration
# aws.allowlist_non_default_regions --> Allowlist Failed Findings in non-default regions for GuardDuty, SecurityHub, DRS and Config
allowlist_non_default_regions: False
@@ -72,6 +73,7 @@ aws:
# AWS EC2 Configuration
# aws.ec2_elastic_ip_shodan
# TODO: create common config
shodan_api_key: null
# aws.ec2_securitygroup_with_many_ingress_egress_rules --> by default is 50 rules
max_security_group_rules: 50
@@ -124,24 +126,24 @@ aws:
]
# AWS Organizations
# organizations_scp_check_deny_regions
# organizations_enabled_regions: [
# 'eu-central-1',
# 'eu-west-1',
# aws.organizations_scp_check_deny_regions
# aws.organizations_enabled_regions: [
# "eu-central-1",
# "eu-west-1",
# "us-east-1"
# ]
organizations_enabled_regions: []
organizations_trusted_delegated_administrators: []
# AWS ECR
# ecr_repositories_scan_vulnerabilities_in_latest_image
# aws.ecr_repositories_scan_vulnerabilities_in_latest_image
# CRITICAL
# HIGH
# MEDIUM
ecr_repository_vulnerability_minimum_severity: "MEDIUM"
# AWS Trusted Advisor
# trustedadvisor_premium_support_plan_subscribed
# aws.trustedadvisor_premium_support_plan_subscribed
verify_premium_support_plans: True
# AWS RDS
@@ -149,13 +151,18 @@ aws:
# Whether to check RDS instance replicas or not
check_rds_instance_replicas: False
# AWS ACM Configuration
# aws.acm_certificates_expiration_check
days_to_expire_threshold: 7
# Azure Configuration
azure:
# Azure Network Configuration
# azure.network_public_ip_shodan
# TODO: create common config
shodan_api_key: null
# Azure App Configuration
# Azure App Service
# azure.app_ensure_php_version_is_latest
php_latest_version: "8.2"
# azure.app_ensure_python_version_is_latest
docs/tutorials/ignore-unused-services.md
+6
View File
@@ -11,6 +11,12 @@ prowler <provider> --ignore-unused-services
## Services that can be ignored
### AWS
#### ACM
You can have certificates in ACM that is not in use by any AWS resource.
Prowler will check if every certificate is going to expire soon, if this certificate is not in use by default it is not going to be check if it is expired, is going to expire soon or it is good.
- `acm_certificates_expiration_check`
#### Athena
When you create an AWS Account, Athena will create a default primary workgroup for you.
Prowler will check if that workgroup is enabled and if it is being used by checking if there were queries in the last 45 days.
prowler/config/config.yaml
+4
View File
@@ -101,6 +101,10 @@ aws:
# Whether to check RDS instance replicas or not
check_rds_instance_replicas: False
# AWS ACM Configuration
# aws.acm_certificates_expiration_check
days_to_expire_threshold: 7
# Azure Configuration
azure:
# Azure Network Configuration
prowler/lib/check/compliance.py
+1 -1
View File
@@ -34,6 +34,7 @@ def update_checks_metadata_with_compliance(
# Save it into the check's metadata
bulk_checks_metadata[check].Compliance = check_compliance
check_compliance = []
# Add requirements of Manual Controls
for framework in bulk_compliance_frameworks.values():
for requirement in framework.Requirements:
@@ -70,7 +71,6 @@ def update_checks_metadata_with_compliance(
"Recommendation": {"Text": "", "Url": ""},
},
"Categories": [],
"Tags": {},
"DependsOn": [],
"RelatedTo": [],
"Notes": "",
prowler/lib/outputs/outputs.py
+3 -3
View File
@@ -67,15 +67,15 @@ def report(check_findings, output_options, audit_info):
compliance in output_options.output_modes
for compliance in available_compliance_frameworks
):
fill_compliance(
add_manual_controls(
output_options,
finding,
audit_info,
file_descriptors,
)
add_manual_controls(
fill_compliance(
output_options,
finding,
audit_info,
file_descriptors,
)
prowler/providers/aws/aws_regions_by_service.json
+101 -45
View File
@@ -157,7 +157,10 @@
"us-west-1",
"us-west-2"
],
"aws-cn": [],
"aws-cn": [
"cn-north-1",
"cn-northwest-1"
],
"aws-us-gov": [
"us-gov-east-1",
"us-gov-west-1"
@@ -395,15 +398,22 @@
"ap-northeast-2",
"ap-northeast-3",
"ap-south-1",
"ap-south-2",
"ap-southeast-1",
"ap-southeast-2",
"ap-southeast-3",
"ap-southeast-4",
"ca-central-1",
"ca-west-1",
"eu-central-1",
"eu-central-2",
"eu-north-1",
"eu-south-1",
"eu-south-2",
"eu-west-1",
"eu-west-2",
"eu-west-3",
"il-central-1",
"me-central-1",
"me-south-1",
"sa-east-1",
@@ -751,6 +761,18 @@
"aws-us-gov": []
}
},
"apptest": {
"regions": {
"aws": [
"ap-southeast-2",
"eu-central-1",
"sa-east-1",
"us-east-1"
],
"aws-cn": [],
"aws-us-gov": []
}
},
"aps": {
"regions": {
"aws": [
@@ -1165,47 +1187,6 @@
]
}
},
"backupstorage": {
"regions": {
"aws": [
"af-south-1",
"ap-east-1",
"ap-northeast-1",
"ap-northeast-2",
"ap-northeast-3",
"ap-south-1",
"ap-south-2",
"ap-southeast-1",
"ap-southeast-2",
"ap-southeast-3",
"ap-southeast-4",
"ca-central-1",
"eu-central-1",
"eu-central-2",
"eu-north-1",
"eu-south-1",
"eu-south-2",
"eu-west-1",
"eu-west-2",
"eu-west-3",
"me-central-1",
"me-south-1",
"sa-east-1",
"us-east-1",
"us-east-2",
"us-west-1",
"us-west-2"
],
"aws-cn": [
"cn-north-1",
"cn-northwest-1"
],
"aws-us-gov": [
"us-gov-east-1",
"us-gov-west-1"
]
}
},
"batch": {
"regions": {
"aws": [
@@ -1249,6 +1230,15 @@
]
}
},
"bcm-data-exports": {
"regions": {
"aws": [
"us-east-1"
],
"aws-cn": [],
"aws-us-gov": []
}
},
"bedrock": {
"regions": {
"aws": [
@@ -1256,9 +1246,12 @@
"ap-south-1",
"ap-southeast-1",
"ap-southeast-2",
"ca-central-1",
"eu-central-1",
"eu-west-1",
"eu-west-2",
"eu-west-3",
"sa-east-1",
"us-east-1",
"us-west-2"
],
@@ -1442,6 +1435,7 @@
"chime-sdk-meetings": {
"regions": {
"aws": [
"af-south-1",
"ap-northeast-1",
"ap-northeast-2",
"ap-south-1",
@@ -2577,6 +2571,8 @@
"connectcases": {
"regions": {
"aws": [
"ap-northeast-1",
"ap-northeast-2",
"ap-southeast-1",
"ap-southeast-2",
"ca-central-1",
@@ -2609,6 +2605,46 @@
]
}
},
"controlcatalog": {
"regions": {
"aws": [
"af-south-1",
"ap-east-1",
"ap-northeast-1",
"ap-northeast-2",
"ap-northeast-3",
"ap-south-1",
"ap-south-2",
"ap-southeast-1",
"ap-southeast-2",
"ap-southeast-3",
"ap-southeast-4",
"ca-central-1",
"ca-west-1",
"eu-central-1",
"eu-central-2",
"eu-north-1",
"eu-south-1",
"eu-south-2",
"eu-west-1",
"eu-west-2",
"eu-west-3",
"il-central-1",
"me-central-1",
"me-south-1",
"sa-east-1",
"us-east-1",
"us-east-2",
"us-west-1",
"us-west-2"
],
"aws-cn": [],
"aws-us-gov": [
"us-gov-east-1",
"us-gov-west-1"
]
}
},
"controltower": {
"regions": {
"aws": [
@@ -5788,7 +5824,10 @@
"aws-cn": [
"cn-north-1"
],
"aws-us-gov": []
"aws-us-gov": [
"us-gov-east-1",
"us-gov-west-1"
]
}
},
"kms": {
@@ -7037,6 +7076,7 @@
"ap-southeast-3",
"ap-southeast-4",
"ca-central-1",
"ca-west-1",
"eu-central-1",
"eu-central-2",
"eu-north-1",
@@ -7292,6 +7332,7 @@
"ap-southeast-3",
"ap-southeast-4",
"ca-central-1",
"ca-west-1",
"eu-central-1",
"eu-central-2",
"eu-north-1",
@@ -7418,10 +7459,12 @@
"ap-south-1",
"ap-southeast-1",
"ap-southeast-2",
"ca-central-1",
"eu-central-1",
"eu-west-1",
"eu-west-2",
"eu-west-3",
"sa-east-1",
"us-east-1",
"us-east-2",
"us-west-2"
@@ -8305,6 +8348,7 @@
"eu-central-2",
"eu-south-2",
"eu-west-3",
"me-central-1",
"sa-east-1",
"us-east-1",
"us-east-2",
@@ -9273,7 +9317,10 @@
"us-west-2"
],
"aws-cn": [],
"aws-us-gov": []
"aws-us-gov": [
"us-gov-east-1",
"us-gov-west-1"
]
}
},
"serverlessrepo": {
@@ -10228,6 +10275,15 @@
]
}
},
"taxsettings": {
"regions": {
"aws": [
"us-east-1"
],
"aws-cn": [],
"aws-us-gov": []
}
},
"textract": {
"regions": {
"aws": [
@@ -11095,4 +11151,4 @@
}
}
}
}
}
prowler/providers/aws/lib/policy_condition_parser/policy_condition_parser.py
+4
View File
@@ -29,6 +29,8 @@ def is_condition_block_restrictive(
"aws:principalaccount",
"aws:resourceaccount",
"aws:sourcearn",
"aws:sourcevpc",
"aws:sourcevpce",
],
"StringLike": [
"aws:sourceaccount",
@@ -37,6 +39,8 @@ def is_condition_block_restrictive(
"aws:principalarn",
"aws:resourceaccount",
"aws:principalaccount",
"aws:sourcevpc",
"aws:sourcevpce",
],
"ArnLike": ["aws:sourcearn", "aws:principalarn"],
"ArnEquals": ["aws:sourcearn", "aws:principalarn"],
prowler/providers/aws/services/acm/acm_certificates_expiration_check/acm_certificates_expiration_check.py
+24 -21
View File
@@ -1,33 +1,36 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.acm.acm_client import acm_client
DAYS_TO_EXPIRE_THRESHOLD = 7
class acm_certificates_expiration_check(Check):
def execute(self):
findings = []
for certificate in acm_client.certificates:
report = Check_Report_AWS(self.metadata())
report.region = certificate.region
if certificate.expiration_days > DAYS_TO_EXPIRE_THRESHOLD:
report.status = "PASS"
report.status_extended = f"ACM Certificate {certificate.id} for {certificate.name} expires in {certificate.expiration_days} days."
report.resource_id = certificate.id
report.resource_details = certificate.name
report.resource_arn = certificate.arn
report.resource_tags = certificate.tags
else:
report.status = "FAIL"
if certificate.expiration_days < 0:
report.status_extended = f"ACM Certificate {certificate.id} for {certificate.name} has expired ({abs(certificate.expiration_days)} days ago)."
if certificate.in_use or acm_client.provider.scan_unused_services:
report = Check_Report_AWS(self.metadata())
report.region = certificate.region
if certificate.expiration_days > acm_client.audit_config.get(
"days_to_expire_threshold", 7
):
report.status = "PASS"
report.status_extended = f"ACM Certificate {certificate.id} for {certificate.name} expires in {certificate.expiration_days} days."
report.resource_id = certificate.id
report.resource_details = certificate.name
report.resource_arn = certificate.arn
report.resource_tags = certificate.tags
else:
report.status_extended = f"ACM Certificate {certificate.id} for {certificate.name} is about to expire in {certificate.expiration_days} days."
report.status = "FAIL"
if certificate.expiration_days < 0:
report.status_extended = f"ACM Certificate {certificate.id} for {certificate.name} has expired ({abs(certificate.expiration_days)} days ago)."
report.check_metadata.Severity = "high"
else:
report.status_extended = f"ACM Certificate {certificate.id} for {certificate.name} is about to expire in {certificate.expiration_days} days."
report.check_metadata.Severity = "medium"
report.resource_id = certificate.id
report.resource_details = certificate.name
report.resource_arn = certificate.arn
report.resource_tags = certificate.tags
report.resource_id = certificate.id
report.resource_details = certificate.name
report.resource_arn = certificate.arn
report.resource_tags = certificate.tags
findings.append(report)
findings.append(report)
return findings
prowler/providers/aws/services/acm/acm_service.py
+2
View File
@@ -50,6 +50,7 @@ class ACM(AWSService):
id=certificate["CertificateArn"].split("/")[-1],
type=certificate["Type"],
expiration_days=certificate_expiration_time,
in_use=certificate.get("InUse", False),
transparency_logging=False,
region=regional_client.region,
)
@@ -99,5 +100,6 @@ class Certificate(BaseModel):
type: str
tags: Optional[list] = []
expiration_days: int
in_use: bool
transparency_logging: Optional[bool]
region: str
prowler/providers/aws/services/cloudwatch/cloudwatch_service.py
+13 -15
View File
@@ -95,7 +95,7 @@ class Logs(AWSService):
1000 # The threshold for number of events to return per log group.
)
self.__threading_call__(self.__get_log_events__)
self.__list_tags_for_resource__()
self.__threading_call__(self.__list_tags_for_resource__, self.log_groups)
def __describe_metric_filters__(self, regional_client):
logger.info("CloudWatch Logs - Describing metric filters...")
@@ -214,21 +214,19 @@ class Logs(AWSService):
f"CloudWatch Logs - Finished retrieving log events in {regional_client.region}..."
)
def __list_tags_for_resource__(self):
logger.info("CloudWatch Logs - List Tags...")
def __list_tags_for_resource__(self, log_group):
logger.info(f"CloudWatch Logs - List Tags for Log Group {log_group.name}...")
try:
for log_group in self.log_groups:
try:
regional_client = self.regional_clients[log_group.region]
response = regional_client.list_tags_log_group(
logGroupName=log_group.name
)["tags"]
log_group.tags = [response]
except ClientError as error:
if error.response["Error"]["Code"] == "ResourceNotFoundException":
log_group.tags = []
continue
regional_client = self.regional_clients[log_group.region]
response = regional_client.list_tags_for_resource(
resourceArn=log_group.arn
)["tags"]
log_group.tags = [response]
except ClientError as error:
if error.response["Error"]["Code"] == "ResourceNotFoundException":
logger.warning(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
prowler/providers/aws/services/codebuild/codebuild_project_older_90_days/codebuild_project_older_90_days.py
+1 -1
View File
@@ -7,7 +7,7 @@ from prowler.providers.aws.services.codebuild.codebuild_client import codebuild_
class codebuild_project_older_90_days(Check):
def execute(self):
findings = []
for project in codebuild_client.projects:
for project in codebuild_client.projects.values():
report = Check_Report_AWS(self.metadata())
report.region = project.region
report.resource_id = project.name
prowler/providers/aws/services/codebuild/codebuild_project_user_controlled_buildspec/codebuild_project_user_controlled_buildspec.py
+1 -1
View File
@@ -7,7 +7,7 @@ from prowler.providers.aws.services.codebuild.codebuild_client import codebuild_
class codebuild_project_user_controlled_buildspec(Check):
def execute(self):
findings = []
for project in codebuild_client.projects:
for project in codebuild_client.projects.values():
report = Check_Report_AWS(self.metadata())
report.region = project.region
report.resource_id = project.name
prowler/providers/aws/services/codebuild/codebuild_service.py
+53 -33
View File
@@ -1,7 +1,8 @@
import datetime
from dataclasses import dataclass
from typing import Optional
from pydantic import BaseModel
from prowler.lib.logger import logger
from prowler.lib.scan_filters.scan_filters import is_resource_filtered
from prowler.providers.aws.lib.service.service import AWSService
@@ -12,12 +13,16 @@ class Codebuild(AWSService):
def __init__(self, audit_info):
# Call AWSService's __init__
super().__init__(__class__.__name__, audit_info)
self.projects = []
self.projects = {}
self.__threading_call__(self.__list_projects__)
self.__list_builds_for_project__()
self.__threading_call__(
self.__list_builds_for_project__, self.projects.values()
)
self.__threading_call__(self.__batch_get_builds__, self.projects.values())
self.__threading_call__(self.__batch_get_projects__, self.projects.values())
def __list_projects__(self, regional_client):
logger.info("Codebuild - listing projects")
logger.info("Codebuild - Listing projects...")
try:
list_projects_paginator = regional_client.get_paginator("list_projects")
for page in list_projects_paginator.paginate():
@@ -26,14 +31,10 @@ class Codebuild(AWSService):
if not self.audit_resources or (
is_resource_filtered(project_arn, self.audit_resources)
):
self.projects.append(
Project(
name=project,
arn=project_arn,
region=regional_client.region,
last_invoked_time=None,
buildspec=None,
)
self.projects[project_arn] = Project(
name=project,
arn=project_arn,
region=regional_client.region,
)
except Exception as error:
@@ -41,38 +42,57 @@ class Codebuild(AWSService):
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def __list_builds_for_project__(self):
logger.info("Codebuild - listing builds from projects")
def __list_builds_for_project__(self, project):
logger.info("Codebuild - Listing builds...")
try:
for project in self.projects:
for region, client in self.regional_clients.items():
if project.region == region:
ids = client.list_builds_for_project(projectName=project.name)
if "ids" in ids:
if len(ids["ids"]) > 0:
builds = client.batch_get_builds(ids=[ids["ids"][0]])
if "builds" in builds:
if "endTime" in builds["builds"][0]:
project.last_invoked_time = builds["builds"][0][
"endTime"
]
regional_client = self.regional_clients[project.region]
build_ids = regional_client.list_builds_for_project(
projectName=project.name
).get("ids", [])
if len(build_ids) > 0:
project.last_build = Build(id=build_ids[0])
except Exception as error:
logger.error(
f"{project.region}: {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
projects = client.batch_get_projects(names=[project.name])[
"projects"
][0]["source"]
if "buildspec" in projects:
project.buildspec = projects["buildspec"]
def __batch_get_builds__(self, project):
logger.info("Codebuild - Getting builds...")
try:
if project.last_build and project.last_build.id:
regional_client = self.regional_clients[project.region]
builds_by_id = regional_client.batch_get_builds(
ids=[project.last_build.id]
).get("builds", [])
if len(builds_by_id) > 0:
project.last_invoked_time = builds_by_id[0].get("endTime")
except Exception as error:
logger.error(
f"{regional_client.region}: {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def __batch_get_projects__(self, project):
logger.info("Codebuild - Getting projects...")
try:
regional_client = self.regional_clients[project.region]
project_source = regional_client.batch_get_projects(names=[project.name])[
"projects"
][0]["source"]
project.buildspec = project_source.get("buildspec", "")
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
@dataclass
class Project:
class Build(BaseModel):
id: str
class Project(BaseModel):
name: str
arn: str
region: str
last_build: Optional[Build]
last_invoked_time: Optional[datetime.datetime]
buildspec: Optional[str]
prowler/providers/aws/services/ec2/ec2_ebs_volume_snapshots_exists/ec2_ebs_volume_snapshots_exists.metadata.json
+2 -2
View File
@@ -6,10 +6,10 @@
"Data Protection"
],
"ServiceName": "ec2",
"SubServiceName": "snapshot",
"SubServiceName": "volume",
"ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
"Severity": "medium",
"ResourceType": "AwsEc2Snapshot",
"ResourceType": "AwsEc2Volume",
"Description": "Check if EBS snapshots exists.",
"Risk": "Ensure that your EBS volumes (available or in-use) have recent snapshots (taken weekly) available for point-in-time recovery for a better, more reliable data backup strategy.",
"RelatedUrl": "https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSSnapshots.html",
prowler/providers/aws/services/ec2/ec2_instance_managed_by_ssm/ec2_instance_managed_by_ssm.py
+16 -14
View File
@@ -7,21 +7,23 @@ class ec2_instance_managed_by_ssm(Check):
def execute(self):
findings = []
for instance in ec2_client.instances:
if instance.state != "terminated":
report = Check_Report_AWS(self.metadata())
report.region = instance.region
report.resource_arn = instance.arn
report.resource_tags = instance.tags
report.status = "PASS"
report = Check_Report_AWS(self.metadata())
report.region = instance.region
report.resource_arn = instance.arn
report.resource_tags = instance.tags
report.status = "PASS"
report.status_extended = (
f"EC2 Instance {instance.id} is managed by Systems Manager."
)
report.resource_id = instance.id
# instances not running should pass the check
if instance.state in ["pending", "terminated", "stopped"]:
report.status_extended = f"EC2 Instance {instance.id} is unmanaged by Systems Manager because it is {instance.state}."
elif not ssm_client.managed_instances.get(instance.id):
report.status = "FAIL"
report.status_extended = (
f"EC2 Instance {instance.id} is managed by Systems Manager."
f"EC2 Instance {instance.id} is not managed by Systems Manager."
)
report.resource_id = instance.id
if not ssm_client.managed_instances.get(instance.id):
report.status = "FAIL"
report.status_extended = (
f"EC2 Instance {instance.id} is not managed by Systems Manager."
)
findings.append(report)
findings.append(report)
return findings
prowler/providers/aws/services/iam/lib/policy.py
+127
View File
@@ -0,0 +1,127 @@
from ipaddress import ip_address, ip_network
from prowler.lib.logger import logger
def is_policy_cross_account(policy: dict, audited_account: str) -> bool:
"""
is_policy_cross_account checks if the policy allows cross-account access.
Args:
policy (dict): The policy to check.
audited_account (str): The account to check if it has access.
Returns:
bool: True if the policy allows cross-account access, False otherwise.
"""
if policy and "Statement" in policy:
if isinstance(policy["Statement"], list):
for statement in policy["Statement"]:
if statement["Effect"] == "Allow" and "AWS" in statement["Principal"]:
if isinstance(statement["Principal"]["AWS"], list):
for aws_account in statement["Principal"]["AWS"]:
if audited_account not in aws_account or "*" == aws_account:
return True
else:
if (
audited_account not in statement["Principal"]["AWS"]
or "*" == statement["Principal"]["AWS"]
):
return True
else:
statement = policy["Statement"]
if statement["Effect"] == "Allow" and "AWS" in statement["Principal"]:
if isinstance(statement["Principal"]["AWS"], list):
for aws_account in statement["Principal"]["AWS"]:
if audited_account not in aws_account or "*" == aws_account:
return True
else:
if (
audited_account not in statement["Principal"]["AWS"]
or "*" == statement["Principal"]["AWS"]
):
return True
return False
def is_policy_public(policy: dict) -> bool:
"""
is_policy_public checks if the policy is publicly accessible.
If the "Principal" element value is set to { "AWS": "*" } and the policy statement is not using any Condition clauses to filter the access, the selected policy is publicly accessible.
Args:
policy (dict): The policy to check.
Returns:
bool: True if the policy is publicly accessible, False otherwise.
"""
if policy and "Statement" in policy:
for statement in policy["Statement"]:
if (
"Principal" in statement
and (
"*" == statement["Principal"]
or "arn:aws:iam::*:root" in statement["Principal"]
)
and "Condition" not in statement
):
return True
elif "Principal" in statement and "AWS" in statement["Principal"]:
if isinstance(statement["Principal"]["AWS"], str):
principals = [statement["Principal"]["AWS"]]
else:
principals = statement["Principal"]["AWS"]
for principal_arn in principals:
if (
principal_arn == "*" or principal_arn == "arn:aws:iam::*:root"
) and "Condition" not in statement:
return True
return False
def is_condition_restricting_from_private_ip(condition_statement: dict) -> bool:
"""Check if the policy condition is coming from a private IP address.
Keyword arguments:
condition_statement -- The policy condition to check. For example:
{
"IpAddress": {
"aws:SourceIp": "X.X.X.X"
}
}
"""
try:
CONDITION_OPERATOR = "IpAddress"
CONDITION_KEY = "aws:sourceip"
is_from_private_ip = False
if condition_statement.get(CONDITION_OPERATOR, {}):
# We need to transform the condition_statement into lowercase
condition_statement[CONDITION_OPERATOR] = {
k.lower(): v for k, v in condition_statement[CONDITION_OPERATOR].items()
}
if condition_statement[CONDITION_OPERATOR].get(CONDITION_KEY, ""):
if not isinstance(
condition_statement[CONDITION_OPERATOR][CONDITION_KEY], list
):
condition_statement[CONDITION_OPERATOR][CONDITION_KEY] = [
condition_statement[CONDITION_OPERATOR][CONDITION_KEY]
]
for ip in condition_statement[CONDITION_OPERATOR][CONDITION_KEY]:
# Select if IP address or IP network searching in the string for '/'
if "/" in ip:
if not ip_network(ip, strict=False).is_private:
break
else:
if not ip_address(ip).is_private:
break
else:
is_from_private_ip = True
except ValueError:
logger.error(f"Invalid IP: {ip}")
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return is_from_private_ip
prowler/providers/aws/services/rds/rds_service.py
+4 -4
View File
@@ -46,7 +46,7 @@ class RDS(AWSService):
DBInstance(
id=instance["DBInstanceIdentifier"],
arn=arn,
endpoint=instance.get("Endpoint"),
endpoint=instance.get("Endpoint", {}),
engine=instance["Engine"],
engine_version=instance["EngineVersion"],
status=instance["DBInstanceStatus"],
@@ -178,7 +178,7 @@ class RDS(AWSService):
db_cluster = DBCluster(
id=cluster["DBClusterIdentifier"],
arn=db_cluster_arn,
endpoint=cluster.get("Endpoint"),
endpoint=cluster.get("Endpoint", ""),
engine=cluster["Engine"],
status=cluster["Status"],
public=cluster.get("PubliclyAccessible", False),
@@ -356,7 +356,7 @@ class DBInstance(BaseModel):
id: str
# arn:{partition}:rds:{region}:{account}:db:{resource_id}
arn: str
endpoint: Optional[dict]
endpoint: dict
engine: str
engine_version: str
status: str
@@ -381,7 +381,7 @@ class DBInstance(BaseModel):
class DBCluster(BaseModel):
id: str
arn: str
endpoint: Optional[str]
endpoint: str
engine: str
status: str
public: bool
prowler/providers/aws/services/s3/s3_bucket_public_access/s3_bucket_public_access.py
+31 -16
View File
@@ -1,4 +1,10 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.lib.policy_condition_parser.policy_condition_parser import (
is_condition_block_restrictive,
)
from prowler.providers.aws.services.iam.lib.policy import (
is_condition_restricting_from_private_ip,
)
from prowler.providers.aws.services.s3.s3_client import s3_client
from prowler.providers.aws.services.s3.s3control_client import s3control_client
@@ -46,26 +52,35 @@ class s3_bucket_public_access(Check):
# 4. Check bucket policy
if bucket.policy:
for statement in bucket.policy["Statement"]:
for statement in bucket.policy.get("Statement", []):
if (
"Principal" in statement
and "*" == statement["Principal"]
and statement["Effect"] == "Allow"
and not is_condition_block_restrictive(
statement.get("Condition", {}), "", True
)
and (
not is_condition_restricting_from_private_ip(
statement.get("Condition", {})
)
if statement.get("Condition", {}).get(
"IpAddress", {}
)
else True
)
):
report.status = "FAIL"
report.status_extended = f"S3 Bucket {bucket.name} has public access due to bucket policy."
else:
if (
"Principal" in statement
and "AWS" in statement["Principal"]
and statement["Effect"] == "Allow"
):
if isinstance(
statement["Principal"]["AWS"], str
):
principals = [statement["Principal"]["AWS"]]
else:
principals = statement["Principal"]["AWS"]
if "*" == statement["Principal"]:
report.status = "FAIL"
report.status_extended = f"S3 Bucket {bucket.name} has public access due to bucket policy."
elif "AWS" in statement["Principal"]:
principals = (
statement["Principal"]["AWS"]
if isinstance(
statement["Principal"]["AWS"], list
)
else [statement["Principal"]["AWS"]]
)
for principal_arn in principals:
if principal_arn == "*":
report.status = "FAIL"
prowler/providers/aws/services/s3/s3_bucket_secure_transport_policy/s3_bucket_secure_transport_policy.py
+1
View File
@@ -22,6 +22,7 @@ class s3_bucket_secure_transport_policy(Check):
if (
statement["Effect"] == "Deny"
and "Condition" in statement
and "Action" in statement
and (
"s3:PutObject" in statement["Action"]
or "*" in statement["Action"]
prowler/providers/aws/services/sns/sns_topics_kms_encryption_at_rest_enabled/sns_topics_kms_encryption_at_rest_enabled.metadata.json
+1 -1
View File
@@ -7,7 +7,7 @@
"SubServiceName": "",
"ResourceIdTemplate": "arn:aws:sns:region:account-id:topic",
"Severity": "high",
"ResourceType": "AwsSNSTopic",
"ResourceType": "AwsSnsTopic",
"Description": "Ensure there are no SNS Topics unencrypted",
"Risk": "If not enabled sensitive information at rest is not protected.",
"RelatedUrl": "https://docs.aws.amazon.com/sns/latest/dg/sns-server-side-encryption.html",
prowler/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible.metadata.json
+1 -1
View File
@@ -7,7 +7,7 @@
"SubServiceName": "",
"ResourceIdTemplate": "arn:aws:sns:region:account-id:topic",
"Severity": "high",
"ResourceType": "AwsSNSTopic",
"ResourceType": "AwsSnsTopic",
"Description": "Check if SNS topics have policy set as Public",
"Risk": "Publicly accessible services could expose sensitive data to bad actors.",
"RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/sns-topic-policy.html",
prowler/providers/aws/services/ssm/ssm_service.py
+5
View File
@@ -1,4 +1,5 @@
import json
import time
from enum import Enum
from typing import Optional
@@ -145,6 +146,10 @@ class SSM(AWSService):
id=resource_id,
region=regional_client.region,
)
# boto3 does not properly handle throttling exceptions for
# ssm:DescribeInstanceInformation when there are large numbers of instances
# AWS support recommends manually reducing frequency of requests
time.sleep(0.1)
except Exception as error:
logger.error(
prowler/providers/azure/services/network/network_service.py
-1
View File
@@ -6,7 +6,6 @@ from prowler.lib.logger import logger
from prowler.providers.azure.lib.service.service import AzureService
########################## SQLServer
class Network(AzureService):
def __init__(self, audit_info):
super().__init__(NetworkManagementClient, audit_info)
prowler/providers/azure/services/network/network_watcher_enabled/network_watcher_enabled.metadata.json
+3 -3
View File
@@ -1,7 +1,7 @@
{
"Provider": "azure",
"CheckID": "network_watcher_enabled",
"CheckTitle": "Ensure that Network Watcher is 'Enabled'",
"CheckTitle": "Ensure that Network Watcher is 'Enabled' for all locations in the Azure subscription",
"CheckType": [],
"ServiceName": "network",
"SubServiceName": "",
@@ -15,12 +15,12 @@
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "https://www.trendmicro.com/cloudoneconformity-staging/knowledge-base/azure/Network/enable-network-watcher.html#",
"Other": "https://www.trendmicro.com/cloudoneconformity-staging/knowledge-base/azure/Network/enable-network-watcher.html",
"Terraform": ""
},
"Recommendation": {
"Text": "Opting out of Network Watcher automatic enablement is a permanent change. Once you opt-out you cannot opt-in without contacting support.",
"Url": "https://docs.azure.cn/zh-cn/cli/network/watcher?view=azure-cli-latest#az_network_watcher_list"
"Url": "https://learn.microsoft.com/en-us/security/benchmark/azure/security-controls-v2-logging-threat-detection#lt-3-enable-logging-for-azure-network-activities"
}
},
"Categories": [],
prowler/providers/azure/services/network/network_watcher_enabled/network_watcher_enabled.py
+19 -18
View File
@@ -3,25 +3,26 @@ from prowler.providers.azure.services.network.network_client import network_clie
class network_watcher_enabled(Check):
def execute(self) -> Check_Report_Azure:
def execute(self) -> list[Check_Report_Azure]:
findings = []
nw_locations = []
for subscription, network_watchers in network_client.network_watchers.items():
for network_watcher in network_watchers:
nw_locations.append(network_watcher.location)
for subscription, locations in network_client.locations.items():
for location in locations:
report = Check_Report_Azure(self.metadata())
report.subscription = subscription
report.resource_name = "Network Watcher"
report.resource_id = f"/subscriptions/{subscription}/providers/Microsoft.Network/networkWatchers/{location}"
if location not in nw_locations:
report.status = "FAIL"
report.status_extended = f"Network Watcher is not enabled for the location {location} in subscription {subscription}."
findings.append(report)
else:
report.status = "PASS"
report.status_extended = f"Network Watcher is enabled for the location {location} in subscription {subscription}."
findings.append(report)
report = Check_Report_Azure(self.metadata())
report.subscription = subscription
report.resource_name = "Network Watcher"
report.location = "Global"
report.resource_id = f"/subscriptions/{network_client.subscriptions[subscription]}/resourceGroups/NetworkWatcherRG/providers/Microsoft.Network/networkWatchers/NetworkWatcher_*"
missing_locations = set(network_client.locations[subscription]) - set(
network_watcher.location for network_watcher in network_watchers
)
if missing_locations:
report.status = "FAIL"
report.status_extended = f"Network Watcher is not enabled for the following locations in subscription '{subscription}': {', '.join(missing_locations)}."
else:
report.status = "PASS"
report.status_extended = f"Network Watcher is enabled for all locations in subscription '{subscription}'."
findings.append(report)
return findings
tests/config/config_test.py
+49 -5
View File
@@ -24,6 +24,44 @@ def mock_prowler_get_latest_release(_, **kwargs):
return response
old_config_aws = {
"shodan_api_key": None,
"max_security_group_rules": 50,
"max_ec2_instance_age_in_days": 180,
"trusted_account_ids": [],
"log_group_retention_days": 365,
"max_idle_disconnect_timeout_in_seconds": 600,
"max_disconnect_timeout_in_seconds": 300,
"max_session_duration_seconds": 36000,
"obsolete_lambda_runtimes": [
"java8",
"go1.x",
"provided",
"python3.6",
"python2.7",
"python3.7",
"nodejs4.3",
"nodejs4.3-edge",
"nodejs6.10",
"nodejs",
"nodejs8.10",
"nodejs10.x",
"nodejs12.x",
"nodejs14.x",
"dotnet5.0",
"dotnetcore1.0",
"dotnetcore2.0",
"dotnetcore2.1",
"dotnetcore3.1",
"ruby2.5",
"ruby2.7",
],
"organizations_enabled_regions": [],
"organizations_trusted_delegated_administrators": [],
"check_rds_instance_replicas": False,
"days_to_expire_threshold": 7,
}
config_aws = {
"shodan_api_key": None,
"max_security_group_rules": 50,
@@ -59,9 +97,17 @@ config_aws = {
"organizations_enabled_regions": [],
"organizations_trusted_delegated_administrators": [],
"check_rds_instance_replicas": False,
"days_to_expire_threshold": 7,
}
config_azure = {"shodan_api_key": None}
config_azure = {
"shodan_api_key": None,
"php_latest_version": "8.2",
"python_latest_version": "3.12",
"java_latest_version": "17",
}
config_gcp = {"shodan_api_key": None}
class Test_Config:
@@ -181,7 +227,6 @@ class Test_Config:
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
config_test_file = f"{path}/fixtures/config.yaml"
provider = "aws"
assert load_and_validate_config_file(provider, config_test_file) == config_aws
def test_load_and_validate_config_file_gcp(self):
@@ -189,7 +234,7 @@ class Test_Config:
config_test_file = f"{path}/fixtures/config.yaml"
provider = "gcp"
assert load_and_validate_config_file(provider, config_test_file) is None
assert load_and_validate_config_file(provider, config_test_file) == config_gcp
def test_load_and_validate_config_file_azure(self):
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
@@ -201,7 +246,6 @@ class Test_Config:
def test_load_and_validate_config_file_old_format(self):
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
config_test_file = f"{path}/fixtures/config_old.yaml"
assert load_and_validate_config_file("aws", config_test_file) == config_aws
assert load_and_validate_config_file("aws", config_test_file) == old_config_aws
assert load_and_validate_config_file("gcp", config_test_file) == {}
assert load_and_validate_config_file("azure", config_test_file) == {}
tests/config/fixtures/config.yaml
+24 -4
View File
@@ -1,7 +1,10 @@
# AWS Configuration
aws:
# AWS Global Configuration
# AWS EC2 Configuration
# aws.ec2_elastic_ip_shodan
# TODO: create common config
shodan_api_key: null
# aws.ec2_securitygroup_with_many_ingress_egress_rules --> by default is 50 rules
max_security_group_rules: 50
@@ -54,24 +57,41 @@ aws:
]
# AWS Organizations
# organizations_scp_check_deny_regions
# organizations_enabled_regions: [
# 'eu-central-1',
# 'eu-west-1',
# aws.organizations_scp_check_deny_regions
# aws.organizations_enabled_regions: [
# "eu-central-1",
# "eu-west-1",
# "us-east-1"
# ]
organizations_enabled_regions: []
organizations_trusted_delegated_administrators: []
# AWS RDS Configuration
# aws.rds_instance_backup_enabled
# Whether to check RDS instance replicas or not
check_rds_instance_replicas: False
# AWS ACM Configuration
# aws.acm_certificates_expiration_check
days_to_expire_threshold: 7
# Azure Configuration
azure:
# Azure Network Configuration
# azure.network_public_ip_shodan
# TODO: create common config
shodan_api_key: null
# Azure App Service
# azure.app_ensure_php_version_is_latest
php_latest_version: "8.2"
# azure.app_ensure_python_version_is_latest
python_latest_version: "3.12"
# azure.app_ensure_java_version_is_latest
java_latest_version: "17"
# GCP Configuration
gcp:
# GCP Compute Configuration
# gcp.compute_public_address_shodan
shodan_api_key: null
tests/config/fixtures/config_old.yaml
+4
View File
@@ -64,3 +64,7 @@ organizations_trusted_delegated_administrators: []
# aws.rds_instance_backup_enabled
# Whether to check RDS instance replicas or not
check_rds_instance_replicas: False
# AWS ACM Configuration
# aws.acm_certificates_expiration_check
days_to_expire_threshold: 7
tests/lib/check/compliance_check_test.py
+193
View File
@@ -0,0 +1,193 @@
from prowler.lib.check.compliance import update_checks_metadata_with_compliance
from prowler.lib.check.compliance_models import (
CIS_Requirement_Attribute,
CIS_Requirement_Attribute_AssessmentStatus,
CIS_Requirement_Attribute_Profile,
Compliance_Base_Model,
Compliance_Requirement,
)
from prowler.lib.check.models import Check_Metadata_Model
class TestCompliance:
provider = "aws"
def get_custom_framework(self):
return {
"framework1": Compliance_Base_Model(
Framework="Framework1",
Provider="Provider1",
Version="1.0",
Description="Framework 1 Description",
Requirements=[
Compliance_Requirement(
Id="1.1.1",
Description="description",
Attributes=[
CIS_Requirement_Attribute(
Section="1. Identity",
Profile=CIS_Requirement_Attribute_Profile("Level 1"),
AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus(
"Manual"
),
Description="Description",
RationaleStatement="Rationale",
ImpactStatement="Impact",
RemediationProcedure="Remediation",
AuditProcedure="Audit",
AdditionalInformation="Additional",
References="References",
)
],
Checks=["check1", "check2"],
),
# Manual requirement
Compliance_Requirement(
Id="1.1.2",
Description="description",
Attributes=[
CIS_Requirement_Attribute(
Section="1. Identity",
Profile=CIS_Requirement_Attribute_Profile("Level 1"),
AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus(
"Manual"
),
Description="Description",
RationaleStatement="Rationale",
ImpactStatement="Impact",
RemediationProcedure="Remediation",
AuditProcedure="Audit",
AdditionalInformation="Additional",
References="References",
)
],
Checks=[],
),
],
)
}
def get_custom_check_metadata(self):
return {
"check1": Check_Metadata_Model(
Provider="aws",
CheckID="check1",
CheckTitle="Check 1",
CheckType=["type1"],
ServiceName="service1",
SubServiceName="subservice1",
ResourceIdTemplate="template1",
Severity="high",
ResourceType="resource1",
Description="Description 1",
Risk="risk1",
RelatedUrl="url1",
Remediation={
"Code": {
"CLI": "cli1",
"NativeIaC": "native1",
"Other": "other1",
"Terraform": "terraform1",
},
"Recommendation": {"Text": "text1", "Url": "url1"},
},
Categories=["categoryone"],
DependsOn=["dependency1"],
RelatedTo=["related1"],
Notes="notes1",
Compliance=[],
),
"check2": Check_Metadata_Model(
Provider="aws",
CheckID="check2",
CheckTitle="Check 2",
CheckType=["type2"],
ServiceName="service2",
SubServiceName="subservice2",
ResourceIdTemplate="template2",
Severity="medium",
ResourceType="resource2",
Description="Description 2",
Risk="risk2",
RelatedUrl="url2",
Remediation={
"Code": {
"CLI": "cli2",
"NativeIaC": "native2",
"Other": "other2",
"Terraform": "terraform2",
},
"Recommendation": {"Text": "text2", "Url": "url2"},
},
Categories=["categorytwo"],
DependsOn=["dependency2"],
RelatedTo=["related2"],
Notes="notes2",
Compliance=[],
),
}
def test_update_checks_metadata(self):
bulk_compliance_frameworks = self.get_custom_framework()
bulk_checks_metadata = self.get_custom_check_metadata()
updated_metadata = update_checks_metadata_with_compliance(
bulk_compliance_frameworks, bulk_checks_metadata
)
assert "check1" in updated_metadata
assert "check2" in updated_metadata
assert "manual_check" in updated_metadata
manual_compliance = updated_metadata["manual_check"].Compliance[0]
check1_compliance = updated_metadata["check1"].Compliance[0]
assert len(updated_metadata["manual_check"].Compliance) == 1
assert len(updated_metadata["check1"].Compliance) == 1
assert manual_compliance.Framework == "Framework1"
assert manual_compliance.Provider == "Provider1"
assert manual_compliance.Version == "1.0"
assert manual_compliance.Description == "Framework 1 Description"
assert len(manual_compliance.Requirements) == 1
manual_requirement = manual_compliance.Requirements[0]
assert manual_requirement.Id == "1.1.2"
assert manual_requirement.Description == "description"
assert len(manual_requirement.Attributes) == 1
manual_attribute = manual_requirement.Attributes[0]
assert manual_attribute.Section == "1. Identity"
assert manual_attribute.Profile == "Level 1"
assert manual_attribute.AssessmentStatus == "Manual"
assert manual_attribute.Description == "Description"
assert manual_attribute.RationaleStatement == "Rationale"
assert manual_attribute.ImpactStatement == "Impact"
assert manual_attribute.RemediationProcedure == "Remediation"
assert manual_attribute.AuditProcedure == "Audit"
assert manual_attribute.AdditionalInformation == "Additional"
assert manual_attribute.References == "References"
assert len(updated_metadata["check1"].Compliance) == 1
assert check1_compliance.Framework == "Framework1"
assert check1_compliance.Provider == "Provider1"
assert check1_compliance.Version == "1.0"
assert check1_compliance.Description == "Framework 1 Description"
assert len(check1_compliance.Requirements) == 1
check1_requirement = check1_compliance.Requirements[0]
assert check1_requirement.Id == "1.1.1"
assert check1_requirement.Description == "description"
assert len(check1_requirement.Attributes) == 1
check1_attribute = check1_requirement.Attributes[0]
assert check1_attribute.Section == "1. Identity"
assert check1_attribute.Profile == "Level 1"
assert check1_attribute.AssessmentStatus == "Manual"
assert check1_attribute.Description == "Description"
assert check1_attribute.RationaleStatement == "Rationale"
assert check1_attribute.ImpactStatement == "Impact"
assert check1_attribute.RemediationProcedure == "Remediation"
assert check1_attribute.AuditProcedure == "Audit"
assert check1_attribute.AdditionalInformation == "Additional"
assert check1_attribute.References == "References"
tests/providers/aws/lib/policy_condition_parser/policy_condition_parser_test.py
+23
View File
@@ -1366,3 +1366,26 @@ class Test_policy_condition_parser:
assert not is_condition_block_restrictive(
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER, True
)
def test_condition_parser_string_equals_vpc(self):
condition_statement = {"StringEquals": {"aws:SourceVpc": "vpc-123456"}}
assert is_condition_block_restrictive(
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER, True
)
def test_condition_parser_string_equals_vpc_list(self):
condition_statement = {"StringEquals": {"aws:sourcevpc": ["vpc-123456"]}}
assert is_condition_block_restrictive(
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER, True
)
def test_condition_parser_string_equals_vpc_list_not_valid(self):
condition_statement = {
"StringEquals": {"aws:SourceVpc": ["vpc-123456", "vpc-654321"]}
}
assert is_condition_block_restrictive(
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER, True
)
tests/providers/aws/services/acm/acm_certificates_expiration_check/acm_certificates_expiration_check_test.py
+99 -2
View File
@@ -33,6 +33,7 @@ class Test_acm_certificates_expiration_check:
certificate_name = "test-certificate.com"
certificate_type = "AMAZON_ISSUED"
expiration_days = 5
in_use = True
acm_client = mock.MagicMock
acm_client.certificates = [
@@ -42,11 +43,14 @@ class Test_acm_certificates_expiration_check:
name=certificate_name,
type=certificate_type,
expiration_days=expiration_days,
in_use=in_use,
transparency_logging=True,
region=AWS_REGION,
)
]
acm_client.audit_config = {"days_to_expire_threshold": 7}
with mock.patch(
"prowler.providers.aws.services.acm.acm_service.ACM",
new=acm_client,
@@ -76,6 +80,7 @@ class Test_acm_certificates_expiration_check:
certificate_name = "test-certificate.com"
certificate_type = "AMAZON_ISSUED"
expiration_days = -400
in_use = True
acm_client = mock.MagicMock
acm_client.certificates = [
@@ -85,16 +90,18 @@ class Test_acm_certificates_expiration_check:
name=certificate_name,
type=certificate_type,
expiration_days=expiration_days,
in_use=in_use,
transparency_logging=True,
region=AWS_REGION,
)
]
acm_client.audit_config = {"days_to_expire_threshold": 7}
with mock.patch(
"prowler.providers.aws.services.acm.acm_service.ACM",
new=acm_client,
):
# Test Check
from prowler.providers.aws.services.acm.acm_certificates_expiration_check.acm_certificates_expiration_check import (
acm_certificates_expiration_check,
)
@@ -119,6 +126,7 @@ class Test_acm_certificates_expiration_check:
certificate_name = "test-certificate.com"
certificate_type = "AMAZON_ISSUED"
expiration_days = 365
in_use = True
acm_client = mock.MagicMock
acm_client.certificates = [
@@ -128,16 +136,18 @@ class Test_acm_certificates_expiration_check:
name=certificate_name,
type=certificate_type,
expiration_days=expiration_days,
in_use=in_use,
transparency_logging=True,
region=AWS_REGION,
)
]
acm_client.audit_config = {"days_to_expire_threshold": 7}
with mock.patch(
"prowler.providers.aws.services.acm.acm_service.ACM",
new=acm_client,
):
# Test Check
from prowler.providers.aws.services.acm.acm_certificates_expiration_check.acm_certificates_expiration_check import (
acm_certificates_expiration_check,
)
@@ -155,3 +165,90 @@ class Test_acm_certificates_expiration_check:
assert result[0].resource_arn == certificate_arn
assert result[0].region == AWS_REGION
assert result[0].resource_tags == []
def test_acm_certificate_not_in_use(self):
certificate_id = str(uuid.uuid4())
certificate_arn = f"arn:aws:acm:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:certificate/{certificate_id}"
certificate_name = "test-certificate.com"
certificate_type = "AMAZON_ISSUED"
expiration_days = 365
in_use = False
acm_client = mock.MagicMock
acm_client.certificates = [
Certificate(
arn=certificate_arn,
id=certificate_id,
name=certificate_name,
type=certificate_type,
expiration_days=expiration_days,
in_use=in_use,
transparency_logging=True,
region=AWS_REGION,
)
]
acm_client.audit_config = {"days_to_expire_threshold": 7}
acm_client.provider = mock.MagicMock(scan_unused_services=False)
with mock.patch(
"prowler.providers.aws.services.acm.acm_service.ACM",
new=acm_client,
):
from prowler.providers.aws.services.acm.acm_certificates_expiration_check.acm_certificates_expiration_check import (
acm_certificates_expiration_check,
)
check = acm_certificates_expiration_check()
result = check.execute()
assert len(result) == 0
def test_acm_certificate_not_in_use_expired_scan_unused_services(self):
certificate_id = str(uuid.uuid4())
certificate_arn = f"arn:aws:acm:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:certificate/{certificate_id}"
certificate_name = "test-certificate.com"
certificate_type = "AMAZON_ISSUED"
expiration_days = -400
in_use = False
acm_client = mock.MagicMock
acm_client.certificates = [
Certificate(
arn=certificate_arn,
id=certificate_id,
name=certificate_name,
type=certificate_type,
expiration_days=expiration_days,
in_use=in_use,
transparency_logging=True,
region=AWS_REGION,
)
]
acm_client.audit_config = {"days_to_expire_threshold": 7}
acm_client.provider = mock.MagicMock(scan_unused_services=True)
with mock.patch(
"prowler.providers.aws.services.acm.acm_service.ACM",
new=acm_client,
):
from prowler.providers.aws.services.acm.acm_certificates_expiration_check.acm_certificates_expiration_check import (
acm_certificates_expiration_check,
)
check = acm_certificates_expiration_check()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"ACM Certificate {certificate_id} for {certificate_name} has expired ({abs(expiration_days)} days ago)."
)
assert result[0].resource_id == certificate_id
assert result[0].resource_arn == certificate_arn
assert result[0].region == AWS_REGION
assert result[0].resource_tags == []
tests/providers/aws/services/acm/acm_certificates_transparency_logs_enabled/acm_certificates_transparency_logs_enabled_test.py
+2
View File
@@ -41,6 +41,7 @@ class Test_acm_certificates_transparency_logs_enabled:
type=certificate_type,
expiration_days=365,
transparency_logging=True,
in_use=True,
region=AWS_REGION,
)
]
@@ -83,6 +84,7 @@ class Test_acm_certificates_transparency_logs_enabled:
type=certificate_type,
expiration_days=365,
transparency_logging=False,
in_use=True,
region=AWS_REGION,
)
]
tests/providers/aws/services/codebuild/codebuild_project_older_90_days/codebuild_project_older_90_days_test.py
+12 -9
View File
@@ -13,15 +13,16 @@ class Test_codebuild_project_older_90_days:
codebuild_client = mock.MagicMock
project_name = "test-project"
project_arn = f"arn:aws:codebuild:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:project/{project_name}"
codebuild_client.projects = [
Project(
codebuild_client.projects = {
project_arn: Project(
name=project_name,
arn=project_arn,
region="eu-west-1",
last_invoked_time=datetime.now(timezone.utc) - timedelta(days=100),
buildspec=None,
)
]
}
with mock.patch(
"prowler.providers.aws.services.codebuild.codebuild_service.Codebuild",
codebuild_client,
@@ -47,15 +48,16 @@ class Test_codebuild_project_older_90_days:
codebuild_client = mock.MagicMock
project_name = "test-project"
project_arn = f"arn:aws:codebuild:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:project/{project_name}"
codebuild_client.projects = [
Project(
codebuild_client.projects = {
project_arn: Project(
name=project_name,
arn=project_arn,
region="eu-west-1",
last_invoked_time=None,
buildspec=None,
)
]
}
with mock.patch(
"prowler.providers.aws.services.codebuild.codebuild_service.Codebuild",
codebuild_client,
@@ -79,15 +81,16 @@ class Test_codebuild_project_older_90_days:
codebuild_client = mock.MagicMock
project_name = "test-project"
project_arn = f"arn:aws:codebuild:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:project/{project_name}"
codebuild_client.projects = [
Project(
codebuild_client.projects = {
project_arn: Project(
name=project_name,
arn=project_arn,
region="eu-west-1",
last_invoked_time=datetime.now(timezone.utc) - timedelta(days=10),
buildspec=None,
)
]
}
with mock.patch(
"prowler.providers.aws.services.codebuild.codebuild_service.Codebuild",
codebuild_client,
tests/providers/aws/services/codebuild/codebuild_project_user_controlled_buildspec/codebuild_project_user_controlled_buildspec_test.py
+13 -12
View File
@@ -12,15 +12,15 @@ class Test_codebuild_project_user_controlled_buildspec:
codebuild_client = mock.MagicMock
project_name = "test-project"
project_arn = f"arn:aws:codebuild:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:project/{project_name}"
codebuild_client.projects = [
Project(
codebuild_client.projects = {
project_arn: Project(
name=project_name,
arn=project_arn,
region="eu-west-1",
last_invoked_time=None,
buildspec=None,
)
]
}
with mock.patch(
"prowler.providers.aws.services.codebuild.codebuild_service.Codebuild",
codebuild_client,
@@ -47,15 +47,16 @@ class Test_codebuild_project_user_controlled_buildspec:
codebuild_client = mock.MagicMock
project_name = "test-project"
project_arn = f"arn:aws:codebuild:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:project/{project_name}"
codebuild_client.projects = [
Project(
codebuild_client.projects = {
project_arn: Project(
name=project_name,
arn=project_arn,
region="eu-west-1",
last_invoked_time=None,
buildspec="arn:aws:s3:::my-codebuild-sample2/buildspec.out",
)
]
}
with mock.patch(
"prowler.providers.aws.services.codebuild.codebuild_service.Codebuild",
codebuild_client,
@@ -82,15 +83,15 @@ class Test_codebuild_project_user_controlled_buildspec:
codebuild_client = mock.MagicMock
project_name = "test-project"
project_arn = f"arn:aws:codebuild:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:project/{project_name}"
codebuild_client.projects = [
Project(
codebuild_client.projects = {
project_arn: Project(
name=project_name,
arn=project_arn,
region="eu-west-1",
last_invoked_time=None,
buildspec="arn:aws:s3:::my-codebuild-sample2/buildspec.yaml",
)
]
}
with mock.patch(
"prowler.providers.aws.services.codebuild.codebuild_service.Codebuild",
codebuild_client,
@@ -116,15 +117,15 @@ class Test_codebuild_project_user_controlled_buildspec:
codebuild_client = mock.MagicMock
project_name = "test-project"
project_arn = f"arn:aws:codebuild:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:project/{project_name}"
codebuild_client.projects = [
Project(
codebuild_client.projects = {
project_arn: Project(
name=project_name,
arn=project_arn,
region="eu-west-1",
last_invoked_time=None,
buildspec="arn:aws:s3:::my-codebuild-sample2/buildspecyaml",
)
]
}
with mock.patch(
"prowler.providers.aws.services.codebuild.codebuild_service.Codebuild",
codebuild_client,
tests/providers/aws/services/codebuild/codebuild_service_test.py
+39 -12
View File
@@ -3,25 +3,34 @@ from unittest.mock import patch
import botocore
from prowler.providers.aws.services.codebuild.codebuild_service import Codebuild
from prowler.providers.aws.services.codebuild.codebuild_service import (
Build,
Codebuild,
Project,
)
from tests.providers.aws.audit_info_utils import (
AWS_ACCOUNT_NUMBER,
AWS_COMMERCIAL_PARTITION,
AWS_REGION_EU_WEST_1,
set_mocked_aws_audit_info,
)
# last time invoked time
project_name = "test"
project_arn = f"arn:{AWS_COMMERCIAL_PARTITION}:codebuild:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:project/{project_name}"
build_spec_project_arn = "arn:aws:s3:::my-codebuild-sample2/buildspec.yml"
buildspec_type = "S3"
build_id = "test:93f838a7-cd20-48ae-90e5-c10fbbc78ca6"
last_invoked_time = datetime.now() - timedelta(days=2)
# Mocking batch_get_projects
make_api_call = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "ListProjects":
return {"projects": ["test"]}
return {"projects": [project_name]}
if operation_name == "ListBuildsForProject":
return {"ids": ["test:93f838a7-cd20-48ae-90e5-c10fbbc78ca6"]}
return {"ids": [build_id]}
if operation_name == "BatchGetBuilds":
return {"builds": [{"endTime": last_invoked_time}]}
if operation_name == "BatchGetProjects":
@@ -29,7 +38,8 @@ def mock_make_api_call(self, operation_name, kwarg):
"projects": [
{
"source": {
"buildspec": "arn:aws:s3:::my-codebuild-sample2/buildspec.yml"
"type": buildspec_type,
"buildspec": build_spec_project_arn,
}
}
]
@@ -65,16 +75,33 @@ class Test_Codebuild_Service:
def test__list_projects__(self):
codebuild = Codebuild(set_mocked_aws_audit_info())
assert len(codebuild.projects) == 1
assert codebuild.projects[0].name == "test"
assert codebuild.projects[0].region == AWS_REGION_EU_WEST_1
print(codebuild.projects)
assert codebuild.projects[project_arn].name == "test"
assert codebuild.projects[project_arn].region == AWS_REGION_EU_WEST_1
def test__list_builds_for_project__(self):
codebuild = Codebuild(set_mocked_aws_audit_info())
assert len(codebuild.projects) == 1
assert codebuild.projects[0].name == "test"
assert codebuild.projects[0].region == AWS_REGION_EU_WEST_1
assert codebuild.projects[0].last_invoked_time == last_invoked_time
assert codebuild.projects[project_arn].name == "test"
assert codebuild.projects[project_arn].region == AWS_REGION_EU_WEST_1
assert codebuild.projects[project_arn].last_invoked_time == last_invoked_time
assert (
codebuild.projects[0].buildspec
codebuild.projects[project_arn].buildspec
== "arn:aws:s3:::my-codebuild-sample2/buildspec.yml"
)
def test_codebuild_service(self):
codebuild = Codebuild(set_mocked_aws_audit_info())
assert codebuild.session.__class__.__name__ == "Session"
assert codebuild.service == "codebuild"
assert len(codebuild.projects) == 1
assert isinstance(codebuild.projects, dict)
assert isinstance(codebuild.projects[project_arn], Project)
assert codebuild.projects[project_arn].name == project_name
assert codebuild.projects[project_arn].arn == project_arn
assert codebuild.projects[project_arn].region == AWS_REGION_EU_WEST_1
assert codebuild.projects[project_arn].last_invoked_time == last_invoked_time
assert codebuild.projects[project_arn].last_build == Build(id=build_id)
assert codebuild.projects[project_arn].buildspec == build_spec_project_arn
tests/providers/aws/services/ec2/ec2_instance_managed_by_ssm/ec2_instance_managed_by_ssm_test.py
+174
View File
@@ -157,3 +157,177 @@ class Test_ec2_instance_managed_by_ssm_test:
== f"EC2 Instance {instance.id} is managed by Systems Manager."
)
assert result[0].resource_id == instance.id
@mock_aws
def test_ec2_instance_managed_by_ssm_running(self):
ec2 = resource("ec2", region_name=AWS_REGION_US_EAST_1)
instances_pending = ec2.create_instances(
ImageId=EXAMPLE_AMI_ID,
MinCount=2,
MaxCount=2,
UserData="This is some user_data",
)
instance_managed = ec2.Instance(instances_pending[0].id)
instance_unmanaged = ec2.Instance(instances_pending[1].id)
assert instance_managed.state["Name"] == "running"
assert instance_unmanaged.state["Name"] == "running"
ssm_client = mock.MagicMock
ssm_client.managed_instances = {
instance_managed.id: ManagedInstance(
arn=f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:instance/{instance_managed.id}",
id=instance_managed.id,
region=AWS_REGION_US_EAST_1,
)
}
from prowler.providers.aws.services.ec2.ec2_service import EC2
aws_provider = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=aws_provider,
), mock.patch(
"prowler.providers.aws.services.ssm.ssm_service.SSM",
new=ssm_client,
), mock.patch(
"prowler.providers.aws.services.ssm.ssm_client.ssm_client",
new=ssm_client,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_instance_managed_by_ssm.ec2_instance_managed_by_ssm.ec2_client",
new=EC2(aws_provider),
):
# Test Check
from prowler.providers.aws.services.ec2.ec2_instance_managed_by_ssm.ec2_instance_managed_by_ssm import (
ec2_instance_managed_by_ssm,
)
check = ec2_instance_managed_by_ssm()
results = check.execute()
assert len(results) == 2
for result in results:
if result.resource_id == instance_managed.id:
assert result.status == "PASS"
assert result.region == AWS_REGION_US_EAST_1
assert result.resource_tags is None
assert (
result.status_extended
== f"EC2 Instance {instance_managed.id} is managed by Systems Manager."
)
if result.resource_id == instance_unmanaged.id:
assert result.status == "FAIL"
assert result.region == AWS_REGION_US_EAST_1
assert result.resource_tags is None
assert (
result.status_extended
== f"EC2 Instance {instance_unmanaged.id} is not managed by Systems Manager."
)
@mock_aws
def test_ec2_instance_managed_by_ssm_stopped(self):
ec2 = resource("ec2", region_name=AWS_REGION_US_EAST_1)
instances_pending = ec2.create_instances(
ImageId=EXAMPLE_AMI_ID,
MinCount=1,
MaxCount=1,
UserData="This is some user_data",
)
instances_pending[0].stop()
instance = ec2.Instance(instances_pending[0].id)
assert instance.state["Name"] == "stopped"
ssm_client = mock.MagicMock
ssm_client.managed_instances = {}
from prowler.providers.aws.services.ec2.ec2_service import EC2
aws_provider = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=aws_provider,
), mock.patch(
"prowler.providers.aws.services.ssm.ssm_service.SSM",
new=ssm_client,
), mock.patch(
"prowler.providers.aws.services.ssm.ssm_client.ssm_client",
new=ssm_client,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_instance_managed_by_ssm.ec2_instance_managed_by_ssm.ec2_client",
new=EC2(aws_provider),
):
# Test Check
from prowler.providers.aws.services.ec2.ec2_instance_managed_by_ssm.ec2_instance_managed_by_ssm import (
ec2_instance_managed_by_ssm,
)
check = ec2_instance_managed_by_ssm()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags is None
assert (
result[0].status_extended
== f"EC2 Instance {instance.id} is unmanaged by Systems Manager because it is stopped."
)
@mock_aws
def test_ec2_instance_managed_by_ssm_terminated(self):
ec2 = resource("ec2", region_name=AWS_REGION_US_EAST_1)
instances_pending = ec2.create_instances(
ImageId=EXAMPLE_AMI_ID,
MinCount=1,
MaxCount=1,
UserData="This is some user_data",
)
instances_pending[0].terminate()
instance = ec2.Instance(instances_pending[0].id)
assert instance.state["Name"] == "terminated"
ssm_client = mock.MagicMock
ssm_client.managed_instances = {}
from prowler.providers.aws.services.ec2.ec2_service import EC2
aws_provider = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=aws_provider,
), mock.patch(
"prowler.providers.aws.services.ssm.ssm_service.SSM",
new=ssm_client,
), mock.patch(
"prowler.providers.aws.services.ssm.ssm_client.ssm_client",
new=ssm_client,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_instance_managed_by_ssm.ec2_instance_managed_by_ssm.ec2_client",
new=EC2(aws_provider),
):
# Test Check
from prowler.providers.aws.services.ec2.ec2_instance_managed_by_ssm.ec2_instance_managed_by_ssm import (
ec2_instance_managed_by_ssm,
)
check = ec2_instance_managed_by_ssm()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags is None
assert (
result[0].status_extended
== f"EC2 Instance {instance.id} is unmanaged by Systems Manager because it is terminated."
)
tests/providers/aws/services/iam/lib/policy_test.py
+180
View File
@@ -0,0 +1,180 @@
from prowler.providers.aws.services.iam.lib.policy import (
is_condition_restricting_from_private_ip,
is_policy_cross_account,
is_policy_public,
)
from tests.providers.aws.audit_info_utils import AWS_ACCOUNT_NUMBER
class Test_Policy:
def test_is_policy_cross_account(self):
policy1 = {
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": ["arn:aws:iam::123456789012:root", "*"]},
"Action": "s3:*",
"Resource": "arn:aws:s3:::example_bucket/*",
}
]
}
policy2 = {
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": ["arn:aws:iam::123456789012:root"]},
"Action": "s3:*",
"Resource": "arn:aws:s3:::example_bucket/*",
}
]
}
policy3 = {
"Statement": [
{
"Effect": "Deny",
"Principal": {"AWS": ["arn:aws:iam::123456789012:root"]},
"Action": "s3:*",
"Resource": "arn:aws:s3:::example_bucket/*",
}
]
}
assert is_policy_cross_account(policy1, AWS_ACCOUNT_NUMBER)
assert not is_policy_cross_account(policy2, AWS_ACCOUNT_NUMBER)
assert not is_policy_cross_account(policy3, AWS_ACCOUNT_NUMBER)
def test_is_policy_public(self):
policy1 = {
"Statement": [
{
"Effect": "Allow",
"Principal": "*",
"Action": "s3:*",
"Resource": "arn:aws:s3:::example_bucket/*",
}
]
}
policy2 = {
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": "*"},
"Action": "s3:*",
"Resource": "arn:aws:s3:::example_bucket/*",
}
]
}
policy3 = {
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": "arn:aws:iam::123456789012:root"},
"Action": "s3:*",
"Resource": "arn:aws:s3:::example_bucket/*",
}
]
}
policy4 = {
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": "*"},
"Action": "s3:*",
"Resource": "arn:aws:s3:::example_bucket/*",
"Condition": {"IpAddress": {"aws:SourceIp": "192.0.2.0/24"}},
}
]
}
assert is_policy_public(policy1)
assert is_policy_public(policy2)
assert not is_policy_public(policy3)
assert not is_policy_public(policy4)
def test_is_condition_restricting_from_private_ip_no_condition(self):
assert not is_condition_restricting_from_private_ip({})
def test_is_condition_restricting_from_private_ip(self):
condition_from_private_ip = {
"IpAddress": {"aws:SourceIp": "10.0.0.22"},
}
assert is_condition_restricting_from_private_ip(condition_from_private_ip)
def test_is_condition_restricting_from_public_ip(self):
condition_not_from_private_ip = {
"IpAddress": {"aws:SourceIp": "1.2.3.4"},
}
assert not is_condition_restricting_from_private_ip(
condition_not_from_private_ip
)
def test_is_condition_restricting_from_private_ipv6(self):
condition_from_private_ipv6 = {
"IpAddress": {"aws:SourceIp": "fd00::1"},
}
assert is_condition_restricting_from_private_ip(condition_from_private_ipv6)
def test_is_condition_restricting_from_public_ipv6(self):
condition_not_from_private_ipv6 = {
"IpAddress": {"aws:SourceIp": "2001:0db8::1"},
}
assert is_condition_restricting_from_private_ip(condition_not_from_private_ipv6)
def test_is_condition_restricting_from_private_ip_network(self):
condition_from_private_ip_network = {
"IpAddress": {"aws:SourceIp": "10.0.0.0/24"},
}
assert is_condition_restricting_from_private_ip(
condition_from_private_ip_network
)
def test_is_condition_restricting_from_public_ip_network(self):
condition_from_public_ip_network = {
"IpAddress": {"aws:SourceIp": "1.2.3.0/24"},
}
assert not is_condition_restricting_from_private_ip(
condition_from_public_ip_network
)
def test_is_condition_restricting_from_private_ipv6_network(self):
condition_from_private_ipv6_network = {
"IpAddress": {"aws:SourceIp": "fd00::/8"},
}
assert is_condition_restricting_from_private_ip(
condition_from_private_ipv6_network
)
def test_is_condition_restricting_from_private_ip_array(self):
condition_from_private_ip_array = {
"IpAddress": {"aws:SourceIp": ["10.0.0.22", "192.168.1.1"]},
}
assert is_condition_restricting_from_private_ip(condition_from_private_ip_array)
def test_is_condition_restricting_from_private_ipv6_array(self):
condition_from_private_ipv6_array = {
"IpAddress": {"aws:SourceIp": ["fd00::1", "fe80::1"]},
}
assert is_condition_restricting_from_private_ip(
condition_from_private_ipv6_array
)
def test_is_condition_restricting_from_mixed_ip_array(self):
condition_from_mixed_ip_array = {
"IpAddress": {"aws:SourceIp": ["10.0.0.22", "2001:0db8::1"]},
}
assert is_condition_restricting_from_private_ip(condition_from_mixed_ip_array)
def test_is_condition_restricting_from_mixed_ip_array_not_private(self):
condition_from_mixed_ip_array_not_private = {
"IpAddress": {"aws:SourceIp": ["1.2.3.4", "2001:0db8::1"]},
}
assert not is_condition_restricting_from_private_ip(
condition_from_mixed_ip_array_not_private
)
def test_is_condition_restricting_from_private_ip_from_invalid_ip(self):
condition_from_invalid_ip = {
"IpAddress": {"aws:SourceIp": "256.256.256.256"},
}
assert not is_condition_restricting_from_private_ip(condition_from_invalid_ip)
tests/providers/aws/services/s3/s3_bucket_public_access/s3_bucket_public_access_test.py
+357 -13
View File
@@ -351,34 +351,50 @@ class Test_s3_bucket_public_access:
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_bucket_not_public(self):
def test_bucket_not_public_due_to_policy_conditions_from_vpc(self):
s3_client = client("s3", region_name=AWS_REGION_US_EAST_1)
bucket_name_us = "bucket_test_us"
s3_client.create_bucket(Bucket=bucket_name_us)
# Generate S3Control Client
s3control_client = client("s3control", region_name=AWS_REGION_US_EAST_1)
s3control_client.put_public_access_block(
AccountId=AWS_ACCOUNT_NUMBER,
PublicAccessBlockConfiguration={
"BlockPublicAcls": False,
"IgnorePublicAcls": False,
"BlockPublicPolicy": False,
"RestrictPublicBuckets": False,
},
)
s3_client.put_public_access_block(
Bucket=bucket_name_us,
PublicAccessBlockConfiguration={
"BlockPublicAcls": True,
"IgnorePublicAcls": True,
"BlockPublicPolicy": True,
"RestrictPublicBuckets": True,
"BlockPublicAcls": False,
"IgnorePublicAcls": False,
"BlockPublicPolicy": False,
"RestrictPublicBuckets": False,
},
)
public_write_policy = '{"Version": "2012-10-17","Id": "PutObjPolicy","Statement": [{"Sid": "PublicWritePolicy","Effect": "Allow","Principal": "*","Action": "s3:PutObject","Resource": "arn:aws:s3:::bucket_test_us/*","Condition": {"StringEquals": {"aws:SourceVpc": "vpc-123456"}}}]}'
s3_client.put_bucket_policy(
Bucket=bucket_name_us,
Policy=public_write_policy,
)
from prowler.providers.aws.services.s3.s3_service import S3, S3Control
audit_info = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1])
aws_provider = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
new=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3_client",
new=S3(audit_info),
new=S3(aws_provider),
):
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3control_client",
new=S3Control(audit_info),
new=S3Control(aws_provider),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access import (
@@ -390,14 +406,275 @@ class Test_s3_bucket_public_access:
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"not public",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} is not public."
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
== f"arn:{aws_provider.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_bucket_not_public_due_to_policy_conditions_from_private_ip(self):
s3_client = client("s3", region_name=AWS_REGION_US_EAST_1)
bucket_name_us = "bucket_test_us"
s3_client.create_bucket(Bucket=bucket_name_us)
# Generate S3Control Client
s3control_client = client("s3control", region_name=AWS_REGION_US_EAST_1)
s3control_client.put_public_access_block(
AccountId=AWS_ACCOUNT_NUMBER,
PublicAccessBlockConfiguration={
"BlockPublicAcls": False,
"IgnorePublicAcls": False,
"BlockPublicPolicy": False,
"RestrictPublicBuckets": False,
},
)
s3_client.put_public_access_block(
Bucket=bucket_name_us,
PublicAccessBlockConfiguration={
"BlockPublicAcls": False,
"IgnorePublicAcls": False,
"BlockPublicPolicy": False,
"RestrictPublicBuckets": False,
},
)
public_write_policy = '{"Version": "2012-10-17","Id": "PutObjPolicy","Statement": [{"Sid": "PublicWritePolicy","Effect": "Allow","Principal": "*","Action": "s3:PutObject","Resource": "arn:aws:s3:::bucket_test_us/*","Condition": {"IpAddress": {"aws:SourceIp": "10.0.0.25"}}}]}'
s3_client.put_bucket_policy(
Bucket=bucket_name_us,
Policy=public_write_policy,
)
from prowler.providers.aws.services.s3.s3_service import S3, S3Control
aws_provider = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=aws_provider,
), mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3_client",
new=S3(aws_provider),
), mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3control_client",
new=S3Control(aws_provider),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access import (
s3_bucket_public_access,
)
check = s3_bucket_public_access()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} is not public."
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{aws_provider.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_bucket_public_due_to_policy_conditions_from_public_ip(self):
s3_client = client("s3", region_name=AWS_REGION_US_EAST_1)
bucket_name_us = "bucket_test_us"
s3_client.create_bucket(Bucket=bucket_name_us)
# Generate S3Control Client
s3control_client = client("s3control", region_name=AWS_REGION_US_EAST_1)
s3control_client.put_public_access_block(
AccountId=AWS_ACCOUNT_NUMBER,
PublicAccessBlockConfiguration={
"BlockPublicAcls": False,
"IgnorePublicAcls": False,
"BlockPublicPolicy": False,
"RestrictPublicBuckets": False,
},
)
s3_client.put_public_access_block(
Bucket=bucket_name_us,
PublicAccessBlockConfiguration={
"BlockPublicAcls": False,
"IgnorePublicAcls": False,
"BlockPublicPolicy": False,
"RestrictPublicBuckets": False,
},
)
public_write_policy = '{"Version": "2012-10-17","Id": "PutObjPolicy","Statement": [{"Sid": "PublicWritePolicy","Effect": "Allow","Principal": "*","Action": "s3:PutObject","Resource": "arn:aws:s3:::bucket_test_us/*","Condition": {"IpAddress": {"aws:SourceIp": "1.2.3.4"}}}]}'
s3_client.put_bucket_policy(
Bucket=bucket_name_us,
Policy=public_write_policy,
)
from prowler.providers.aws.services.s3.s3_service import S3, S3Control
aws_provider = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=aws_provider,
), mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3_client",
new=S3(aws_provider),
), mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3control_client",
new=S3Control(aws_provider),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access import (
s3_bucket_public_access,
)
check = s3_bucket_public_access()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has public access due to bucket policy."
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{aws_provider.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_bucket_public_due_to_policy_conditions_from_public_and_private_ips(self):
s3_client = client("s3", region_name=AWS_REGION_US_EAST_1)
bucket_name_us = "bucket_test_us"
s3_client.create_bucket(Bucket=bucket_name_us)
# Generate S3Control Client
s3control_client = client("s3control", region_name=AWS_REGION_US_EAST_1)
s3control_client.put_public_access_block(
AccountId=AWS_ACCOUNT_NUMBER,
PublicAccessBlockConfiguration={
"BlockPublicAcls": False,
"IgnorePublicAcls": False,
"BlockPublicPolicy": False,
"RestrictPublicBuckets": False,
},
)
s3_client.put_public_access_block(
Bucket=bucket_name_us,
PublicAccessBlockConfiguration={
"BlockPublicAcls": False,
"IgnorePublicAcls": False,
"BlockPublicPolicy": False,
"RestrictPublicBuckets": False,
},
)
public_write_policy = '{"Version": "2012-10-17","Id": "PutObjPolicy","Statement": [{"Sid": "PublicWritePolicy","Effect": "Allow","Principal": "*","Action": "s3:PutObject","Resource": "arn:aws:s3:::bucket_test_us/*","Condition": {"IpAddress": {"aws:SourceIp": ["192.168.10.0/24", "2001:DB8:1234:5678::/64", "1.2.3.4"]}}}]}'
s3_client.put_bucket_policy(
Bucket=bucket_name_us,
Policy=public_write_policy,
)
from prowler.providers.aws.services.s3.s3_service import S3, S3Control
aws_provider = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=aws_provider,
), mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3_client",
new=S3(aws_provider),
), mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3control_client",
new=S3Control(aws_provider),
):
from prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access import (
s3_bucket_public_access,
)
check = s3_bucket_public_access()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has public access due to bucket policy."
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{aws_provider.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_bucket_not_public(self):
s3_client = client("s3", region_name=AWS_REGION_US_EAST_1)
bucket_name_us = "bucket_test_us"
s3_client.create_bucket(Bucket=bucket_name_us)
# Generate S3Control Client
s3control_client = client("s3control", region_name=AWS_REGION_US_EAST_1)
s3control_client.put_public_access_block(
AccountId=AWS_ACCOUNT_NUMBER,
PublicAccessBlockConfiguration={
"BlockPublicAcls": False,
"IgnorePublicAcls": False,
"BlockPublicPolicy": False,
"RestrictPublicBuckets": False,
},
)
s3_client.put_public_access_block(
Bucket=bucket_name_us,
PublicAccessBlockConfiguration={
"BlockPublicAcls": False,
"IgnorePublicAcls": False,
"BlockPublicPolicy": False,
"RestrictPublicBuckets": False,
},
)
public_write_policy = '{"Version": "2012-10-17","Id": "PutObjPolicy","Statement": [{"Sid": "PublicWritePolicy","Effect": "Allow","Principal": "*","Action": "s3:PutObject","Resource": "arn:aws:s3:::bucket_test_us/*","Condition": {"StringEquals": {"aws:SourceVpc": "vpc-123456"}}}]}'
s3_client.put_bucket_policy(
Bucket=bucket_name_us,
Policy=public_write_policy,
)
from prowler.providers.aws.services.s3.s3_service import S3, S3Control
aws_provider = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3_client",
new=S3(aws_provider),
):
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3control_client",
new=S3Control(aws_provider),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access import (
s3_bucket_public_access,
)
check = s3_bucket_public_access()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} is not public."
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{aws_provider.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == AWS_REGION_US_EAST_1
@@ -443,3 +720,70 @@ class Test_s3_bucket_public_access:
result = check.execute()
assert len(result) == 0
@mock_aws
def test_bucket_public_with_aws_principals(self):
s3_client = client("s3", region_name=AWS_REGION_US_EAST_1)
bucket_name_us = "bucket_test_us"
s3_client.create_bucket(Bucket=bucket_name_us)
# Generate S3Control Client
s3control_client = client("s3control", region_name=AWS_REGION_US_EAST_1)
s3control_client.put_public_access_block(
AccountId=AWS_ACCOUNT_NUMBER,
PublicAccessBlockConfiguration={
"BlockPublicAcls": False,
"IgnorePublicAcls": False,
"BlockPublicPolicy": False,
"RestrictPublicBuckets": False,
},
)
s3_client.put_public_access_block(
Bucket=bucket_name_us,
PublicAccessBlockConfiguration={
"BlockPublicAcls": False,
"IgnorePublicAcls": False,
"BlockPublicPolicy": False,
"RestrictPublicBuckets": False,
},
)
public_write_policy = '{"Version": "2012-10-17","Id": "PutObjPolicy","Statement": [{"Sid": "PublicWritePolicy","Effect": "Allow","Principal": {"AWS": ["arn:aws:iam::123456789012:root", "*"]},"Action": "s3:PutObject","Resource": "arn:aws:s3:::bucket_test_us/*"}]}'
s3_client.put_bucket_policy(
Bucket=bucket_name_us,
Policy=public_write_policy,
)
from prowler.providers.aws.services.s3.s3_service import S3, S3Control
aws_provider = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3_client",
new=S3(aws_provider),
):
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3control_client",
new=S3Control(aws_provider),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access import (
s3_bucket_public_access,
)
check = s3_bucket_public_access()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has public access due to bucket policy."
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{aws_provider.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == AWS_REGION_US_EAST_1
tests/providers/aws/services/s3/s3_bucket_secure_transport_policy/s3_bucket_secure_transport_policy_test.py
+62
View File
@@ -175,3 +175,65 @@ class Test_s3_bucket_secure_transport_policy:
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_bucket_uncomply_policy_without_action(self):
s3_client_us_east_1 = client("s3", region_name=AWS_REGION_US_EAST_1)
bucket_name_us = "bucket_test_us"
s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
ssl_policy = """
{
"Version": "2012-10-17",
"Id": "PutObjPolicy",
"Statement": [
{
"Sid": "s3-bucket-ssl-requests-only",
"Effect": "Deny",
"Principal": "*",
"Resource": "arn:aws:s3:::bucket_test_us/*",
"Condition": {
"Bool": {
"aws:SecureTransport": "false"
}
}
}
]
}
"""
s3_client_us_east_1.put_bucket_policy(
Bucket=bucket_name_us,
Policy=ssl_policy,
)
from prowler.providers.aws.services.s3.s3_service import S3
aws_provider = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy.s3_client",
new=S3(aws_provider),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy import (
s3_bucket_secure_transport_policy,
)
check = s3_bucket_secure_transport_policy()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} allows requests over insecure transport in the bucket policy."
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{aws_provider.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == AWS_REGION_US_EAST_1
tests/providers/azure/azure_fixtures.py
+1
View File
@@ -9,6 +9,7 @@ from prowler.providers.azure.lib.audit_info.models import (
)
AZURE_SUBSCRIPTION = str(uuid4())
AZURE_SUBSCRIPTION_ID = str(uuid4())
# Azure Identity
IDENTITY_ID = "00000000-0000-0000-0000-000000000000"
tests/providers/azure/services/network/network_watcher_enabled/network_watcher_enabled_test.py
+12 -5
View File
@@ -1,7 +1,10 @@
from unittest import mock
from prowler.providers.azure.services.network.network_service import NetworkWatcher
from tests.providers.azure.azure_fixtures import AZURE_SUBSCRIPTION
from tests.providers.azure.azure_fixtures import (
AZURE_SUBSCRIPTION,
AZURE_SUBSCRIPTION_ID,
)
class Test_network_watcher_enabled:
@@ -31,8 +34,9 @@ class Test_network_watcher_enabled:
network_client = mock.MagicMock
locations = ["location"]
network_client.locations = {AZURE_SUBSCRIPTION: locations}
network_client.subscriptions = {AZURE_SUBSCRIPTION: AZURE_SUBSCRIPTION_ID}
network_watcher_name = "Network Watcher"
network_watcher_id = f"/subscriptions/{AZURE_SUBSCRIPTION}/providers/Microsoft.Network/networkWatchers/{locations[0]}"
network_watcher_id = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/NetworkWatcherRG/providers/Microsoft.Network/networkWatchers/NetworkWatcher_*"
network_client.network_watchers = {
AZURE_SUBSCRIPTION: [
@@ -62,18 +66,20 @@ class Test_network_watcher_enabled:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Network Watcher is not enabled for the location {locations[0]} in subscription {AZURE_SUBSCRIPTION}."
== f"Network Watcher is not enabled for the following locations in subscription '{AZURE_SUBSCRIPTION}': location."
)
assert result[0].subscription == AZURE_SUBSCRIPTION
assert result[0].resource_name == network_watcher_name
assert result[0].resource_id == network_watcher_id
assert result[0].location == "Global"
def test_network_valid_network_watchers(self):
network_client = mock.MagicMock
locations = ["location"]
network_client.locations = {AZURE_SUBSCRIPTION: locations}
network_client.subscriptions = {AZURE_SUBSCRIPTION: AZURE_SUBSCRIPTION_ID}
network_watcher_name = "Network Watcher"
network_watcher_id = f"/subscriptions/{AZURE_SUBSCRIPTION}/providers/Microsoft.Network/networkWatchers/{locations[0]}"
network_watcher_id = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/NetworkWatcherRG/providers/Microsoft.Network/networkWatchers/NetworkWatcher_*"
network_client.network_watchers = {
AZURE_SUBSCRIPTION: [
@@ -103,8 +109,9 @@ class Test_network_watcher_enabled:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Network Watcher is enabled for the location {locations[0]} in subscription {AZURE_SUBSCRIPTION}."
== f"Network Watcher is enabled for all locations in subscription '{AZURE_SUBSCRIPTION}'."
)
assert result[0].subscription == AZURE_SUBSCRIPTION
assert result[0].resource_name == network_watcher_name
assert result[0].resource_id == network_watcher_id
assert result[0].location == "Global"