Fix bad casing in buckets rule. Add unit tests. (#701)

* fix casing in rule

* tweak

google/cloud/security/scanner/audit/buckets_rules_engine.py

@@ -73,9 +73,9 @@ def find_policy_violations(self, buckets_acls,
         resource_rules = self.rule_book.get_resource_rules()
 
         for rule in resource_rules:
-            violations = itertools.chain(violations,
-                                         rule.\
-                                         find_policy_violations(buckets_acls))
+            violations = itertools.chain(
+                violations,
+                rule.find_policy_violations(buckets_acls))
         return violations
 
     def add_rules(self, rules):

rules/bucket_rules.yaml

@@ -15,7 +15,7 @@
 rules:
   - name: sample bucket acls rule to search for public buckets
     bucket: '*'
-    entity: AllUsers
+    entity: allUsers
     email: '*'
     domain: '*'
     role: '*'
@@ -24,7 +24,7 @@ rules:
           - YOUR_ORG_ID / YOUR_PROJECT_ID
   - name: sample bucket acls rule to search for exposed buckets
     bucket: '*'
-    entity: AllAuthenticatedUsers
+    entity: allAuthenticatedUsers
     email: '*'
     domain: '*'
     role: '*' 

tests/scanner/audit/buckets_rules_engine_test.py

@@ -21,6 +21,7 @@
 import yaml
 
 from tests.unittest_utils import ForsetiTestCase
+from google.cloud.security.common.gcp_type import bucket_access_controls
 from google.cloud.security.common.util import file_loader
 from google.cloud.security.scanner.audit.errors import InvalidRulesSchemaError
 from google.cloud.security.scanner.audit import base_rules_engine as bre
@@ -86,6 +87,46 @@ def test_build_rule_book_no_resource_type_fails(self):
         with self.assertRaises(InvalidRulesSchemaError):
             rules_engine.build_rule_book()
 
+    def test_find_violation_for_publicly_exposed_acls(self):
+
+        rules_local_path = get_datafile_path(__file__,
+            'buckets_test_rules_1.yaml')
+        rules_engine = bre.BucketsRulesEngine(rules_file_path=rules_local_path)
+        rules_engine.build_rule_book()
+        rules_map = rules_engine.rule_book.resource_rules_map
+        allUsers_rule = rules_map[0]
+        allAuthenticatedUsers_rule = rules_map[1]
+
+        # Everything is allowed.
+        acl = bucket_access_controls.BucketAccessControls(
+            '*', '*', '*', '*', '*', '111111')
+        violation = allUsers_rule.find_policy_violations(acl)
+        self.assertEquals(0, len(list(violation)))
+
+        # Exposed to everyone in the world.
+        acl = bucket_access_controls.BucketAccessControls(
+            '*', 'allUsers', '*', '*', '*', '111111')
+        violation = allUsers_rule.find_policy_violations(acl)
+        self.assertEquals(1, len(list(violation)))
+
+        # Test case sensitivity.
+        acl = bucket_access_controls.BucketAccessControls(
+            '*', 'AllUsers', '*', '*', '*', '111111')
+        violation = allUsers_rule.find_policy_violations(acl)
+        self.assertEquals(0, len(list(violation)))
+
+        # Exposed to all google-authenticated users in the world.
+        acl = bucket_access_controls.BucketAccessControls(
+            '*', 'allAuthenticatedUsers', '*', '*', '*', '111111')
+        violation = allAuthenticatedUsers_rule.find_policy_violations(acl)
+        self.assertEquals(1, len(list(violation)))
+
+        # Test case sensitivity.
+        acl = bucket_access_controls.BucketAccessControls(
+            '*', 'AllAuthenticatedUsers', '*', '*', '*', '111111')
+        violation = allAuthenticatedUsers_rule.find_policy_violations(acl)
+        self.assertEquals(0, len(list(violation)))
+
 
 if __name__ == '__main__':
     unittest.main()

tests/scanner/audit/data/buckets_test_rules_1.yaml

@@ -1,7 +1,7 @@
 rules:
   - name: sample bucket acls rule to search for public buckets
     bucket: '*'
-    entity: AllUsers
+    entity: allUsers
     email: '*'
     domain: '*'
     role: '*'