#!/usr/bin/env python3 """ Account Assessment for AWS Organizations ========================================= Read-only discovery of an AWS Organization's governance posture. Scope: 1. Trusted access services (org-enabled service principals) 2. Delegated administrators (+ delegated services per admin) 3. Trust policies (org resource policy + IAM role trust relationships) The AWS profile is passed into the boto3 Session object (never hard-coded creds). Usage: python3 org_assessment.py [--profile PROFILE] [--out PATH.json] Client: Scrivas | Engagement: cloud footprint discovery """ import argparse import json import os import sys from datetime import datetime, date, timezone import boto3 from botocore.exceptions import ClientError, BotoCoreError DEFAULT_PROFILE = "dasnuve-scrivas-louis-impersonation" DEFAULT_OUT = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "findings", "org_assessment_report.json", ) # Known third-party vendor account IDs, for annotating cross-account trusts. VENDOR_ACCOUNTS = { "123311413059": "Intruder.io (external vulnerability scanning)", "728997465891": "Secureframe (SOC 2 / compliance automation)", } def _default(o): if isinstance(o, (datetime, date)): return o.isoformat() return str(o) def dump(label, obj): print(f"\n{'=' * 70}\n{label}\n{'=' * 70}") print(json.dumps(obj, indent=2, default=_default)) def safe(fn, *args, **kwargs): """Call an API, returning (data, error_string).""" try: return fn(*args, **kwargs), None except (ClientError, BotoCoreError) as e: return None, str(e) def paginate(client, op, key, **kwargs): out = [] try: for page in client.get_paginator(op).paginate(**kwargs): out.extend(page.get(key, [])) except (ClientError, BotoCoreError) as e: return out, str(e) return out, None def assess(session): ident = session.client("sts").get_caller_identity() org = session.client("organizations") report = { "generated": datetime.now(timezone.utc).isoformat(), "caller": {"Account": ident["Account"], "Arn": ident["Arn"]}, } # Organization overview o, err = safe(org.describe_organization) report["organization"] = o.get("Organization") if o else {"error": err} # 1. Trusted access services svcs, err = paginate(org, "list_aws_service_access_for_organization", "EnabledServicePrincipals") report["trusted_access_services"] = {"error": err} if err else svcs # 2. Delegated administrators (+ their delegated services) admins, err = paginate(org, "list_delegated_administrators", "DelegatedAdministrators") if err: report["delegated_administrators"] = {"error": err} else: for a in admins: svc, serr = paginate(org, "list_delegated_services_for_account", "DelegatedServices", AccountId=a["Id"]) a["DelegatedServices"] = {"error": serr} if serr else svc report["delegated_administrators"] = admins # 3a. Trust policies: organization resource-based policy rp, err = safe(org.describe_resource_policy) if err: report["org_resource_policy"] = {"error": err} else: content = rp["ResourcePolicy"]["Content"] report["org_resource_policy"] = json.loads(content) if content else None # 3b. Trust policies: IAM role assume-role (trust) policies iam = session.client("iam") roles, err = paginate(iam, "list_roles", "Roles") if err: report["iam_role_trust_policies"] = {"error": err} else: trust = [] for r in roles: doc = r.get("AssumeRolePolicyDocument", {}) flat = json.dumps(doc) kind = [] if '"Service"' in flat: kind.append("service") if '"AWS"' in flat: kind.append("cross-account/aws") if '"Federated"' in flat: kind.append("federated") statements = [] vendor = None for stmt in doc.get("Statement", []): principal = stmt.get("Principal", {}) aws_p = principal.get("AWS") if isinstance(principal, dict) else None for acct, name in VENDOR_ACCOUNTS.items(): if aws_p and acct in json.dumps(aws_p): vendor = name statements.append({ "Effect": stmt.get("Effect"), "Action": stmt.get("Action"), "Principal": principal, "Condition": stmt.get("Condition"), }) trust.append({ "RoleName": r["RoleName"], "Path": r.get("Path"), "Arn": r["Arn"], "Kind": kind, "Vendor": vendor, "Statements": statements, }) report["iam_role_trust_policies"] = trust return report def summarize(report): """One-line-per-item human summary of the key governance signals.""" print(f"\n{'#' * 70}\n# SUMMARY\n{'#' * 70}") org = report.get("organization", {}) print(f"Org: {org.get('Id')} | mgmt acct {org.get('MasterAccountId')} " f"({org.get('MasterAccountEmail')}) | FeatureSet {org.get('FeatureSet')}") svcs = report.get("trusted_access_services", []) if isinstance(svcs, list): print(f"Trusted access services: {len(svcs)} -> " + ", ".join(s["ServicePrincipal"] for s in svcs)) admins = report.get("delegated_administrators", []) if isinstance(admins, list): flag = " <-- GAP: none set" if not admins else "" print(f"Delegated administrators: {len(admins)}{flag}") roles = report.get("iam_role_trust_policies", []) if isinstance(roles, list): xacct = [r for r in roles if "cross-account/aws" in r["Kind"]] fed = [r for r in roles if "federated" in r["Kind"]] print(f"IAM roles: {len(roles)} total | {len(xacct)} cross-account | " f"{len(fed)} federated") for r in xacct: tag = f" [{r['Vendor']}]" if r.get("Vendor") else "" print(f" - {r['RoleName']}{tag}") def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--profile", default=DEFAULT_PROFILE) ap.add_argument("--out", default=DEFAULT_OUT) ap.add_argument("--quiet", action="store_true", help="skip verbose section dumps, print summary only") args = ap.parse_args() session = boto3.Session(profile_name=args.profile) report = assess(session) if not args.quiet: dump("CALLER / ORG", {"caller": report["caller"], "organization": report["organization"]}) dump("1. TRUSTED ACCESS SERVICES", report["trusted_access_services"]) dump("2. DELEGATED ADMINISTRATORS", report["delegated_administrators"]) dump("3a. ORG RESOURCE POLICY", report["org_resource_policy"]) dump("3b. IAM ROLE TRUST POLICIES", report["iam_role_trust_policies"]) summarize(report) os.makedirs(os.path.dirname(args.out), exist_ok=True) with open(args.out, "w") as f: json.dump(report, f, indent=2, default=_default) print(f"\nFull JSON written to: {args.out}") if __name__ == "__main__": sys.exit(main())