Introduction to Threat Detection and Management on AWS

[import boto3



# Create a VPC

ec2 = boto3.resource('ec2')

vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16')



# Create public and private subnets

subnet_public = vpc.create_subnet(CidrBlock='10.0.0.0/24', AvailabilityZone='us-east-1a')

subnet_private = vpc.create_subnet(CidrBlock='10.0.1.0/24', AvailabilityZone='us-east-1b')



# Create network ACLs and security groups

acl_public = vpc.create_network_acl()

acl_private = vpc.create_network_acl()

sg_web = vpc.create_security_group(GroupName='WebServerSG', Description='Allow HTTP/HTTPS')

sg_web.authorize_ingress(IpProtocol='tcp', CidrIp='0.0.0.0/0', FromPort=80, ToPort=80)

sg_web.authorize_ingress(IpProtocol='tcp', CidrIp='0.0.0.0/0', FromPort=443, ToPort=443)](<import boto3
from base64 import b64encode
import time

# Create AWS clients
ec2_resource = boto3.resource("ec2")
ec2_client = boto3.client("ec2")
elbv2 = boto3.client('elbv2')
acm = boto3.client('acm')

print("Creating VPC")
vpc = ec2_resource.create_vpc(CidrBlock="172.17.0.0/16")
vpc.create_tags(Tags=[{"Key": "Name", "Value": "my_threat-detection_vpc"}])
vpc.wait_until_available()
print(f"VPC created: {vpc.id}")

print("Creating Internet Gateway")
ig = ec2_resource.create_internet_gateway()
vpc.attach_internet_gateway(InternetGatewayId=ig.id)
print(f"Internet Gateway created: {ig.id}")

print("Creating Route Table")
route_table = vpc.create_route_table()
route = route_table.create_route(DestinationCidrBlock="0.0.0.0/0", GatewayId=ig.id)
print(f"Route Table created: {route_table.id}")

print("Creating Public Subnet 1")
subnet_public1 = vpc.create_subnet(CidrBlock="172.17.1.0/24", AvailabilityZone="us-east-1a")
print(f"Public Subnet 1 created: {subnet_public1.id}")

print("Creating Public Subnet 2")
subnet_public2 = vpc.create_subnet(CidrBlock="172.17.2.0/24", AvailabilityZone="us-east-1b")
print(f"Public Subnet 2 created: {subnet_public2.id}")

print("Creating Private Subnet 1")
subnet_private1 = vpc.create_subnet(CidrBlock="172.17.3.0/24", AvailabilityZone="us-east-1a")
print(f"Private Subnet 1 created: {subnet_private1.id}")

print("Creating Private Subnet 2")
subnet_private2 = vpc.create_subnet(CidrBlock="172.17.5.0/24", AvailabilityZone="us-east-1b")
print(f"Private Subnet 2 created: {subnet_private2.id}")

print("Creating Private Route Table")
route_table_private = vpc.create_route_table()
print(f"Private Route Table created: {route_table_private.id}")

route_table_private.associate_with_subnet(SubnetId=subnet_private1.id)
route_table_private.associate_with_subnet(SubnetId=subnet_private2.id)

route_table.associate_with_subnet(SubnetId=subnet_public1.id)
route_table.associate_with_subnet(SubnetId=subnet_public2.id)

print("Creating Network ACLs")
acl_public = vpc.create_network_acl()
acl_private = vpc.create_network_acl()
print(f"Public Network ACL created: {acl_public.id}")
print(f"Private Network ACL created: {acl_private.id}")

print("Creating Security Group")
sg_web = vpc.create_security_group(GroupName="WebServerSG", Description="Allow HTTP/HTTPS")
sg_web.authorize_ingress(IpProtocol="tcp", CidrIp="0.0.0.0/0", FromPort=80, ToPort=80)
sg_web.authorize_ingress(IpProtocol="tcp", CidrIp="0.0.0.0/0", FromPort=443, ToPort=443)
print(f"Security Group created: {sg_web.id}")

print("Creating Application Load Balancer")
load_balancer = elbv2.create_load_balancer(
Name='MyWebAppLoadBalancer',
Subnets=[subnet_public1.id, subnet_public2.id],
SecurityGroups=[sg_web.id],
Scheme='internet-facing',
Type='application'
)
print(f"Load Balancer created: {load_balancer['LoadBalancers'][0]['LoadBalancerArn']}")

# Defines the user data script
user_data_script = """#!/bin/bash
yum update -y
yum install -y httpd aws-ec2-instance-connect-plugin
systemctl enable httpd
systemctl start httpd
"""

# Encode the user data script as base64
user_data = b64encode(user_data_script.encode('utf-8')).decode('utf-8')

print("Launching EC2 instances")
instances = ec2_resource.create_instances(
ImageId='ami-051f8a213df8bc089',
MinCount=2,
MaxCount=2,
InstanceType='t2.micro',
KeyName='my-demo-environment',
UserData=user_data,
NetworkInterfaces=[
{
'AssociatePublicIpAddress': False,
'DeviceIndex': 0,
'SubnetId': subnet_private1.id,
'Groups': [sg_web.id]
}
]
)

# Function to check the instance state
def check_instance_state(instance_ids, desired_state):
instances = ec2_resource.instances.filter(InstanceIds=instance_ids)
for instance in instances:
instance.wait_until_running()
instance_state = instance.state['Name']
if instance_state != desired_state:
return False
return True

# Wait for the instances to be running
instance_ids = [instance.id for instance in instances]
waiter = ec2_client.get_waiter('instance_running')
start_time = time.time() # Get the current time

try:
waiter.wait(
InstanceIds=instance_ids,
Filters=[
{
'Name': 'instance-state-name',
'Values': ['running']
}
],
WaiterConfig={
'MaxAttempts': 20, # Check 20 times before giving up
'Delay': 15 # Wait 15 seconds between each attempt
}
)
except:
print("Instances did not reach the running state after 5 minutes.")

# Check if the instances are running or if the timeout was reached
if all(instance.state['Name'] == 'running' for instance in ec2_resource.instances.filter(InstanceIds=instance_ids)):
print("EC2 instances are running.")
else:
elapsed_time = time.time() - start_time
if elapsed_time %3E= 300: # 5 minutes * 60 seconds
print("Timeout reached. Continuing with the script.")
else:
print("Instances did not reach the running state within the expected time.")
# Optionally, you can raise an exception or exit the script here

# Wait for the instances to be in the running state
while not check_instance_state(instance_ids, 'running'):
time.sleep(10)

print("Creating Target Group")
target_group_response = elbv2.create_target_group(
Name='MyWebAppTargetGroup',
Protocol='HTTP',
Port=80,
VpcId=vpc.id,
HealthCheckPath='/',
TargetType='instance'
)
target_group = target_group_response['TargetGroups'][0]
print(f"Target Group created: {target_group['TargetGroupArn']}")

# Register instances as targets
targets = []
for instance in instances:
targets.append({'Id': instance.id, 'Port': 80})

# Use the elbv2 client to register targets
register_targets_response = elbv2.register_targets(
TargetGroupArn=target_group['TargetGroupArn'], # Access TargetGroupArn directly
Targets=targets
)

# Request an SSL/TLS certificate from ACM
domain_name = 'example.brandonjcarroll.com'
cert_arn = None

print(f"Checking if certificate for {domain_name} already exists")
certificate_list = acm.list_certificates()['CertificateSummaryList']
for certificate in certificate_list:
if certificate['DomainName'] == domain_name:
cert_arn = certificate['CertificateArn']
print(f"Certificate already exists: {cert_arn}")
break

if not cert_arn:
print(f"Certificate for {domain_name} does not exist. Requesting a new certificate.")
cert_arn = acm.request_certificate(
DomainName=domain_name,
ValidationMethod='DNS'
)['CertificateArn']
print("Please go to the AWS ACM console and complete the CNAME validation to issue the certificate.")

# Wait for the certificate to be issued
cert_status = acm.describe_certificate(CertificateArn=cert_arn)['Certificate']['Status']
while cert_status != 'ISSUED':
time.sleep(10)
cert_status = acm.describe_certificate(CertificateArn=cert_arn)['Certificate']['Status']
print(f"Certificate issued: {cert_arn}")

print("Creating HTTPS Listener")
https_listener = elbv2.create_listener(
LoadBalancerArn=load_balancer['LoadBalancers'][0]['LoadBalancerArn'],
Protocol='HTTPS',
Port=443,
Certificates=[
{
'CertificateArn': cert_arn
}
],
DefaultActions=[
{
'Type': 'forward',
'TargetGroupArn': target_group['TargetGroupArn']
}
]
)
print(f"HTTPS Listener created: {https_listener['Listeners'][0]['ListenerArn']}")

print("Creating HTTP Listener")
http_listener = elbv2.create_listener(
LoadBalancerArn=load_balancer['LoadBalancers'][0]['LoadBalancerArn'],
Protocol='HTTP',
Port=80,
DefaultActions=[
{
'Type': 'redirect',
'RedirectConfig': {
'Protocol': 'HTTPS',
'Port': '443',
'Host': '#{host}',
'Path': '/#{path}',
'Query': '#{query}',
'StatusCode': 'HTTP_301'
}
}
]
)
print(f"HTTP Listener created: {http_listener['Listeners'][0]['ListenerArn']}")

# Create EC2 Instance Connect endpoints for the instances
print("Creating EC2 Instance Connect endpoints")
for instance in instances:
response = ec2_client.create_instance_connect_endpoint(
SubnetId=subnet_private1.id, # Provide the SubnetId instead of InstanceId
PreserveClientIp=True, # Set this to True to allow connections from any IP address
DryRun=False
)
endpoint_id = response['InstanceConnectEndpoint']['InstanceConnectEndpointId'] # Access the key 'InstanceConnectEndpoint'
print(f"EC2 Instance Connect endpoint created for instance {instance.id}: {response['InstanceConnectEndpoint']}")

# Wait for the EC2 Instance Connect Endpoint to be available
print("Waiting for EC2 Instance Connect endpoint to be available...")
while True:
response = ec2_client.describe_instance_connect_endpoints(
InstanceConnectEndpointIds=[endpoint_id]
)
endpoint_state = response['InstanceConnectEndpoints'][0]['State']
if endpoint_state == 'available':
print(f"EC2 Instance Connect endpoint {endpoint_id} is available.")
break
elif endpoint_state == 'failed':
print(f"EC2 Instance Connect endpoint {endpoint_id} creation failed.")
break
time.sleep(10)

resources = {
"VPC": vpc.id,
"Internet Gateway": ig.id,
"Public Route Table": route_table.id,
"Private Route Table": route_table_private.id,
"Public Subnet 1": subnet_public1.id,
"Public Subnet 2": subnet_public2.id,
"Private Subnet 1": subnet_private1.id,
"Private Subnet 2": subnet_private2.id,
"Public Network ACL": acl_public.id,
"Private Network ACL": acl_private.id,
"Security Group": sg_web.id,
"Load Balancer": load_balancer["LoadBalancers"][0]["LoadBalancerArn"],
"Target Group": target_group["TargetGroupArn"],
"HTTPS Listener": https_listener["Listeners"][0]["ListenerArn"],
"HTTP Listener": http_listener["Listeners"][0]["ListenerArn"]
}

print("\nYour resources have been created as follows:")
for resource, resource_id in resources.items():
print(f"{resource}: {resource_id}")>)