feat: export all policies for an IAM principal

README.md

@@ -26,6 +26,7 @@ It now contains much more functionality, including auditing for security issues.
 - `weboftrust`: Show Web Of Trust. More details [here](https://summitroute.com/blog/2018/06/13/cloudmapper_wot/).
 - `report`: Generate HTML report. Includes summary of the accounts and audit findings. More details [here](https://summitroute.com/blog/2019/03/04/cloudmapper_report_generation/).
 - `iam_report`: Generate HTML report for the IAM information of an account. More details [here](https://summitroute.com/blog/2019/03/11/cloudmapper_iam_report_command/).
+- `iam_policies`: Print all collected inline and managed IAM policies attached to a user or role, including policies inherited through groups. Run `python cloudmapper.py iam_policies --accounts <account> --principal <name-or-arn>`.
 
 
 If you want to add your own private commands, you can create a `private_commands` directory and add them there.

commands/iam_policies.py

@@ -0,0 +1,154 @@
+from __future__ import print_function
+
+import argparse
+import json
+from urllib.parse import unquote
+
+__description__ = "Print all collected IAM policies for a principal"
+
+
+def _policy_document(policy):
+    """Return the default-version document from a collected managed policy."""
+    for version in policy.get("PolicyVersionList", []):
+        if version.get("IsDefaultVersion") or version.get("VersionId") == policy.get(
+            "DefaultVersionId"
+        ):
+            document = version.get("Document")
+            if isinstance(document, str):
+                try:
+                    return json.loads(unquote(document))
+                except ValueError:
+                    return document
+            return document
+    return None
+
+
+def _principal_name(principal):
+    return principal.get("UserName") or principal.get("RoleName")
+
+
+def _matches_principal(principal, requested):
+    requested = requested.rstrip("/")
+    candidates = {
+        principal.get("Arn", "").rstrip("/"),
+        principal.get("UserName", ""),
+        principal.get("RoleName", ""),
+    }
+    return requested in candidates
+
+
+def _managed_policy(policy_ref, managed_by_arn, source):
+    policy = managed_by_arn.get(policy_ref.get("PolicyArn"), {})
+    return {
+        "source": source,
+        "name": policy_ref.get("PolicyName") or policy.get("PolicyName"),
+        "arn": policy_ref.get("PolicyArn") or policy.get("Arn"),
+        "document": _policy_document(policy) if policy else None,
+    }
+
+
+def _inline_policies(principal, policy_key, source):
+    policies = []
+    for policy in principal.get(policy_key, []):
+        policies.append(
+            {
+                "source": source,
+                "name": policy.get("PolicyName"),
+                "arn": None,
+                "document": policy.get("PolicyDocument"),
+            }
+        )
+    return policies
+
+
+def build_principal_policy_report(auth_details, requested):
+    """Build a JSON-serializable policy report from IAM auth details.
+
+    The input is the response saved by CloudMapper's collect command. No AWS
+    calls are made here, which keeps the policy attribution logic testable.
+    """
+    managed_by_arn = {
+        policy.get("Arn"): policy
+        for policy in auth_details.get("Policies", [])
+        if policy.get("Arn")
+    }
+    principals = [
+        ("user", principal, "UserPolicyList")
+        for principal in auth_details.get("UserDetailList", [])
+    ] + [
+        ("role", principal, "RolePolicyList")
+        for principal in auth_details.get("RoleDetailList", [])
+    ]
+    matches = [entry for entry in principals if _matches_principal(entry[1], requested)]
+    if not matches:
+        raise ValueError("Principal {!r} was not found in collected IAM data".format(requested))
+
+    groups_by_name = {
+        group.get("GroupName"): group
+        for group in auth_details.get("GroupDetailList", [])
+        if group.get("GroupName")
+    }
+    reports = []
+    for principal_type, principal, inline_key in matches:
+        policies = _inline_policies(principal, inline_key, "inline:principal")
+        for policy_ref in principal.get("AttachedManagedPolicies", []):
+            policies.append(_managed_policy(policy_ref, managed_by_arn, "managed:principal"))
+
+        for group_name in principal.get("GroupList", []):
+            group = groups_by_name.get(group_name)
+            if not group:
+                continue
+            policies.extend(_inline_policies(group, "GroupPolicyList", "inline:group"))
+            for policy_ref in group.get("AttachedManagedPolicies", []):
+                policy = _managed_policy(policy_ref, managed_by_arn, "managed:group")
+                policy["group"] = group_name
+                policies.append(policy)
+
+        reports.append(
+            {
+                "principal": {
+                    "type": principal_type,
+                    "name": _principal_name(principal),
+                    "arn": principal.get("Arn"),
+                },
+                "policies": policies,
+            }
+        )
+    return reports
+
+
+def run(arguments):
+    from shared.common import custom_serializer, get_us_east_1, parse_arguments
+    from shared.nodes import Account
+    from shared.query import query_aws
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--principal",
+        required=True,
+        help="IAM user or role name or ARN to inspect",
+    )
+    args, accounts, _ = parse_arguments(arguments, parser)
+
+    for account_data in accounts:
+        account = Account(None, account_data)
+        region = get_us_east_1(account)
+        auth_details = query_aws(
+            account, "iam-get-account-authorization-details", region
+        )
+        try:
+            reports = build_principal_policy_report(auth_details, args.principal)
+        except ValueError as error:
+            parser.error("{} in account {}".format(error, account.name))
+        print(
+            json.dumps(
+                {
+                    "account": account.name,
+                    "principal_query": args.principal,
+                    "results": reports,
+                },
+                indent=2,
+                sort_keys=True,
+                default=custom_serializer,
+            )
+        )

tests/unit/test_iam_policies.py

@@ -0,0 +1,113 @@
+import pytest
+
+from commands.iam_policies import build_principal_policy_report
+
+
+def test_builds_policy_report_with_all_sources():
+    details = {
+        "Policies": [
+            {
+                "Arn": "arn:aws:iam::123456789012:policy/Direct",
+                "PolicyName": "Direct",
+                "DefaultVersionId": "v1",
+                "PolicyVersionList": [
+                    {
+                        "VersionId": "v1",
+                        "IsDefaultVersion": True,
+                        "Document": {"Statement": [{"Action": "s3:GetObject"}]},
+                    }
+                ],
+            },
+            {
+                "Arn": "arn:aws:iam::123456789012:policy/Group",
+                "PolicyName": "Group",
+                "DefaultVersionId": "v1",
+                "PolicyVersionList": [
+                    {
+                        "VersionId": "v1",
+                        "IsDefaultVersion": True,
+                        "Document": {"Statement": [{"Action": "logs:CreateLogGroup"}]},
+                    }
+                ],
+            },
+        ],
+        "UserDetailList": [
+            {
+                "Arn": "arn:aws:iam::123456789012:user/alice",
+                "UserName": "alice",
+                "AttachedManagedPolicies": [
+                    {
+                        "PolicyArn": "arn:aws:iam::123456789012:policy/Direct",
+                        "PolicyName": "Direct",
+                    }
+                ],
+                "UserPolicyList": [
+                    {
+                        "PolicyName": "InlineUser",
+                        "PolicyDocument": {"Statement": [{"Action": "iam:GetUser"}]},
+                    }
+                ],
+                "GroupList": ["developers"],
+            }
+        ],
+        "RoleDetailList": [],
+        "GroupDetailList": [
+            {
+                "GroupName": "developers",
+                "GroupPolicyList": [
+                    {
+                        "PolicyName": "InlineGroup",
+                        "PolicyDocument": {"Statement": [{"Action": "ec2:Describe*"}]},
+                    }
+                ],
+                "AttachedManagedPolicies": [
+                    {
+                        "PolicyArn": "arn:aws:iam::123456789012:policy/Group",
+                        "PolicyName": "Group",
+                    }
+                ],
+            }
+        ],
+    }
+
+    report = build_principal_policy_report(details, "alice")
+
+    assert len(report) == 1
+    assert report[0]["principal"]["arn"].endswith("/alice")
+    assert [policy["source"] for policy in report[0]["policies"]] == [
+        "inline:principal",
+        "managed:principal",
+        "inline:group",
+        "managed:group",
+    ]
+    assert report[0]["policies"][1]["document"]["Statement"][0]["Action"] == "s3:GetObject"
+    assert report[0]["policies"][3]["group"] == "developers"
+
+
+def test_build_report_accepts_role_arn():
+    details = {
+        "RoleDetailList": [
+            {
+                "Arn": "arn:aws:iam::123456789012:role/worker",
+                "RoleName": "worker",
+                "AttachedManagedPolicies": [],
+                "RolePolicyList": [],
+            }
+        ]
+    }
+
+    report = build_principal_policy_report(
+        details, "arn:aws:iam::123456789012:role/worker"
+    )
+
+    assert report[0]["principal"] == {
+        "type": "role",
+        "name": "worker",
+        "arn": "arn:aws:iam::123456789012:role/worker",
+    }
+    assert report[0]["policies"] == []
+
+
+def test_build_report_rejects_unknown_principal():
+    with pytest.raises(ValueError, match="was not found"):
+        build_principal_policy_report({}, "nobody")