audit-labs/audit-tools

  1"""
  2Collect IAM Identity Center (SSO) permission-set assignments for an account.
  3
  4Requires AWS Organizations + IAM Identity Center. Returns an empty list with a
  5warning if no Identity Center instance is available or the target account can't
  6be resolved. When no account name is configured, the current account is used.
  7"""
  8
  9import sys
 10
 11from botocore.exceptions import ClientError
 12
 13from . import api
 14
 15
 16def sso_assignments(cfg):
 17    session = cfg["session"]
 18    sso = session.client("sso-admin")
 19
 20    instances = sso.list_instances().get("Instances", [])
 21    if not instances:
 22        print(
 23            "Warning: no IAM Identity Center instance found -- skipping.",
 24            file=sys.stderr,
 25        )
 26        return []
 27    instance_arn = instances[0]["InstanceArn"]
 28    identity_store_id = instances[0]["IdentityStoreId"]
 29
 30    account = _resolve_account(cfg)
 31    if not account:
 32        return []
 33
 34    ids = session.client("identitystore")
 35    ps_cache = {}
 36    name_cache = {}
 37    rows = []
 38
 39    for ps_arn in _provisioned_permission_sets(sso, instance_arn, account):
 40        assignments = _account_assignments(sso, instance_arn, account, ps_arn)
 41        if not assignments:
 42            continue
 43        ps = _permission_set(sso, instance_arn, ps_arn, ps_cache)
 44        for a in assignments:
 45            rows.append(
 46                {
 47                    "principal": _principal_name(ids, identity_store_id, a, name_cache),
 48                    "type": a["PrincipalType"],
 49                    "permission_set": ps["name"],
 50                    "managed_policies": ps["managed"],
 51                    "inline_policy": ps["inline"],
 52                }
 53            )
 54    return rows
 55
 56
 57def _resolve_account(cfg):
 58    """Resolve the target account ID from the configured account name, or fall
 59    back to the current account when no name is given."""
 60    name = cfg.get("account", "")
 61    if not name:
 62        return api.account_id(cfg)
 63
 64    orgs = cfg["session"].client("organizations")
 65    try:
 66        for page in orgs.get_paginator("list_accounts").paginate():
 67            for acct in page["Accounts"]:
 68                if acct["Name"] == name and acct["Status"] == "ACTIVE":
 69                    return acct["Id"]
 70    except ClientError:
 71        print(
 72            "Warning: could not list organization accounts (needs the management "
 73            "account) -- skipping SSO assignments.",
 74            file=sys.stderr,
 75        )
 76        return ""
 77
 78    print(f"Warning: no active account named '{name}' -- skipping.", file=sys.stderr)
 79    return ""
 80
 81
 82def _provisioned_permission_sets(sso, instance_arn, account):
 83    arns = []
 84    paginator = sso.get_paginator("list_permission_sets_provisioned_to_account")
 85    for page in paginator.paginate(InstanceArn=instance_arn, AccountId=account):
 86        arns.extend(page.get("PermissionSets", []))
 87    return arns
 88
 89
 90def _account_assignments(sso, instance_arn, account, ps_arn):
 91    out = []
 92    paginator = sso.get_paginator("list_account_assignments")
 93    for page in paginator.paginate(
 94        InstanceArn=instance_arn, AccountId=account, PermissionSetArn=ps_arn
 95    ):
 96        out.extend(page.get("AccountAssignments", []))
 97    return out
 98
 99
100def _permission_set(sso, instance_arn, ps_arn, cache):
101    if ps_arn in cache:
102        return cache[ps_arn]
103    name = sso.describe_permission_set(
104        InstanceArn=instance_arn, PermissionSetArn=ps_arn
105    )["PermissionSet"]["Name"]
106    managed = [
107        m["Arn"]
108        for m in sso.list_managed_policies_in_permission_set(
109            InstanceArn=instance_arn, PermissionSetArn=ps_arn
110        ).get("AttachedManagedPolicies", [])
111    ]
112    inline = sso.get_inline_policy_for_permission_set(
113        InstanceArn=instance_arn, PermissionSetArn=ps_arn
114    ).get("InlinePolicy", "")
115    cache[ps_arn] = {
116        "name": name,
117        "managed": ", ".join(managed) or "(none)",
118        "inline": "yes" if inline else "no",
119    }
120    return cache[ps_arn]
121
122
123def _principal_name(ids, store_id, assignment, cache):
124    pid = assignment["PrincipalId"]
125    if pid in cache:
126        return cache[pid]
127    ptype = assignment["PrincipalType"]
128    name = pid
129    try:
130        if ptype == "USER":
131            name = ids.describe_user(IdentityStoreId=store_id, UserId=pid).get(
132                "UserName", pid
133            )
134        elif ptype == "GROUP":
135            name = ids.describe_group(IdentityStoreId=store_id, GroupId=pid).get(
136                "DisplayName", pid
137            )
138    except ClientError:
139        pass
140    cache[pid] = name
141    return name