Scripts-for-Attack-Path-Analysis-in-AWS/Lambda.py at main · sabarish20/Scripts-for-Attack-Path-Analysis-in-AWS · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
import boto3
import argparse
import json

def check_perms(access_key_id, secret_access_key, region, session_token=None):

    sts_kwargs = {
        'aws_access_key_id': access_key_id,
        'aws_secret_access_key': secret_access_key,
        'region_name': region
    }
    if session_token:
        sts_kwargs['aws_session_token'] = session_token
    sts = boto3.client('sts', **sts_kwargs)


    try:
        response = sts.get_caller_identity()
    except sts.exceptions.ClientError as e:
        print(f"Error: {e.response['Error']['Message']}")
        return

    caller_arn = response['Arn']
    principal_type = caller_arn.split(":")[5].split("/")[-2]
    principal_name = caller_arn.split("/")[-1]


    iam = boto3.client('iam', **sts_kwargs)
    attached_policies = []
    inline_policies = []

    if principal_type == 'user':
        attached_policies = iam.list_attached_user_policies(UserName=principal_name)
        inline_policies = iam.list_user_policies(UserName=principal_name)
    elif principal_type == 'role':
        attached_policies = iam.list_attached_role_policies(RoleName=principal_name)
        inline_policies = iam.list_role_policies(RoleName=principal_name)


    required_permissions = [
        'iam:PassRole',
        'lambda:CreateFunction',
        'lambda:InvokeFunction',
        'lambda:AddPermission'
    ]


    def check_policy_document(policy_document, required_permissions):
        if isinstance(policy_document, str):
            policy_document = json.loads(policy_document)
        matched_permissions = []
        for statement in policy_document.get('Statement', []):
            if isinstance(statement, dict) and statement.get('Effect') == 'Allow':
                actions = statement.get('Action', [])
                if isinstance(actions, str):
                    actions = [actions]
                matched_permissions.extend([action for action in actions if action in required_permissions])
        return matched_permissions


    found_permissions = set()
    for policy in attached_policies.get('AttachedPolicies', []):
        policy_arn = policy['PolicyArn']
        policy_name = policy['PolicyName']
        policy_versions = iam.list_policy_versions(PolicyArn=policy_arn)
        for version in policy_versions['Versions']:
            version_id = version['VersionId']
            try:
                policy_version_details = iam.get_policy_version(PolicyArn=policy_arn, VersionId=version_id)
                policy_document = policy_version_details['PolicyVersion']['Document']
                matched_permissions = check_policy_document(policy_document, required_permissions)
                if matched_permissions:
                    found_permissions.update(matched_permissions)
            except iam.exceptions.NoSuchEntityException:
                continue


    for policy_name in inline_policies.get('PolicyNames', []):
        try:
            if principal_type == 'user':
                policy_document = iam.get_user_policy(UserName=principal_name, PolicyName=policy_name)['PolicyDocument']
            elif principal_type == 'role':
                policy_document = iam.get_role_policy(RoleName=principal_name, PolicyName=policy_name)['PolicyDocument']
            matched_permissions = check_policy_document(policy_document, required_permissions)
            if matched_permissions:
                found_permissions.update(matched_permissions)
        except iam.exceptions.NoSuchEntityException:
            continue

    if found_permissions:
        print(f"{principal_type} {principal_name} has the following permissions which can lead to potential attack pathways:")
        for permission in found_permissions:
            print(f" - {permission}")
    else:
        print(f"{principal_type} {principal_name} does not have any of the specified permissions.")

if __name__ == '__main__':

    parser = argparse.ArgumentParser(description='Check IAM and Lambda permissions for the caller')
    parser.add_argument('--access-key-id', required=True)
    parser.add_argument('--secret-access-key', required=True)
    parser.add_argument('--region', required=True)
    parser.add_argument('--session-token', help='AWS session token (optional)', default=None)
    args = parser.parse_args()

    check_perms(args.access_key_id, args.secret_access_key, args.region, args.session_token)