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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
| from typing import List, Dict
import boto3
from botocore.exceptions import ClientError
import inquirer
def list_ecs_clusters():
ecs = boto3.client("ecs")
clusters = ecs.list_clusters()
cluster_arns = clusters["clusterArns"]
return cluster_arns
def list_ecs_services(cluster) -> List[str]:
ecs = boto3.client("ecs")
services = ecs.list_services(cluster=cluster)
return services["serviceArns"]
def list_rds_instances() -> Dict[str, str]:
rds = boto3.client("rds")
return rds.describe_db_instances()["DBInstances"]
def get_security_group_from_ecs_service(cluster, service_arn) -> str:
ecs = boto3.client("ecs")
details = ecs.describe_services(cluster=cluster, services=[service_arn])
service = details["services"][0]
deployment = service["deployments"][0]
groups = deployment["networkConfiguration"]["awsvpcConfiguration"]["securityGroups"]
return groups[0]
def get_security_group_ids_for_rds_instance(instance_identifier) -> List[str]:
"""
Returns the security group IDs for a given RDS instance identifier.
:param instance_identifier: The identifier of the RDS instance
:return: A list of security group IDs associated with the RDS instance
"""
rds = boto3.client("rds")
try:
response = rds.describe_db_instances(DBInstanceIdentifier=instance_identifier)
db_instances = response["DBInstances"]
if db_instances:
# Assuming each instance has at least one security group associated with it
security_groups = db_instances[0]["VpcSecurityGroups"]
security_group_ids = [sg["VpcSecurityGroupId"] for sg in security_groups]
return security_group_ids
else:
return []
except Exception as e:
print(
f"Error fetching security group IDs for RDS instance '{instance_identifier}': {e}"
)
return []
def modify_security_group_rules(
security_group_id,
protocol,
from_port,
to_port,
source_security_group_id: str,
description: str,
dry_run: bool,
) -> None:
ec2 = boto3.client("ec2")
if dry_run:
print(
f"Dry run: Would update Security Group {security_group_id} "
f"to allow from {source_security_group_id}"
)
return
try:
ec2.authorize_security_group_ingress(
GroupId=security_group_id,
IpPermissions=[
{
"IpProtocol": protocol,
"FromPort": from_port,
"ToPort": to_port,
"UserIdGroupPairs": [
{
"GroupId": source_security_group_id,
"Description": description,
}
],
}
],
)
print(f"Security Group {security_group_id} updated successfully.")
except ClientError as e:
print(f"Error updating Security Group: {e}")
if __name__ == "__main__":
dry_run = inquirer.confirm(
message="Dry run (no changes will be made)?",
default=True,
)
ecs_clusters = list_ecs_clusters()
rds_instances = list_rds_instances()
# Select ECS Cluster
resource_questions = [
inquirer.List(
"cluster",
message="Select ECS Cluster that contains the service that requires databases access",
choices=ecs_clusters,
default=lambda answers: next(
(cluster for cluster in ecs_clusters if "grafana" in cluster), None
),
),
inquirer.List(
"service",
message="Select ECS Service in {cluster} that requires database access",
choices=lambda answers: list_ecs_services(answers["cluster"]),
),
inquirer.Checkbox(
"rds_instances",
message="Select RDS Instances that will be accessed via {cluster}/{service}",
choices=[instance["DBInstanceIdentifier"] for instance in rds_instances],
),
inquirer.Text(
"description",
message="Provide a description for the connection",
default=lambda answers: f"Allow connections from ECS service: {answers['service'].split(':')[-1]}",
),
]
resources = inquirer.prompt(resource_questions)
# Get Security Group of selected RDS Instances
rds_security_groups = [
get_security_group_ids_for_rds_instance(rds_instance)[0]
for rds_instance in resources["rds_instances"]
]
# Get Security Group of selected ECS Service
ecs_security_group = get_security_group_from_ecs_service(
resources["cluster"], resources["service"]
)
# Update RDS Instances' Security Groups to allow inbound connections from ECS Service
print(f"{rds_security_groups=}")
for rds_sg_id in rds_security_groups:
modify_security_group_rules(
security_group_id=rds_sg_id,
source_security_group_id=ecs_security_group,
protocol="tcp",
to_port=5432,
from_port=5432,
description=resources["description"],
dry_run=dry_run,
)
|