Merge pull request #498 from 0xdabbad00/find_other_privs

Find other privs

README.md

@@ -14,7 +14,7 @@ CloudMapper helps you analyze your Amazon Web Services (AWS) environments.  The
 - `api_endpoints`: List the URLs that can be called via API Gateway.
 - `audit`: Check for potential misconfigurations.
 - `collect`: Collect metadata about an account. More details [here](https://summitroute.com/blog/2018/06/05/cloudmapper_collect/).
-- `find_admins`: Look at IAM policies to identify admin users and roles and spot potential IAM issues. More details [here](https://summitroute.com/blog/2018/06/12/cloudmapper_find_admins/).
+- `find_admins`: Look at IAM policies to identify admin users and roles, or principals with specific privileges. More details [here](https://summitroute.com/blog/2018/06/12/cloudmapper_find_admins/).
 - `find_unused`: Look for unused resources in the account.  Finds unused Security Groups, Elastic IPs, network interfaces, and volumes.
 - `prepare`/`webserver`: See [Network Visualizations](docs/network_visualizations.md)
 - `public`: Find public hosts and port ranges. More details [here](https://summitroute.com/blog/2018/06/13/cloudmapper_public/).

cloudmapper.py

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

commands/audit.py

@@ -24,7 +24,7 @@ def audit_command(accounts, config, args):
         if args.json:
             finding = json.loads(str(finding))
             finding['finding_type_metadata']= conf
-            print(json.dumps(finding))
+            print(json.dumps(finding, sort_keys=True))
         elif args.markdown:
             print(
                 "*Audit Finding: [{}] - {}*\\nAccount: {} ({}) - {}\\nDescription: {}\\nResource: `{}`\\nDetails:```{}```".format(

commands/find_admins.py

@@ -1,12 +1,37 @@
+import argparse
+import json
+
 from shared.common import parse_arguments
 from shared.iam_audit import find_admins
 
-__description__ = "Find admins in accounts"
+__description__ = "Find privileged users and roles in accounts"
 
 
 def run(arguments):
-    _, accounts, _ = parse_arguments(arguments)
-    admins = find_admins(accounts, findings=set())
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--privs",
+        help="The privileges to look for. Defaults to iam privileges that allow direct admin or escalation.",
+        default=None,
+    )
+    parser.add_argument(
+        "--json",
+        help="Print the json of the issues",
+        default=False,
+        action="store_true",
+    )
+    parser.add_argument(
+        "--include_restricted",
+        help="Include references to these privs that may be restricted by resources or conditions",
+        default=False,
+        action="store_true",
+    )
+    args, accounts, config = parse_arguments(arguments, parser)
+
+    admins = find_admins(accounts, args, findings=set())
 
     for admin in admins:
-        print("{}\t{}\t{}".format(admin["account"], admin["type"], admin["name"]))
+        if args.json:
+            print(json.dumps(admin, sort_keys=True))
+        else:
+            print("{}\t{}\t{}".format(admin["account"], admin["type"], admin["name"]))

shared/iam_audit.py

@@ -49,7 +49,9 @@ def policy_action_count(policy_doc, location):
     return actions_count
 
 
-def is_admin_policy(policy_doc, location, findings, region):
+def is_admin_policy(
+    policy_doc, location, findings, region, privs_to_look_for, include_retricted
+):
     # This attempts to identify policies that directly allow admin privs, or indirectly through possible
     # privilege escalation (ex. iam:PutRolePolicy to add an admin policy to itself).
     # It is a best effort. It will have false negatives, meaning it may not identify an admin policy
@@ -94,32 +96,14 @@ def is_admin_policy(policy_doc, location, findings, region):
                         )
                     return True
                 # Look for privilege escalations
-                if (
-                    stmt.get("Resource", "") == "*"
-                    and stmt.get("Condition", "") == ""
-                    and (
-                        action_matches(
-                            action,
-                            [
-                                "iam:PutRolePolicy",
-                                "iam:AddUserToGroup",
-                                "iam:AddRoleToInstanceProfile",
-                                "iam:AttachGroupPolicy",
-                                "iam:AttachRolePolicy",
-                                "iam:AttachUserPolicy",
-                                "iam:ChangePassword",
-                                "iam:CreateAccessKey",
-                                # Check for the rare possibility that an actor has a Deny policy on themselves,
-                                # so they try to escalate privs by removing that policy
-                                "iam:DeleteUserPolicy",
-                                "iam:DetachGroupPolicy",
-                                "iam:DetachRolePolicy",
-                                "iam:DetachUserPolicy",
-                            ],
-                        )
-                    )
-                ):
-                    return True
+                if action_matches(action, privs_to_look_for):
+                    if include_retricted:
+                        return True
+                    elif (
+                        stmt.get("Resource", "") == "*"
+                        and stmt.get("Condition", "") == ""
+                    ):
+                        return True
 
     return False
 
@@ -151,17 +135,46 @@ def check_for_bad_policy(findings, region, arn, policy_text):
                 return
 
 
-def find_admins(accounts, findings):
+def find_admins(accounts, args, findings):
+    privs_to_look_for = None
+    if "privs" in args and args.privs:
+        privs_to_look_for = args.privs.split(",")
+    include_restricted = False
+    if "include_restricted" in args:
+        include_restricted = args.include_restricted
+
     admins = []
     for account in accounts:
         account = Account(None, account)
         region = get_us_east_1(account)
-        admins.extend(find_admins_in_account(region, findings))
+        admins.extend(
+            find_admins_in_account(
+                region, findings, privs_to_look_for, include_restricted
+            )
+        )
 
     return admins
 
 
-def find_admins_in_account(region, findings):
+def find_admins_in_account(
+    region, findings, privs_to_look_for=None, include_restricted=False
+):
+    if privs_to_look_for is None:
+        privs_to_look_for = [
+            "iam:PutRolePolicy",
+            "iam:AddUserToGroup",
+            "iam:AddRoleToInstanceProfile",
+            "iam:AttachGroupPolicy",
+            "iam:AttachRolePolicy",
+            "iam:AttachUserPolicy",
+            "iam:ChangePassword",
+            "iam:CreateAccessKey",
+            "iam:DeleteUserPolicy",
+            "iam:DetachGroupPolicy",
+            "iam:DetachRolePolicy",
+            "iam:DetachUserPolicy",
+        ]
+
     account = region.account
     location = {"account": account.name}
 
@@ -185,7 +198,14 @@ def find_admins_in_account(region, findings):
 
         policy_action_counts[policy["Arn"]] = policy_action_count(policy_doc, location)
 
-        if is_admin_policy(policy_doc, location, findings, region):
+        if is_admin_policy(
+            policy_doc,
+            location,
+            findings,
+            region,
+            privs_to_look_for,
+            include_restricted,
+        ):
             admin_policies.append(policy["Arn"])
             if (
                 "arn:aws:iam::aws:policy/AdministratorAccess" in policy["Arn"]
@@ -240,7 +260,14 @@ def find_admins_in_account(region, findings):
 
         for policy in role["RolePolicyList"]:
             policy_doc = policy["PolicyDocument"]
-            if is_admin_policy(policy_doc, location, findings, region):
+            if is_admin_policy(
+                policy_doc,
+                location,
+                findings,
+                region,
+                privs_to_look_for,
+                include_restricted,
+            ):
                 reasons.append("Custom policy: {}".format(policy["PolicyName"]))
                 findings.add(
                     Finding(
@@ -348,7 +375,14 @@ def find_admins_in_account(region, findings):
                 )
         for policy in group["GroupPolicyList"]:
             policy_doc = policy["PolicyDocument"]
-            if is_admin_policy(policy_doc, location, findings, region):
+            if is_admin_policy(
+                policy_doc,
+                location,
+                findings,
+                region,
+                privs_to_look_for,
+                include_restricted,
+            ):
                 is_admin = True
                 findings.add(
                     Finding(
@@ -390,7 +424,14 @@ def find_admins_in_account(region, findings):
                 )
         for policy in user.get("UserPolicyList", []):
             policy_doc = policy["PolicyDocument"]
-            if is_admin_policy(policy_doc, location, findings, region):
+            if is_admin_policy(
+                policy_doc,
+                location,
+                findings,
+                region,
+                privs_to_look_for,
+                include_restricted,
+            ):
                 reasons.append("Custom user policy: {}".format(policy["PolicyName"]))
                 findings.add(
                     Finding(

tests/unit/test_iam_audit.py

@@ -13,6 +13,6 @@ def test_find_admins(self):
         )
 
         findings = set()
-        admins = find_admins(accounts, findings)
+        admins = find_admins(accounts, args, findings)
 
         assert_equal(admins, [{"account": "demo", "name": "bad_role", "type": "role"}])