Merge pull request #222 from duo-labs/update

Update

Author: 0xdabbad00

.coveragerc

@@ -0,0 +1,6 @@
+[run]
+source = parliament
+omit = parliament/cli.py
+
+[report]
+fail_under = 75

README.md

@@ -217,13 +217,9 @@ ignore_locations:
 To create unit tests for our new private auditor, create the directory `./parliament/private_auditors/tests/` and create the file `test_sensitive_bucket_access.py` there with the contents:
 
 ```
-import unittest
-from nose.tools import raises, assert_equal
-
-# import parliament
 from parliament import analyze_policy_string
 
-class TestCustom(unittest.TestCase):
+class TestCustom():
     """Test class for custom auditor"""
 
     def test_my_auditor(self):

parliament/__init__.py

@@ -1,7 +1,7 @@
 """
 This library is a linter for AWS IAM policies.
 """
-__version__ = "1.5.2"
+__version__ = "1.6.0"
 
 import fnmatch
 import functools
@@ -62,7 +62,11 @@ def analyze_policy_string(
         policy_json = jsoncfg.loads_config(policy_str)
     except jsoncfg.parser.JSONConfigParserException as e:
         policy = Policy(None)
-        policy.add_finding("MALFORMED_JSON", detail="json parsing error: {}".format(e), location={'line': e.line, 'column': e.column})
+        policy.add_finding(
+            "MALFORMED_JSON",
+            detail="json parsing error: {}".format(e),
+            location={"line": e.line, "column": e.column},
+        )
         return policy
 
     policy = Policy(policy_json, filepath, config)
@@ -97,7 +101,7 @@ def is_arn_match(resource_type, arn_format, resource):
     - arn_format: ARN regex from the docs
     - resource: ARN regex from IAM policy
 
-    
+
     We can cheat some because after the first sections of the arn match, meaning until the 5th colon (with some
     rules there to allow empty or asterisk sections), we only need to match the ID part.
     So the above is simplified to "*/*" and "*personalize*".
@@ -147,6 +151,7 @@ def is_arn_match(resource_type, arn_format, resource):
     resource_id = re.sub(r"\[.+?\]", "*", resource_id)
     return is_glob_match(arn_id, resource_id)
 
+
 def is_arn_strictly_valid(resource_type, arn_format, resource):
     """
     Strictly validate the arn_format specified in the docs, with the resource
@@ -179,7 +184,7 @@ def is_arn_strictly_valid(resource_type, arn_format, resource):
         arn_id_resource_type = re.match(r"(^[^\*][\w-]+)[\/\:].+", arn_id)
 
         if arn_id_resource_type != None and resource_id != "*":
-            
+
             # https://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html#genref-aws-service-namesspaces
             # The following is not allowed: arn:aws:iam::123456789012:u*
             if not (resource_id.startswith(arn_id_resource_type[1])):
@@ -193,9 +198,11 @@ def is_arn_strictly_valid(resource_type, arn_format, resource):
         return True
     return False
 
+
 def strip_var_from_arn(arn, replace_with=""):
     return re.sub(r"\$\{aws.[\w\/]+\}", replace_with, arn)
 
+
 def is_glob_match(s1, s2):
     # This comes from https://github.com/duo-labs/parliament/issues/36#issuecomment-574001764
 
@@ -275,20 +282,20 @@ def expand_action(action, raise_exceptions=True):
 
 
 def get_resource_type_matches_from_arn(arn):
-    """ Given an ARN such as "arn:aws:s3:::mybucket", find resource types that match it.
-        This would return:
-        [
-            "resource": {
-                "arn": "arn:${Partition}:s3:::${BucketName}",
-                "condition_keys": [],
-                "resource": "bucket"
-            },
-            "service": {
-                "service_name": "Amazon S3",
-                "privileges": [...]
-                ...
-            }
-        ]
+    """Given an ARN such as "arn:aws:s3:::mybucket", find resource types that match it.
+    This would return:
+    [
+        "resource": {
+            "arn": "arn:${Partition}:s3:::${BucketName}",
+            "condition_keys": [],
+            "resource": "bucket"
+        },
+        "service": {
+            "service_name": "Amazon S3",
+            "privileges": [...]
+            ...
+        }
+    ]
     """
     matches = []
     for service in iam_definition:
@@ -300,8 +307,7 @@ def get_resource_type_matches_from_arn(arn):
 
 
 def get_privilege_matches_for_resource_type(resource_type_matches):
-    """ Given the response from get_resource_type_matches_from_arn(...), this will identify the relevant privileges.
-    """
+    """Given the response from get_resource_type_matches_from_arn(...), this will identify the relevant privileges."""
     privilege_matches = []
     for match in resource_type_matches:
         for privilege in match["service"]["privileges"]:

parliament/cli.py

@@ -137,10 +137,12 @@ def main():
         help='Provide a string such as \'{"Version": "2012-10-17","Statement": {"Effect": "Allow","Action": ["s3:GetObject", "s3:PutBucketPolicy"],"Resource": ["arn:aws:s3:::bucket1", "arn:aws:s3:::bucket2/*"]}}\'',
         type=str,
     )
-    parser.add_argument('--file',
-                            help="Provide a policy via stdin (e.g. through piping) or --file",
-                            type=argparse.FileType('r'),
-                            default=sys.stdin)
+    parser.add_argument(
+        "--file",
+        help="Provide a policy via stdin (e.g. through piping) or --file",
+        type=argparse.FileType("r"),
+        default=sys.stdin,
+    )
     parser.add_argument(
         "--directory", help="Provide a path to directory with policy files", type=str
     )
@@ -236,9 +238,7 @@ def main():
             with open(file_path) as f:
                 contents = f.read()
                 policy_file_json = jsoncfg.loads_config(contents)
-                policy_string = json.dumps(
-                    policy_file_json.PolicyVersion.Document()
-                )
+                policy_string = json.dumps(policy_file_json.PolicyVersion.Document())
                 policy = analyze_policy_string(
                     policy_string,
                     file_path,
@@ -271,15 +271,16 @@ def main():
                     if not version.IsDefaultVersion():
                         continue
                     policy = analyze_policy_string(
-                        json.dumps(version.Document()), policy.Arn(),
+                        json.dumps(version.Document()),
+                        policy.Arn(),
                     )
                     findings.extend(policy.findings)
 
             # Review the inline policies on Users, Roles, and Groups
             for user in auth_details_json.UserDetailList:
                 for policy in user.UserPolicyList([]):
                     policy = analyze_policy_string(
-                        json.dumps(policy['PolicyDocument']),
+                        json.dumps(policy["PolicyDocument"]),
                         user.Arn(),
                         private_auditors_custom_path=args.private_auditors,
                         include_community_auditors=args.include_community_auditors,
@@ -289,7 +290,7 @@ def main():
             for role in auth_details_json.RoleDetailList:
                 for policy in role.RolePolicyList([]):
                     policy = analyze_policy_string(
-                        json.dumps(policy['PolicyDocument']),
+                        json.dumps(policy["PolicyDocument"]),
                         role.Arn(),
                         private_auditors_custom_path=args.private_auditors,
                         include_community_auditors=args.include_community_auditors,
@@ -299,7 +300,7 @@ def main():
             for group in auth_details_json.GroupDetailList:
                 for policy in group.GroupPolicyList([]):
                     policy = analyze_policy_string(
-                        json.dumps(policy['PolicyDocument']),
+                        json.dumps(policy["PolicyDocument"]),
                         group.Arn(),
                         private_auditors_custom_path=args.private_auditors,
                         include_community_auditors=args.include_community_auditors,

parliament/community_auditors/single_value_condition_too_permissive.py

@@ -7,6 +7,7 @@
 from parliament import Policy
 from parliament.misc import make_list
 
+
 def audit(policy: Policy) -> None:
     global_single_valued_condition_keys = [
         "aws:CalledViaFirst",
@@ -52,9 +53,12 @@ def audit(policy: Policy) -> None:
             operator = condition[0]
             condition_block = condition[1]
             if re.match(r"^For(All|Any)Values:", operator):
-                keys = list(k for k,_v in condition_block)
-                if any(any(re.match(k, key) for k in global_single_valued_condition_keys) for key in keys):
+                keys = list(k for k, _v in condition_block)
+                if any(
+                    any(re.match(k, key) for k in global_single_valued_condition_keys)
+                    for key in keys
+                ):
                     policy.add_finding(
                         "SINGLE_VALUE_CONDITION_TOO_PERMISSIVE",
-                        detail='Checking a single value conditional key against a set of values results in overly permissive policies.',
+                        detail="Checking a single value conditional key against a set of values results in overly permissive policies.",
                     )

parliament/community_auditors/tests/test_advanced_policy_elements.py

@@ -1,13 +1,9 @@
-import unittest
-
-from nose.tools import assert_equal
-
 from parliament import analyze_policy_string
 
 S3_STAR_FINDINGS = {"PERMISSIONS_MANAGEMENT_ACTIONS", "RESOURCE_MISMATCH"}
 
 
-class TestAdvancedPolicyElements(unittest.TestCase):
+class TestAdvancedPolicyElements:
     def test_notresource_allow(self):
         # NotResource is OK with Effect: Deny. This denies access to
         # all S3 buckets except Payroll buckets. This example is taken from

parliament/community_auditors/tests/test_credentials_exposure.py

@@ -1,12 +1,7 @@
-import unittest
-
-from nose.tools import assert_equal
-
-# import parliament
 from parliament import analyze_policy_string
 
 
-class TestCredentialsManagement(unittest.TestCase):
+class TestCredentialsManagement:
     """Test class for Credentials Management auditor"""
 
     def test_credentials_management(self):

parliament/community_auditors/tests/test_permissions_management.py

@@ -1,12 +1,7 @@
-import unittest
-
-from nose.tools import assert_equal
-
-# import parliament
 from parliament import analyze_policy_string
 
 
-class TestPermissionsManagement(unittest.TestCase):
+class TestPermissionsManagement:
     """Test class for Permissions Management auditor"""
 
     def test_permissions_management(self):

parliament/community_auditors/tests/test_privilege_escalation.py

@@ -1,12 +1,7 @@
-import unittest
-
-from nose.tools import assert_equal
-
-# import parliament
 from parliament import analyze_policy_string
 
 
-class TestPrivilegeEscalation(unittest.TestCase):
+class TestPrivilegeEscalation:
     """Test class for Privilege Escalation auditor"""
 
     def test_privilege_escalation(self):

parliament/community_auditors/tests/test_sensitive_access.py

@@ -1,11 +1,7 @@
-import unittest
-
-from nose.tools import assert_equal
-
 from parliament import analyze_policy_string
 
 
-class TestSensitiveAccess(unittest.TestCase):
+class TestSensitiveAccess:
     """Test class for Sensitive access auditor"""
 
     def test_sensitive_access(self):