Created
August 14, 2019 03:00
-
-
Save mpomery/f734fa6ffa06689712fdfc6d4c8a5927 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import json | |
import boto3 | |
import os | |
import logging | |
import time | |
import re | |
logger = logging.getLogger() | |
logger.setLevel(logging.INFO) | |
sts = boto3.client('sts') | |
role_name = 'crossaccount_role' | |
accounts = ['123456789012', '123456789012'] | |
sourceaccount = sts.get_caller_identity()['Account'] | |
def aws_session(account_id): | |
if str(account_id) == str(sourceaccount): | |
session = boto3.Session() | |
return session | |
else: | |
try: | |
role_arn = 'arn:aws:iam::' + account_id + ':role/' + role_name | |
logger.info('Trying to assume role: ' + str(role_arn)) | |
if account_id: | |
try: | |
response = sts.assume_role(RoleArn=role_arn, RoleSessionName="AssumedRoleSession") | |
session = boto3.Session( | |
aws_access_key_id=response['Credentials']['AccessKeyId'], | |
aws_secret_access_key=response['Credentials']['SecretAccessKey'], | |
aws_session_token=response['Credentials']['SessionToken']) | |
return session | |
except Exception as e: | |
pass | |
except: | |
pass | |
def main(): | |
# Run across all stated | |
empty_buckets = 0 | |
near_empty_buckets = 0 | |
used_buckets = 0 | |
for account in accounts: | |
session = aws_session(str(account)) | |
if session: | |
print("Account: {}".format(account)) | |
s3_client = session.client('s3') | |
buckets = s3_client.list_buckets() | |
for bucket in buckets['Buckets']: | |
s3Bucket = session.resource('s3').Bucket(bucket['Name']) | |
objects = s3Bucket.objects.all() | |
obj_count = 0 | |
obj_size = 0 | |
try: | |
for s3object in objects: | |
obj_count += 1 | |
if obj_count > 5: | |
break | |
obj_size += s3object.size | |
obj_size_mbs = obj_size // 1048576 | |
if obj_count == 0: | |
empty_buckets += 1 | |
print("\t{}: Empty Bucket".format(bucket['Name'])) | |
elif obj_count <= 5: | |
near_empty_buckets += 1 | |
print("\t{}: {} files taking {}MB".format(bucket['Name'], obj_count, obj_size_mbs)) | |
else: | |
used_buckets += 1 | |
except: | |
print("Unable to access bucket: {}".format(bucket['Name'])) | |
else: | |
print("Unable To Access Account: {}".format(account)) | |
print() | |
print("Empty Buckets: {}".format(empty_buckets)) | |
print("Near Empty Buckets: {}".format(near_empty_buckets)) | |
print("Used Buckets: {}".format(used_buckets)) | |
if __name__ == "__main__": | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment