pmh_only/bp-check
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
EC2 Default User 2024-08-05 02:30:34 +00:00
commit 5db67720fd
33 changed files with 1921 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.gitignore vendored Normal file
View File

@ -0,0 +1,6 @@
.vscode
__pycache__
venv
env
bp.json

628
bp-base.json Normal file
View File

@ -0,0 +1,628 @@
{
"ALB": {
"enabled": true,
"rules": {
"alb-http-drop-invalid-header-enabled": {
"enabled": true,
"level": 2
},
"alb-waf-enabled": {
"enabled": true,
"level": 2
},
"elb-cross-zone-load-balancing-enabled": {
"enabled": true,
"level": 2
},
"elb-deletion-protection-enabled": {
"enabled": true,
"level": 1
},
"elb-logging-enabled": {
"enabled": true,
"level": 2
}
}
},
"API GW": {
"enabled": true,
"rules": {
"api-gwv2-access-logs-enabled": {
"enabled": true,
"level": 2
},
"api-gwv2-authorization-type-configured": {
"enabled": true,
"level": 1
},
"api-gw-associated-with-waf": {
"enabled": true,
"level": 2
},
"api-gw-cache-enabled-and-encrypted": {
"enabled": true,
"level": 2
},
"api-gw-execution-logging-enabled": {
"enabled": true,
"level": 2
},
"api-gw-xray-enabled": {
"enabled": true,
"level": 1
}
}
},
"RDS": {
"enabled": true,
"rules": {
"aurora-last-backup-recovery-point-created": {
"enabled": true,
"level": 2
},
"aurora-mysql-backtracking-enabled": {
"enabled": true,
"level": 2
},
"db-instance-backup-enabled": {
"enabled": true,
"level": 2
},
"rds-cluster-auto-minor-version-upgrade-enable": {
"enabled": true,
"level": 2
},
"rds-cluster-default-admin-check": {
"enabled": true,
"level": 2
},
"rds-cluster-deletion-protection-enabled": {
"enabled": true,
"level": 1
},
"rds-cluster-encrypted-at-rest": {
"enabled": true,
"level": 2
},
"rds-cluster-iam-authentication-enabled": {
"enabled": true,
"level": 2
},
"rds-cluster-multi-az-enabled": {
"enabled": true,
"level": 2
},
"rds-db-security-group-not-allowed": {
"enabled": true,
"level": 2
},
"rds-enhanced-monitoring-enabled": {
"enabled": true,
"level": 2
},
"rds-instance-deletion-protection-enabled": {
"enabled": true,
"level": 1
},
"rds-instance-public-access-check": {
"enabled": true,
"level": 2
},
"rds-logging-enabled": {
"enabled": true,
"level": 2
},
"rds-snapshot-encrypted": {
"enabled": true,
"level": 2
}
}
},
"ASG": {
"enabled": true,
"rules": {
"autoscaling-group-elb-healthcheck-required": {
"enabled": true,
"level": 2
},
"autoscaling-multiple-az": {
"enabled": true,
"level": 2
}
}
},
"EC2": {
"enabled": true,
"rules": {
"autoscaling-launch-template": {
"enabled": true,
"level": 2
},
"ec2-ebs-encryption-by-default": {
"enabled": true,
"level": 2
},
"ec2-imdsv2-check": {
"enabled": true,
"level": 2
},
"ec2-instance-detailed-monitoring-enabled": {
"enabled": true,
"level": 2
},
"ec2-instance-managed-by-systems-manager": {
"enabled": true,
"level": 2
},
"ec2-instance-profile-attached": {
"enabled": true,
"level": 2
},
"ec2-no-amazon-key-pair": {
"enabled": true,
"level": 1
},
"ec2-stopped-instance": {
"enabled": true,
"level": 2
},
"ec2-token-hop-limit-check": {
"enabled": true,
"level": 2
}
}
},
"CloudFront": {
"enabled": true,
"rules": {
"cloudfront-accesslogs-enabled": {
"enabled": true,
"level": 2
},
"cloudfront-associated-with-waf": {
"enabled": true,
"level": 2
},
"cloudfront-default-root-object-configured": {
"enabled": true,
"level": 2
},
"cloudfront-no-deprecated-ssl-protocols": {
"enabled": true,
"level": 2
},
"cloudfront-s3-origin-access-control-enabled": {
"enabled": true,
"level": 2
},
"cloudfront-viewer-policy-https": {
"enabled": true,
"level": 2
}
}
},
"KMS": {
"enabled": true,
"rules": {
"cmk-backing-key-rotation-enabled": {
"enabled": true,
"level": 2
}
}
},
"CodeSeries": {
"enabled": true,
"rules": {
"codebuild-project-environment-privileged-check": {
"enabled": true,
"level": 1
},
"codebuild-project-logging-enabled": {
"enabled": true,
"level": 2
},
"codedeploy-auto-rollback-monitor-enabled": {
"enabled": true,
"level": 2
}
}
},
"CloudWatch": {
"enabled": true,
"rules": {
"cw-loggroup-retention-period-check": {
"enabled": true,
"level": 2
},
"cloudwatch-alarm-settings-check": {
"enabled": true,
"level": 2
}
}
},
"DocDB": {
"enabled": true,
"rules": {
"docdb-cluster-audit-logging-enabled": {
"enabled": true,
"level": 2
},
"docdb-cluster-backup-retention-check": {
"enabled": true,
"level": 2
},
"docdb-cluster-deletion-protection-enabled": {
"enabled": true,
"level": 1
},
"docdb-cluster-encrypted": {
"enabled": true,
"level": 2
}
}
},
"DynamoDB": {
"enabled": true,
"rules": {
"dynamodb-autoscaling-enabled": {
"enabled": true,
"level": 2
},
"dynamodb-last-backup-recovery-point-created": {
"enabled": true,
"level": 2
},
"dynamodb-pitr-enabled": {
"enabled": true,
"level": 2
},
"dynamodb-table-deletion-protection-enabled": {
"enabled": true,
"level": 1
},
"dynamodb-table-encrypted-kms": {
"enabled": true,
"level": 2
},
"dynamodb-table-encryption-enabled": {
"enabled": true,
"level": 2
}
}
},
"ECR": {
"enabled": true,
"rules": {
"ecr-private-image-scanning-enabled": {
"enabled": true,
"level": 2
},
"ecr-private-lifecycle-policy-configured": {
"enabled": true,
"level": 2
},
"ecr-private-tag-immutability-enabled": {
"enabled": true,
"level": 2
},
"ecr-kms-encryption-1": {
"enabled": true,
"level": 2
}
}
},
"ECS": {
"enabled": true,
"rules": {
"ecs-awsvpc-networking-enabled": {
"enabled": true,
"level": 2
},
"ecs-containers-nonprivileged": {
"enabled": true,
"level": 2
},
"ecs-containers-readonly-access": {
"enabled": true,
"level": 2
},
"ecs-container-insights-enabled": {
"enabled": true,
"level": 2
},
"ecs-fargate-latest-platform-version": {
"enabled": true,
"level": 2
},
"ecs-task-definition-log-configuration": {
"enabled": true,
"level": 2
},
"ecs-task-definition-memory-hard-limit": {
"enabled": true,
"level": 1
},
"ecs-task-definition-nonroot-user": {
"enabled": true,
"level": 1
}
}
},
"EFS": {
"enabled": true,
"rules": {
"efs-access-point-enforce-root-directory": {
"enabled": true,
"level": 2
},
"efs-access-point-enforce-user-identity": {
"enabled": true,
"level": 2
},
"efs-automatic-backups-enabled": {
"enabled": true,
"level": 2
},
"efs-encrypted-check": {
"enabled": true,
"level": 2
},
"efs-mount-target-public-accessible": {
"enabled": true,
"level": 2
}
}
},
"EKS": {
"enabled": true,
"rules": {
"eks-cluster-logging-enabled": {
"enabled": true,
"level": 2
},
"eks-cluster-secrets-encrypted": {
"enabled": true,
"level": 2
},
"eks-endpoint-no-public-access": {
"enabled": true,
"level": 1
},
"eks-secrets-encrypted": {
"enabled": true,
"level": 2
}
}
},
"ElastiCache": {
"enabled": true,
"rules": {
"elasticache-auto-minor-version-upgrade-check": {
"enabled": true,
"level": 2
},
"elasticache-redis-cluster-automatic-backup-check": {
"enabled": true,
"level": 2
},
"elasticache-repl-grp-auto-failover-enabled": {
"enabled": true,
"level": 2
},
"elasticache-repl-grp-encrypted-at-rest": {
"enabled": true,
"level": 2
},
"elasticache-repl-grp-encrypted-in-transit": {
"enabled": true,
"level": 2
},
"elasticache-subnet-group-check": {
"enabled": true,
"level": 2
}
}
},
"IAM": {
"enabled": true,
"rules": {
"iam-policy-no-statements-with-admin-access": {
"enabled": true,
"level": 1
},
"iam-policy-no-statements-with-full-access": {
"enabled": true,
"level": 1
},
"iam-role-managed-policy-check": {
"enabled": true,
"level": 1
}
}
},
"Lambda": {
"enabled": true,
"rules": {
"lambda-dlq-check": {
"enabled": true,
"level": 1
},
"lambda-function-public-access-prohibited": {
"enabled": true,
"level": 2
},
"lambda-function-settings-check": {
"enabled": true,
"level": 2
},
"lambda-inside-vpc": {
"enabled": true,
"level": 1
}
}
},
"Tags": {
"enabled": true,
"rules": {
"required-tags": {
"enabled": true,
"level": 2
}
}
},
"Route53": {
"enabled": true,
"rules": {
"route53-query-logging-enabled": {
"enabled": true,
"level": 2
}
}
},
"S3": {
"enabled": true,
"rules": {
"s3-access-point-in-vpc-only": {
"enabled": true,
"level": 1
},
"s3-bucket-default-lock-enabled": {
"enabled": true,
"level": 1
},
"s3-bucket-level-public-access-prohibited": {
"enabled": true,
"level": 2
},
"s3-bucket-logging-enabled": {
"enabled": true,
"level": 1
},
"s3-bucket-ssl-requests-only": {
"enabled": true,
"level": 2
},
"s3-bucket-versioning-enabled": {
"enabled": true,
"level": 2
},
"s3-default-encryption-kms": {
"enabled": true,
"level": 2
},
"s3-event-notifications-enabled": {
"enabled": true,
"level": 1
},
"s3-last-backup-recovery-point-created": {
"enabled": true,
"level": 1
},
"s3-lifecycle-policy-check": {
"enabled": true,
"level": 2
}
}
},
"Secrets Manager": {
"enabled": true,
"rules": {
"secretsmanager-rotation-enabled-check": {
"enabled": true,
"level": 2
},
"secretsmanager-scheduled-rotation-success-check": {
"enabled": true,
"level": 1
},
"secretsmanager-secret-periodic-rotation": {
"enabled": true,
"level": 2
}
}
},
"Security Hub": {
"enabled": true,
"rules": {
"securityhub-enabled": {
"enabled": true,
"level": 1
}
}
},
"SNS": {
"enabled": true,
"rules": {
"sns-encrypted-kms": {
"enabled": true,
"level": 2
},
"sns-topic-message-delivery-notification-enabled": {
"enabled": true,
"level": 2
}
}
},
"VPC": {
"enabled": true,
"rules": {
"ec2-transit-gateway-auto-vpc-attach-disabled": {
"enabled": true,
"level": 1
},
"restricted-ssh": {
"enabled": true,
"level": 2
},
"restricted-common-ports": {
"enabled": true,
"level": 2
},
"subnet-auto-assign-public-ip-disabled": {
"enabled": true,
"level": 1
},
"vpc-default-security-group-closed": {
"enabled": true,
"level": 2
},
"vpc-flow-logs-enabled": {
"enabled": true,
"level": 2
},
"vpc-network-acl-unused-check": {
"enabled": true,
"level": 2
},
"vpc-peering-dns-resolution-check": {
"enabled": true,
"level": 2
},
"vpc-sg-open-only-to-authorized-ports": {
"enabled": true,
"level": 2
}
}
},
"WAFv2": {
"enabled": true,
"rules": {
"wafv2-logging-enabled": {
"enabled": true,
"level": 2
},
"wafv2-rulegroup-logging-enabled": {
"enabled": true,
"level": 2
},
"wafv2-rulegroup-not-empty": {
"enabled": true,
"level": 2
},
"wafv2-webacl-not-empty": {
"enabled": true,
"level": 2
}
}
}
}

93
main.py Normal file
View File

@ -0,0 +1,93 @@
import json
from pprint import pprint
from PyInquirer import prompt, style_from_dict, Token
from colorama import Style, Fore
from utils import *
import services
custom_style_2 = style_from_dict(
{
Token.Separator: "#6C6C6C",
Token.QuestionMark: "#FF9D00 bold",
# Token.Selected: '', # default
Token.Selected: "#5F819D",
Token.Pointer: "#FF9D00 bold",
Token.Instruction: "", # default
Token.Answer: "#5F819D bold",
Token.Question: "",
}
)
def ask_services_to_enable(bp):
cli_questions = [
{
"type": "checkbox",
"message": "Select AWS Services to inspect",
"name": "services",
"choices": [
{"name": k, "checked": bool(v["enabled"])} for k, v in bp.items()
],
}
]
answers = prompt(questions=cli_questions, style=custom_style_2)
for service in bp.keys():
bp[service]["enabled"] = service in answers["services"]
return bp
def perform_bp_rules_check(bp):
for service_name, service in bp.items():
if not service["enabled"]:
continue
if service_name == "Lambda":
service_name = "_lambda"
module = getattr(services, convert_snake_case(service_name))
for rule_name, rule in service["rules"].items():
if not rule["enabled"]:
continue
rule["result"] = getattr(module, convert_snake_case(rule_name))()
return bp
def show_bp_result(bp):
for service_name, service in bp.items():
if not service["enabled"]:
continue
print(f"{'=' * 25} {service_name + ' ':=<30}")
for rule_name, rule in service["rules"].items():
if not rule["enabled"]:
continue
if rule["result"].passed:
style = Style.DIM
color = Fore.GREEN
mark = "✅"
elif rule["level"] == 2 and not rule["result"].passed:
style = Style.BRIGHT
color = Fore.RED
mark = "❌"
elif rule["level"] == 1 and not rule["result"].passed:
style = Style.NORMAL
color = Fore.LIGHTRED_EX
mark = "❕"
print(f"{style}{rule_name:50}{Style.RESET_ALL} - {color}{mark}{Fore.RESET}")
for resource in rule["result"].non_compliant_resources:
print(f" - {color}{resource}{Fore.RESET}")
print()
if __name__ == "__main__":
bp = load_bp_from_file()
bp = ask_services_to_enable(bp)
save_bp_to_file(bp)
bp = perform_bp_rules_check(bp)
show_bp_result(bp)

7
models.py Normal file
View File

@ -0,0 +1,7 @@
from pydantic import BaseModel
class RuleCheckResult(BaseModel):
passed: bool
compliant_resources: list[str]
non_compliant_resources: list[str]

17
requirements.txt Normal file
View File

@ -0,0 +1,17 @@
annotated-types==0.7.0
boto3==1.34.153
botocore==1.34.153
colorama==0.4.6
jmespath==1.0.1
prompt-toolkit==1.0.14
pydantic==2.8.2
pydantic_core==2.20.1
Pygments==2.18.0
PyInquirer==1.0.3
python-dateutil==2.9.0.post0
regex==2024.7.24
s3transfer==0.10.2
six==1.16.0
typing_extensions==4.12.2
urllib3==1.26.19
wcwidth==0.2.13

28
services/__init__.py Normal file
View File

@ -0,0 +1,28 @@
from . import (
alb,
api_gw,
rds,
asg,
ec2,
cloudfront,
kms,
codeseries,
cloudwatch,
docdb,
dynamodb,
ecr,
ecs,
efs,
eks,
elasticache,
iam,
_lambda,
tags,
route53,
s3,
secrets_manager,
security_hub,
sns,
vpc,
wafv2,
)

29
services/_lambda.py Normal file
View File

@ -0,0 +1,29 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def lambda_dlq_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def lambda_function_public_access_prohibited():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def lambda_function_settings_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def lambda_inside_vpc():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

37
services/alb.py Normal file
View File

@ -0,0 +1,37 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def alb_http_drop_invalid_header_enabled():
return RuleCheckResult(
passed=False,
compliant_resources=[],
non_compliant_resources=[],
)
def alb_waf_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def elb_cross_zone_load_balancing_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def elb_deletion_protection_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def elb_logging_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

41
services/api_gw.py Normal file
View File

@ -0,0 +1,41 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def api_gwv2_access_logs_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def api_gwv2_authorization_type_configured():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def api_gw_associated_with_waf():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def api_gw_cache_enabled_and_encrypted():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def api_gw_execution_logging_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def api_gw_xray_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

17
services/asg.py Normal file
View File

@ -0,0 +1,17 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def autoscaling_group_elb_healthcheck_required():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def autoscaling_multiple_az():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

41
services/cloudfront.py Normal file
View File

@ -0,0 +1,41 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def cloudfront_accesslogs_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def cloudfront_associated_with_waf():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def cloudfront_default_root_object_configured():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def cloudfront_no_deprecated_ssl_protocols():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def cloudfront_s3_origin_access_control_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def cloudfront_viewer_policy_https():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

17
services/cloudwatch.py Normal file
View File

@ -0,0 +1,17 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def cw_loggroup_retention_period_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def cloudwatch_alarm_settings_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

23
services/codeseries.py Normal file
View File

@ -0,0 +1,23 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def codebuild_project_environment_privileged_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def codebuild_project_logging_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def codedeploy_auto_rollback_monitor_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

29
services/docdb.py Normal file
View File

@ -0,0 +1,29 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def docdb_cluster_audit_logging_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def docdb_cluster_backup_retention_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def docdb_cluster_deletion_protection_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def docdb_cluster_encrypted():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

41
services/dynamodb.py Normal file
View File

@ -0,0 +1,41 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def dynamodb_autoscaling_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def dynamodb_last_backup_recovery_point_created():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def dynamodb_pitr_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def dynamodb_table_deletion_protection_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def dynamodb_table_encrypted_kms():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def dynamodb_table_encryption_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

59
services/ec2.py Normal file
View File

@ -0,0 +1,59 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def autoscaling_launch_template():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ec2_ebs_encryption_by_default():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ec2_imdsv2_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ec2_instance_detailed_monitoring_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ec2_instance_managed_by_systems_manager():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ec2_instance_profile_attached():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ec2_no_amazon_key_pair():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ec2_stopped_instance():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ec2_token_hop_limit_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

29
services/ecr.py Normal file
View File

@ -0,0 +1,29 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def ecr_private_image_scanning_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ecr_private_lifecycle_policy_configured():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ecr_private_tag_immutability_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ecr_kms_encryption_1():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

53
services/ecs.py Normal file
View File

@ -0,0 +1,53 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def ecs_awsvpc_networking_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ecs_containers_nonprivileged():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ecs_containers_readonly_access():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ecs_container_insights_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ecs_fargate_latest_platform_version():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ecs_task_definition_log_configuration():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ecs_task_definition_memory_hard_limit():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def ecs_task_definition_nonroot_user():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

35
services/efs.py Normal file
View File

@ -0,0 +1,35 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def efs_access_point_enforce_root_directory():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def efs_access_point_enforce_user_identity():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def efs_automatic_backups_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def efs_encrypted_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def efs_mount_target_public_accessible():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

29
services/eks.py Normal file
View File

@ -0,0 +1,29 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def eks_cluster_logging_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def eks_cluster_secrets_encrypted():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def eks_endpoint_no_public_access():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def eks_secrets_encrypted():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

41
services/elasticache.py Normal file
View File

@ -0,0 +1,41 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def elasticache_auto_minor_version_upgrade_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def elasticache_redis_cluster_automatic_backup_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def elasticache_repl_grp_auto_failover_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def elasticache_repl_grp_encrypted_at_rest():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def elasticache_repl_grp_encrypted_in_transit():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def elasticache_subnet_group_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

23
services/iam.py Normal file
View File

@ -0,0 +1,23 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def iam_policy_no_statements_with_admin_access():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def iam_policy_no_statements_with_full_access():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def iam_role_managed_policy_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

11
services/kms.py Normal file
View File

@ -0,0 +1,11 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def cmk_backing_key_rotation_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

95
services/rds.py Normal file
View File

@ -0,0 +1,95 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def aurora_last_backup_recovery_point_created():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def aurora_mysql_backtracking_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def db_instance_backup_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def rds_cluster_auto_minor_version_upgrade_enable():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def rds_cluster_default_admin_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def rds_cluster_deletion_protection_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def rds_cluster_encrypted_at_rest():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def rds_cluster_iam_authentication_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def rds_cluster_multi_az_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def rds_db_security_group_not_allowed():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def rds_enhanced_monitoring_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def rds_instance_deletion_protection_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def rds_instance_public_access_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def rds_logging_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def rds_snapshot_encrypted():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

11
services/route53.py Normal file
View File

@ -0,0 +1,11 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def route53_query_logging_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

65
services/s3.py Normal file
View File

@ -0,0 +1,65 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def s3_access_point_in_vpc_only():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def s3_bucket_default_lock_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def s3_bucket_level_public_access_prohibited():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def s3_bucket_logging_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def s3_bucket_ssl_requests_only():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def s3_bucket_versioning_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def s3_default_encryption_kms():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def s3_event_notifications_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def s3_last_backup_recovery_point_created():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def s3_lifecycle_policy_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

23
services/secrets_manager.py Normal file
View File

@ -0,0 +1,23 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def secretsmanager_rotation_enabled_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def secretsmanager_scheduled_rotation_success_check():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def secretsmanager_secret_periodic_rotation():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

11
services/security_hub.py Normal file
View File

@ -0,0 +1,11 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def securityhub_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

17
services/sns.py Normal file
View File

@ -0,0 +1,17 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def sns_encrypted_kms():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def sns_topic_message_delivery_notification_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

11
services/tags.py Normal file
View File

@ -0,0 +1,11 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def required_tags():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

257
services/vpc.py Normal file
View File

@ -0,0 +1,257 @@
from models import RuleCheckResult
from pprint import pprint
import boto3
ec2 = boto3.client("ec2")
def ec2_transit_gateway_auto_vpc_attach_disabled():
response = ec2.describe_transit_gateways()
non_compliant_resources = [
resource["TransitGatewayArn"]
for resource in filter(
lambda x: x["Options"]["AutoAcceptSharedAttachments"] == "enable",
response["TransitGateways"],
)
]
compliant_resources = list(
set([resource["TransitGatewayArn"] for resource in response["TransitGateways"]])
- set(non_compliant_resources)
)
return RuleCheckResult(
passed=not non_compliant_resources,
compliant_resources=compliant_resources,
non_compliant_resources=non_compliant_resources,
)
def restricted_ssh():
response = ec2.describe_security_group_rules()
non_compliant_resources = [
f'{resource["GroupId"]} / {resource["SecurityGroupRuleId"]}'
for resource in filter(
lambda x: x["IsEgress"] == False
and x["FromPort"] <= 22
and x["ToPort"] >= 22
and x.get("CidrIpv4") == "0.0.0.0/0",
response["SecurityGroupRules"],
)
]
compliant_resources = list(
set(
[
f'{resource["GroupId"]} / {resource["SecurityGroupRuleId"]}'
for resource in response["SecurityGroupRules"]
]
)
- set(non_compliant_resources)
)
return RuleCheckResult(
passed=not non_compliant_resources,
compliant_resources=compliant_resources,
non_compliant_resources=non_compliant_resources,
)
def restricted_common_ports():
common_ports = [
22, # SSH
80, # HTTP
3306, # MySQL
3389, # RDP
5432, # PostgreSQL
6379, # Redis
11211, # Memcached
]
response = ec2.describe_security_group_rules()
non_compliant_resources = [
f'{resource["GroupId"]} / {resource["SecurityGroupRuleId"]}'
for resource in filter(
lambda x: x["IsEgress"] == False
and x["FromPort"] in common_ports
and x["ToPort"] in common_ports
and x.get("PrefixListId") is None,
response["SecurityGroupRules"],
)
]
compliant_resources = list(
set(
f'{resource["GroupId"]} / {resource["SecurityGroupRuleId"]}'
for resource in response["SecurityGroupRules"]
)
- set(non_compliant_resources)
)
return RuleCheckResult(
passed=not non_compliant_resources,
compliant_resources=compliant_resources,
non_compliant_resources=non_compliant_resources,
)
def subnet_auto_assign_public_ip_disabled():
response = ec2.describe_subnets()
non_compliant_resources = [
resource["SubnetId"]
for resource in filter(lambda x: x["MapPublicIpOnLaunch"], response["Subnets"])
]
compliant_resources = list(
set(resource["SubnetId"] for resource in response["Subnets"])
- set(non_compliant_resources)
)
return RuleCheckResult(
passed=not non_compliant_resources,
compliant_resources=compliant_resources,
non_compliant_resources=non_compliant_resources,
)
def vpc_default_security_group_closed():
response = ec2.describe_security_groups(
Filters=[{"Name": "group-name", "Values": ["default"]}]
)
non_compliant_resources = [
resource["GroupId"]
for resource in filter(
lambda x: x["IpPermissions"] or x["IpPermissionsEgress"],
response["SecurityGroups"],
)
]
compliant_resources = list(
set(resource["GroupId"] for resource in response["SecurityGroups"])
- set(non_compliant_resources)
)
return RuleCheckResult(
passed=not non_compliant_resources,
compliant_resources=compliant_resources,
non_compliant_resources=non_compliant_resources,
)
def vpc_flow_logs_enabled():
response = ec2.describe_flow_logs()
flow_log_enabled_vpcs = [
resource["ResourceId"] for resource in response["FlowLogs"]
]
response = ec2.describe_vpcs()
non_compliant_resources = [
resource["VpcId"]
for resource in filter(
lambda x: x["VpcId"] not in flow_log_enabled_vpcs, response["Vpcs"]
)
]
compliant_resources = list(
set(resource["VpcId"] for resource in response["Vpcs"])
- set(non_compliant_resources)
)
return RuleCheckResult(
passed=not non_compliant_resources,
compliant_resources=compliant_resources,
non_compliant_resources=non_compliant_resources,
)
def vpc_network_acl_unused_check():
response = ec2.describe_network_acls()
non_compliant_resources = [
resource["NetworkAclId"]
for resource in filter(lambda x: not x["Associations"], response["NetworkAcls"])
]
compliant_resources = list(
set(resource["NetworkAclId"] for resource in response["NetworkAcls"])
- set(non_compliant_resources)
)
return RuleCheckResult(
passed=not non_compliant_resources,
compliant_resources=compliant_resources,
non_compliant_resources=non_compliant_resources,
)
def vpc_peering_dns_resolution_check():
response = ec2.describe_vpc_peering_connections()
non_compliant_resources = [
resource["VpcPeeringConnectionId"]
for resource in filter(
lambda x: x["Status"]["Code"] not in ["deleted", "deleting"]
and (
not x["AccepterVpcInfo"].get("PeeringOptions")
or not x["AccepterVpcInfo"]["PeeringOptions"][
"AllowDnsResolutionFromRemoteVpc"
]
or not x["RequesterVpcInfo"]["PeeringOptions"][
"AllowDnsResolutionFromRemoteVpc"
]
),
response["VpcPeeringConnections"],
)
]
compliant_resources = list(
set(
resource["VpcPeeringConnectionId"]
for resource in response["VpcPeeringConnections"]
)
- set(non_compliant_resources)
)
return RuleCheckResult(
passed=not non_compliant_resources,
compliant_resources=compliant_resources,
non_compliant_resources=non_compliant_resources,
)
def vpc_sg_open_only_to_authorized_ports():
response = ec2.describe_security_group_rules()
authorized_port = [
# 80
]
non_compliant_resources = [
f'{resource["GroupId"]} / {resource["SecurityGroupRuleId"]}'
for resource in filter(
lambda x: x["IsEgress"] == False
and (x.get("CidrIpv4") == "0.0.0.0/0" or x.get("CidrIpv6") == "::/0")
and x["FromPort"] not in authorized_port
and x["ToPort"] not in authorized_port,
response["SecurityGroupRules"],
)
]
compliant_resources = list(
set(
f'{resource["GroupId"]} / {resource["SecurityGroupRuleId"]}'
for resource in response["SecurityGroupRules"]
)
- set(non_compliant_resources)
)
return RuleCheckResult(
passed=not non_compliant_resources,
compliant_resources=compliant_resources,
non_compliant_resources=non_compliant_resources,
)

29
services/wafv2.py Normal file
View File

@ -0,0 +1,29 @@
from models import RuleCheckResult
import boto3
# client = boto3.client("")
def wafv2_logging_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def wafv2_rulegroup_logging_enabled():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def wafv2_rulegroup_not_empty():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
def wafv2_webacl_not_empty():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)

68
utils.py Normal file
View File

@ -0,0 +1,68 @@
import json
import shutil
def load_bp_from_file(filepath="bp.json"):
try:
with open(filepath, "r") as f:
content = "".join(f.readlines())
except FileNotFoundError:
shutil.copy("bp-base.json", filepath)
with open(filepath, "r") as f:
content = "".join(f.readlines())
return json.loads(content)
def save_bp_to_file(bp, filepath="bp.json"):
with open(filepath, "w") as f:
f.write(json.dumps(bp, indent=2))
def convert_snake_case(text):
return text.lower().replace(" ", "_").replace("-", "_")
def convert_bp_to_snake_case(bp):
bp = {
service_name.lower().replace(" ", "_"): value
for service_name, value in bp.items()
}
for v in bp.values():
v["rules"] = {
rule_name.lower().replace("-", "_"): rule
for rule_name, rule in v["rules"].items()
}
return bp
if __name__ == "__main__":
bp = load_bp_from_file()
rules = [
(
k.lower().replace(" ", "_"),
list(map(lambda x: x.replace("-", "_"), v["rules"].keys())),
)
for k, v in bp.items()
]
print(json.dumps(rules, indent=2))
for rule in rules:
file_name = rule[0]
rule_names = rule[1]
file_template = f"""from models import RuleCheckResult
import boto3
# client = boto3.client("")
"""
with open(f"services/{file_name}.py", "w") as f:
f.write(file_template)
for rule_name in rule_names:
function_template = f"""
def {rule_name}():
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
)
"""
f.write(function_template)