Summarize IAM policy with policy_sentry

summarize.py

import json

from policy_sentry.analysis.analyze import analyze_by_access_level, determine_actions_to_expand
from policy_sentry.shared.database import connect_db

DB_SESSION = connect_db('bundled')  # Use the bundled data that comes with Policy Sentry

sample_policy = json.loads("""
{
    "Statement":[
        {
            "Action":[
                "s3:ListBucket",
                "s3:ListBucketVersions",
                "s3:Get*",
                "s3:PutObject",
                "s3:PutObjectTagging",
                "s3:PutObjectVersionTagging",
                "s3:ListMultipartUploadParts*",
                "s3:AbortMultipartUpload",
                "s3:RestoreObject",
                "s3:DeleteObject",
                "s3:DeleteObjectTagging",
                "s3:DeleteObjectVersion",
                "s3:DeleteObjectVersionTagging"
            ],
            "Effect":"Allow",
            "Resource":[
                "arn:aws:s3:::role-simulation-test/*",
                "arn:aws:s3:::role-simulation-test"
            ]
        }
    ]
}
""")

def get_access_levels(policy):
    """Check a policy to see which access levels are allowed."""
    allowed_access_level = []
    for access_level in ['read', 'list', 'write', 'tagging', 'permissions-management']:
        if analyze_by_access_level(DB_SESSION, policy, access_level):
            allowed_access_level.append(access_level)
    return allowed_access_level

def get_resources(policy):
    """Naively return resources from the first statement in a policy."""
    return policy["Statement"][0]["Resource"]

def summarize(policy):
    """Generate a human-readable summary of a policy."""
    access = ', '.join(get_access_levels(policy))
    resources = ', '.join(get_resources(policy))
    print(f"This policy allows {access} actions on {resources}")

if __name__ == '__main__':
    summarize(sample_policy)