-
Notifications
You must be signed in to change notification settings - Fork 6
/
inventory.py
95 lines (74 loc) · 2.54 KB
/
inventory.py
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
import boto3
class Connect(object):
def __init__(self, region='us-west-2', service='ec2'):
self.region = region
self.service = service
try:
self.client = boto3.client(service,region)
except:
raise StandardError(
"No AWS Credentials could be found."
)
class Region(object):
def __init__(self):
self.client = Connect().client
def get_all(self):
availRegions = []
regions = self.client.describe_regions()
for region in regions['Regions']:
availRegions.append(region['RegionName'])
return availRegions
class Instance(object):
def __init__(self):
self.client = Connect().client
def get_running_by_region(self, region):
inventory = []
reservations = Connect(region).client.describe_instances(
Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
)['Reservations']
for reservation in reservations:
for instance in reservation['Instances']:
instance_data = self.__extract_data(instance)
instance_data['region'] = region
inventory.append(instance_data)
return inventory
def __extract_data(self, instance):
return dict(
public_ip_address = instance.get('PublicIpAddress', None),
instance_id = instance['InstanceId'],
launch_time = instance['LaunchTime'],
platform = instance.get('Platform', None),
vpc_id = instance['VpcId'],
ami_id = instance['ImageId'],
volume_ids = [ bdm['Ebs']['VolumeId'] for bdm in instance.get('BlockDeviceMappings', [] ) ],
)
def get_all_running(self):
inventory = {}
for region in Region().get_all():
inventory[region] = self.get_running_by_region(region)
return inventory
class Who(object):
def __init__(self):
self.client = boto3.client('sts')
self.identity = self.client.get_caller_identity()
class CloudTrail(object):
def __init__(self):
self.client = boto3.client('cloudtrail', 'us-west-2')
def get_trails(self):
response = self.client.describe_trails(
includeShadowTrails=True
)
return response
def disrupted(self, trail, trailArn):
response = self.client.get_trail_status(
Name=trailArn
)
description = self.client.describe_trails(
trailNameList=[
trail
]
)
if response['IsLogging'] == False or description['trailList'][0]['KmsKeyId'] != None:
return True
else:
return False