-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathec2_attach_sg.py
63 lines (53 loc) · 2.12 KB
/
ec2_attach_sg.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
'''
## ec2_attach_sg
What it does: Replaces any existing security groups with
the specified security group to an EC2 instance.
Usage: 'AUTO: ec2_atach_sg name_of_sg_to_attach
Limitations: None
'''
import boto3
from botocore.exceptions import ClientError
def run_action(boto_session,rule,entity,params):
instance_id = entity['id']
vpc_id = entity['vpc']['id']
ec2_resource = boto_session.resource('ec2')
ec2_client = boto_session.client('ec2')
# Retrieve params, throw exception if not present
try:
param_group = params[0]
except Exception as e:
return (e)
#Check the specified SG name exists
try:
result = ec2_client.describe_security_groups(
Filters=[
{
'Name': 'group-name',
'Values': [param_group]
},
{
'Name': 'vpc-id',
'Values': [vpc_id]
}
]
)
if result['SecurityGroups']:
sg_group_id = result['SecurityGroups'][0]['GroupId']
text_output = "Existing security group ID: %s \n" % sg_group_id
else:
text_output = text_output + "ERROR: Security group '" + param_group + "' does not exist!"
return text_output
except ClientError as e:
text_output = "Unexpected error: %s \n" % e
text_output = text_output + "Updating the instance SG attachments to contain the noted SG\n"
#Attach the specified security group to the instance, remove others.
try:
result = ec2_resource.Instance(instance_id).modify_attribute(Groups=[sg_group_id])
responseCode = result['ResponseMetadata']['HTTPStatusCode']
if responseCode >= 400:
text_output = text_output + "Unexpected error: %s \n" % str(result)
else:
text_output = text_output + "SG attached: %s \n" % instance_id
except ClientError as e:
text_output = text_output + "Unexpected error: %s \n" % e
return text_output