Prevent S3 lockouts, tests for S3:* DENY resource policies.
ADD
checkov/terraform/checks/resource/aws/S3ProtectAgainstPolicyLockout.py
+from checkov.common.models.enums import CheckResult, CheckCategories
+from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
+import json
+
+
+class S3ProtectAgainstPolicyLockout(BaseResourceCheck):
+
+ def __init__(self):
+ name = "Ensure S3 bucket policy does not lockout all but root user. (Prevent lockouts needing root account fixes)"
+ id = "CKV_AWS_89"
+ supported_resources = ['aws_s3_bucket', 'aws_s3_bucket_policy']
+ categories = [CheckCategories.IAM]
+ super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
+
+ def scan_resource_conf(self, conf):
+ if 'policy' in conf.keys():
+ if isinstance(conf['policy'][0], str):
+ try:
+ policy_block = json.loads(conf['policy'][0])
+ if 'Statement' in policy_block.keys():
+ for statement in policy_block['Statement']:
+ if statement['Effect'] == 'Allow':
+ continue
+
+ principal = statement['Principal']
+
+ if principal == '*':
+ return CheckResult.FAILED
+ elif 'AWS' in statement['Principal']:
+ # Can be a string or an array of strings
+ aws = statement['Principal']['AWS']
+ if (type(aws) == str and aws == '*') or (type(aws) == list and '*' in aws):
+ return CheckResult.FAILED
+
+ action = statement['Action']
+
+ if action == '*':
+ return CheckResult.FAILED
+ elif 's3' in statement['Action']:
+ # Can be a string or an array of strings
+ s3 = statement['Action']['s3']
+ if (type(s3) == str and s3 == '*') or (type(s3) == list and '*' in s3):
+ return CheckResult.FAILED
+ except: # nosec
+ pass
+ return CheckResult.PASSED
+
+check = S3ProtectAgainstPolicyLockout()

ADD
tests/terraform/checks/resource/aws/test_S3ProtectAgainstPolicyLockout.py
+import unittest
+import hcl2
+
+from checkov.terraform.checks.resource.aws.S3ProtectAgainstPolicyLockout import check
+from checkov.common.models.enums import CheckResult
+
+
+class TestS3ProtectAgainstPolicyLockout(unittest.TestCase):
+
+ def test_failure(self):
+ hcl_res = hcl2.loads("""
+ resource "aws_s3_bucket" "s3" {
+ bucket = "bucket"
+
+ policy = <<POLICY
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Principal": {
+ "AWS": [
+ "*"
+ ]
+ },
+ "Effect": "Deny",
+ "Action": [
+ "s3:*"
+ ],
+ "Resource": [
+ "*"
+ ]
+ }
+ ]
+ }
+ POLICY
+ }
+ """)
+ resource_conf = hcl_res['resource'][0]['aws_s3_bucket']['s3']
+ scan_result = check.scan_resource_conf(conf=resource_conf)
+ self.assertEqual(CheckResult.FAILED, scan_result)
+
+ def test_failure_2(self):
+ hcl_res = hcl2.loads("""
+ resource "aws_s3_bucket_policy" "s3" {
+ bucket = "bucket"
+
+ policy = <<POLICY
+ {
+ "Version": "2012-10-17",
+ "Statement": [{
+ "Principal": {
+ "AWS": [
+ "*"
+ ]
+ },
+ "Effect": "Deny",
+ "Action": "s3:*",
+ "Resource": [
+ "*"
+ ]
+ }]
+ }
+ POLICY
+ }
+ """)
+ resource_conf = hcl_res['resource'][0]['aws_s3_bucket_policy']['s3']
+ scan_result = check.scan_resource_conf(conf=resource_conf)
+ self.assertEqual(CheckResult.FAILED, scan_result)
+
+ def test_failure_3(self):
+ hcl_res = hcl2.loads("""
+ resource "aws_s3_bucket" "s3" {
+ bucket = "bucket"
+
+ policy = <<POLICY
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Principal": "*",
+ "Effect": "Deny",
+ "Action": "s3:*"
+ }
+ ]
+ }
+ POLICY
+ }
+ """)
+ resource_conf = hcl_res['resource'][0]['aws_s3_bucket']['s3']
+ scan_result = check.scan_resource_conf(conf=resource_conf)
+ self.assertEqual(CheckResult.FAILED, scan_result)
+
+ def test_success_policyobj(self):
+ hcl_res = hcl2.loads("""
+ resource "aws_s3_bucket_policy" "s3" {
+ bucket = "bucket"
+
+ policy = <<POLICY
+ {
+ "Id": "Policy1597273448050",
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Sid": "Stmt1597273446725",
+ "Action": [
+ "s3:GetObject"
+ ],
+ "Effect": "Deny",
+ "Resource": "arn:aws:s3:::bucket/*",
+ "Principal": {
+ "AWS": "some_arn"
+ }
+ }
+ ]
+ }
+ POLICY
+ }
+ """)
+ resource_conf = hcl_res['resource'][0]['aws_s3_bucket_policy']['s3']
+ scan_result = check.scan_resource_conf(conf=resource_conf)
+ self.assertEqual(CheckResult.PASSED, scan_result)
+
+
+if __name__ == '__main__':
+ unittest.main()

Prevent S3 lockouts, tests for S3:* DENY resource policies.