Merge pull request #601 from 0xdabbad00/parliament_checks

Added parliament support

Pipfile

@@ -15,6 +15,7 @@ matplotlib = "==2.2.2"
 policyuniverse = "==1.1.0.1"
 PyYAML = "==4.2b4"
 Jinja2 = "==2.10.1"
+parliament = "==0.2.3"
 
 [dev-packages]
 autoflake = "==0.7"

audit_config.yaml

@@ -204,6 +204,12 @@ IAM_UNEXPECTED_S3_EXFIL_PRINCIPAL:
   is_global: True
   group: IAM
 
+IAM_LINTER:
+  title: IAM linting issues
+  description: Issues identified by the IAM linter Parliament
+  severity: Low
+  is_global: True
+  group: IAM
 
 IAM_NAME_DOES_NOT_INDICATE_ADMIN:
   title: Name does not indicate admin

cloudmapper.py

@@ -30,7 +30,7 @@
 import pkgutil
 import importlib
 
-__version__ = "2.7.2"
+__version__ = "2.8.0"
 
 
 def show_help(commands):

shared/audit.py

@@ -206,7 +206,9 @@ def audit_guardduty(findings, region):
 
 
 def audit_iam(findings, region):
+    # By calling the code to find the admins, we'll excercise the code that finds problems.
     find_admins_in_account(region, findings)
+
     # By default we get the findings for the admins, but we can also look for specific
     # privileges, so we'll look for who has s3:ListAllMyBuckets and then only use those
     # findings that are for a compute resource having this privilege

shared/iam_audit.py

@@ -6,12 +6,14 @@
 import os.path
 
 from policyuniverse.policy import Policy
+from parliament import analyze_policy_string
 
 from netaddr import IPNetwork
 from shared.common import Finding, make_list, get_us_east_1
 from shared.query import query_aws, get_parameter_file
 from shared.nodes import Account, Region
 
+
 KNOWN_BAD_POLICIES = {
     "arn:aws:iam::aws:policy/service-role/AmazonEC2RoleforSSM": "Use AmazonSSMManagedInstanceCore instead and add privs as needed",
     "arn:aws:iam::aws:policy/service-role/AmazonMachineLearningRoleforRedshiftDataSource": "Use AmazonMachineLearningRoleforRedshiftDataSourceV2 instead",
@@ -196,6 +198,17 @@ def find_admins_in_account(
 
         check_for_bad_policy(findings, region, policy["Arn"], policy_doc)
 
+        analyzed_policy = analyze_policy_string(json.dumps(policy_doc))
+        for f in analyzed_policy.findings:
+            findings.add(
+                    Finding(
+                        region,
+                        "IAM_LINTER",
+                        policy["Arn"],
+                        resource_details={"issue": str(f.issue), "severity": str(f.severity), "location": str(f.location), "policy": policy_doc},
+                    )
+                )
+
         policy_action_counts[policy["Arn"]] = policy_action_count(policy_doc, location)
 
         if is_admin_policy(
@@ -260,6 +273,18 @@ def find_admins_in_account(
 
         for policy in role["RolePolicyList"]:
             policy_doc = policy["PolicyDocument"]
+
+            analyzed_policy = analyze_policy_string(json.dumps(policy_doc))
+            for f in analyzed_policy.findings:
+                findings.add(
+                        Finding(
+                            region,
+                            "IAM_LINTER",
+                            policy["Arn"],
+                            resource_details={"issue": str(f.issue), "severity": str(f.severity), "location": str(f.location), "policy": policy_doc},
+                        )
+                    )
+
             if is_admin_policy(
                 policy_doc,
                 location,
@@ -430,6 +455,18 @@ def find_admins_in_account(
                 )
         for policy in user.get("UserPolicyList", []):
             policy_doc = policy["PolicyDocument"]
+
+            analyzed_policy = analyze_policy_string(json.dumps(policy_doc))
+            for f in analyzed_policy.findings:
+                findings.add(
+                        Finding(
+                            region,
+                            "IAM_LINTER",
+                            policy["Arn"],
+                            resource_details={"issue": str(f.issue), "severity": str(f.severity), "location": str(f.location), "policy": policy_doc},
+                        )
+                    )
+
             if is_admin_policy(
                 policy_doc,
                 location,

tests/unit/test_audit.py

@@ -44,7 +44,8 @@ def test_audit(self):
                     "IAM_KNOWN_BAD_POLICY",
                     "IAM_ROLE_ALLOWS_ASSUMPTION_FROM_ANYWHERE",
                     "EC2_OLD",
-                    "IAM_UNEXPECTED_S3_EXFIL_PRINCIPAL"
+                    "IAM_UNEXPECTED_S3_EXFIL_PRINCIPAL",
+                    "IAM_LINTER"
                 ]
             ),
         )