#!/usr/bin/env python3 """ ================================================================ Vulnerable Hybrid-Cloud Demo Setup -- S3 + EC2 + IAM + CloudTrail ================================================================ Creates a deliberately misconfigured AWS environment covering all five categories used in this dissertation's experimental design: 1. S3 - public bucket, public-read policy, no encryption 2. EC2 - security group open to 0.0.0.0/0 on SSH/RDP/HTTP 3. IAM - user with AdministratorAccess attached 4. IAM/MFA - the same user has no MFA device registered 5. CloudTrail - verified absent (no trail is created; this category needs no API call to be "vulnerable", so this script checks and reports the account's starting state instead of manufacturing it) Use with scan_and_fix.py to measure the baseline, then apply the suggested fixes and re-scan to measure the post-remediation state. USAGE ----- python setup_vulnerable_demo.py # full baseline python setup_vulnerable_demo.py --with-instance # also launches a t2.micro python setup_vulnerable_demo.py --skip-iam # S3 + EC2 only (legacy mode) python setup_vulnerable_demo.py --cleanup # removes everything WARNING ------- Only run this in an isolated personal AWS test account. Never run against production or shared environments. Creating an IAM user with AdministratorAccess and a live access key is a genuine security risk if run anywhere other than an isolated account created solely for this dissertation's experiments. ================================================================ """ import boto3 import json import time import argparse from botocore.exceptions import ClientError REGION = 'eu-west-2' TIMESTAMP = int(time.time()) BUCKET_NAME = f'demo-vulnerable-{TIMESTAMP}' SG_NAME = f'demo-open-sg-{TIMESTAMP}' IAM_USER_NAME = f'demo-admin-user-{TIMESTAMP}' STATE_FILE = 'demo_resources.json' # remembers what we created, for cleanup ADMIN_POLICY_ARN = 'arn:aws:iam::aws:policy/AdministratorAccess' def save_state(data): with open(STATE_FILE, 'w') as f: json.dump(data, f, indent=2) def load_state(): try: with open(STATE_FILE) as f: return json.load(f) except FileNotFoundError: return None def create_vulnerable_s3(s3): print(f"[1/4] Creating S3 bucket: {BUCKET_NAME}") if REGION == 'us-east-1': s3.create_bucket(Bucket=BUCKET_NAME) else: s3.create_bucket( Bucket=BUCKET_NAME, CreateBucketConfiguration={'LocationConstraint': REGION} ) # Misconfiguration 1 - disable Block Public Access s3.put_public_access_block( Bucket=BUCKET_NAME, PublicAccessBlockConfiguration={ 'BlockPublicAcls': False, 'IgnorePublicAcls': False, 'BlockPublicPolicy': False, 'RestrictPublicBuckets': False, } ) # Misconfiguration 2 - public-read bucket policy s3.put_bucket_policy( Bucket=BUCKET_NAME, Policy=json.dumps({ "Version": "2012-10-17", "Statement": [{ "Sid": "PublicReadAll", "Effect": "Allow", "Principal": "*", "Action": "s3:GetObject", "Resource": f"arn:aws:s3:::{BUCKET_NAME}/*" }] }) ) # Upload demo files (Misconfiguration 3 - no encryption specified). # Three files with different, realistic-looking dummy content so the # public-read leak actually looks like a leak rather than one placeholder # line. All values are synthetic and pattern-generated; none is a real # person, account, or credential. s3.put_object( Bucket=BUCKET_NAME, Key='customers.csv', Body=(b'customer_id,full_name,email,phone,city,plan\n' b'C1001,Alice Rodriguez,alice.r@example.com,+44-7700-900001,London,Premium\n' b'C1002,Ben Okafor,ben.o@example.com,+44-7700-900002,Manchester,Basic\n' b'C1003,Priya Shah,priya.s@example.com,+44-7700-900003,Birmingham,Premium\n' b'C1004,David Chen,david.c@example.com,+44-7700-900004,Leeds,Enterprise\n' b'C1005,Sarah Williams,sarah.w@example.com,+44-7700-900005,Bristol,Basic\n' b'\n[SYNTHETIC DEMO DATA - NOT REAL CUSTOMERS]\n') ) s3.put_object( Bucket=BUCKET_NAME, Key='internal_credentials.txt', Body=(b'# Internal service credentials - DO NOT SHARE\n' b'db_host: prod-db.internal.example.com\n' b'db_user: app_service\n' b'db_password: FAKE_DEMO_PASSWORD_NotARealSecret_2026\n' b'stripe_api_key: sk_test_FAKE_DEMO_KEY_0000000000000000\n' b'\n[SYNTHETIC DEMO DATA - NOT REAL CREDENTIALS]\n') ) s3.put_object( Bucket=BUCKET_NAME, Key='financial_report_Q4.txt', Body=(b'CONFIDENTIAL - Q4 2025 Internal Financial Summary\n' b'Revenue: GBP 2,847,000\n' b'Operating margin: 18.3%\n' b'Customer accounts: 1,247\n' b'Notable churn events: 12\n' b'\n[SYNTHETIC DEMO DATA - NOT REAL FINANCIALS]\n') ) print(f" Public access: DISABLED (bucket is public)") print(f" Bucket policy: public read for everyone") print(f" Encryption: not configured") print(f" Test files: customers.csv, internal_credentials.txt, financial_report_Q4.txt uploaded\n") return BUCKET_NAME def create_vulnerable_sg(ec2): print(f"[2/4] Creating EC2 security group: {SG_NAME}") vpc_id = ec2.describe_vpcs( Filters=[{'Name': 'isDefault', 'Values': ['true']}] )['Vpcs'][0]['VpcId'] sg = ec2.create_security_group( GroupName=SG_NAME, Description='DEMO - deliberately open security group', VpcId=vpc_id ) sg_id = sg['GroupId'] ec2.authorize_security_group_ingress( GroupId=sg_id, IpPermissions=[ {'IpProtocol': 'tcp', 'FromPort': 22, 'ToPort': 22, 'IpRanges': [{'CidrIp': '0.0.0.0/0', 'Description': 'DEMO - REMOVE AFTER USE'}]}, {'IpProtocol': 'tcp', 'FromPort': 3389, 'ToPort': 3389, 'IpRanges': [{'CidrIp': '0.0.0.0/0', 'Description': 'DEMO - REMOVE AFTER USE'}]}, {'IpProtocol': 'tcp', 'FromPort': 80, 'ToPort': 80, 'IpRanges': [{'CidrIp': '0.0.0.0/0', 'Description': 'DEMO - REMOVE AFTER USE'}]}, ] ) print(f" Port 22 (SSH): open to 0.0.0.0/0") print(f" Port 3389 (RDP): open to 0.0.0.0/0") print(f" Port 80 (HTTP): open to 0.0.0.0/0") print(f" Security group ID: {sg_id}\n") return sg_id, vpc_id def create_vulnerable_iam(iam): """ Misconfiguration 4 & 5 (IAM): create a user, attach the broadest managed policy AWS offers, issue it a live access key, and leave it with no MFA device. This is the single account most of the literature reviewed in Chapter 2 points to: over-permissioned and unprotected by a second factor at the same time. """ print(f"[3/4] Creating IAM user: {IAM_USER_NAME}") iam.create_user(UserName=IAM_USER_NAME) iam.attach_user_policy(UserName=IAM_USER_NAME, PolicyArn=ADMIN_POLICY_ARN) key = iam.create_access_key(UserName=IAM_USER_NAME) access_key_id = key['AccessKey']['AccessKeyId'] secret_access_key = key['AccessKey']['SecretAccessKey'] print(f" Policy attached: AdministratorAccess (full account privileges)") print(f" MFA device: none registered") print(f" Access key ID: {access_key_id}") print(f" NOTE: AWS shows the secret access key exactly once, at creation.") print(f" This script does not print it or store it in {STATE_FILE};") print(f" it exists only so the scanner has a genuine credential to find,") print(f" matching the exposed-access-key pattern documented in Chapter 2.\n") return IAM_USER_NAME, access_key_id def check_cloudtrail(cloudtrail): """ Misconfiguration 5 (CloudTrail): nothing is created here. An absent trail requires no API call to bring about; it is simply the state of a freshly created AWS account. This function only confirms and reports that state so the baseline scan in Section 5.5 has something explicit to reference. """ print(f"[4/4] Checking CloudTrail") trails = cloudtrail.describe_trails()['trailList'] if not trails: print(f" No CloudTrail trail exists in this account.") print(f" This is the default, unmodified state -- nothing to create.\n") else: logging_trails = [] for t in trails: status = cloudtrail.get_trail_status(Name=t['TrailARN']) if status.get('IsLogging'): logging_trails.append(t['Name']) if logging_trails: print(f" WARNING: {len(logging_trails)} trail(s) already logging: " f"{', '.join(logging_trails)}") print(f" The baseline assumes no active trail; results in Chapter 6") print(f" may not reflect a clean starting state.\n") else: print(f" {len(trails)} trail(s) exist but none are actively logging.\n") def launch_demo_instance(ec2, sg_id): print("[extra] Launching a t2.micro instance behind the open security group...") # Get the latest Amazon Linux 2023 AMI via SSM public parameter ssm = boto3.client('ssm', region_name=REGION) ami_id = ssm.get_parameter( Name='/aws/service/ami-amazon-linux-latest/al2023-ami-kernel-default-x86_64' )['Parameter']['Value'] instance = ec2.run_instances( ImageId=ami_id, InstanceType='t2.micro', MinCount=1, MaxCount=1, SecurityGroupIds=[sg_id], TagSpecifications=[{ 'ResourceType': 'instance', 'Tags': [{'Key': 'Name', 'Value': 'demo-vulnerable-instance'}] }] ) instance_id = instance['Instances'][0]['InstanceId'] print(f" Instance launched: {instance_id} (t2.micro, Free Tier eligible)\n") return instance_id def cleanup(): state = load_state() if not state: print("No demo_resources.json found - nothing to clean up.") return s3 = boto3.client('s3', region_name=REGION) ec2 = boto3.client('ec2', region_name=REGION) iam = boto3.client('iam', region_name=REGION) print("Cleaning up demo resources...\n") if state.get('instance_id'): try: ec2.terminate_instances(InstanceIds=[state['instance_id']]) print(f" Terminating instance {state['instance_id']} ...") waiter = ec2.get_waiter('instance_terminated') waiter.wait(InstanceIds=[state['instance_id']]) print(f" Instance terminated.") except ClientError as e: print(f" ! Instance: {e}") if state.get('sg_id'): try: ec2.delete_security_group(GroupId=state['sg_id']) print(f" Security group {state['sg_id']} deleted.") except ClientError as e: print(f" ! Security group: {e}") if state.get('bucket'): try: s3_res = boto3.resource('s3', region_name=REGION) bkt = s3_res.Bucket(state['bucket']) bkt.objects.all().delete() s3.delete_bucket(Bucket=state['bucket']) print(f" Bucket {state['bucket']} deleted.") except ClientError as e: print(f" ! Bucket: {e}") if state.get('iam_user'): # AWS requires access keys and attached policies to be removed # before a user can be deleted, in that order. try: if state.get('access_key_id'): iam.delete_access_key( UserName=state['iam_user'], AccessKeyId=state['access_key_id'] ) iam.detach_user_policy( UserName=state['iam_user'], PolicyArn=ADMIN_POLICY_ARN ) iam.delete_user(UserName=state['iam_user']) print(f" IAM user {state['iam_user']} deleted.") except ClientError as e: print(f" ! IAM user: {e}") import os os.remove(STATE_FILE) print("\nCleanup complete.") def main(): parser = argparse.ArgumentParser(description='Vulnerable hybrid-cloud demo setup (S3 + EC2 + IAM + CloudTrail)') parser.add_argument('--with-instance', action='store_true', help='Also launch a t2.micro instance behind the open SG') parser.add_argument('--skip-iam', action='store_true', help='Skip IAM user creation (legacy S3 + EC2 only mode)') parser.add_argument('--cleanup', action='store_true', help='Delete all demo resources') args = parser.parse_args() if args.cleanup: cleanup() return session = boto3.Session(region_name=REGION) s3 = session.client('s3') ec2 = session.client('ec2') iam = session.client('iam') # IAM is a global service; cloudtrail = session.client('cloudtrail') # region_name is accepted but unused by AWS for it print("=" * 60) print("CREATING VULNERABLE HYBRID-CLOUD DEMO ENVIRONMENT") print("=" * 60) print("Only run this in your own isolated AWS test account.\n") bucket = create_vulnerable_s3(s3) sg_id, vpc_id = create_vulnerable_sg(ec2) state = { 'bucket': bucket, 'sg_id': sg_id, 'vpc_id': vpc_id, 'instance_id': None, 'iam_user': None, 'access_key_id': None, } if not args.skip_iam: iam_user, access_key_id = create_vulnerable_iam(iam) state['iam_user'] = iam_user state['access_key_id'] = access_key_id else: print("[3/4] Skipping IAM user creation (--skip-iam set)\n") check_cloudtrail(cloudtrail) if args.with_instance: instance_id = launch_demo_instance(ec2, sg_id) state['instance_id'] = instance_id save_state(state) print("=" * 60) print("DONE - vulnerable demo environment created.") print(f" Bucket: {bucket}") print(f" Security group: {sg_id}") if state['iam_user']: print(f" IAM user: {state['iam_user']}") if state['instance_id']: print(f" Instance: {state['instance_id']}") print("\nNext: run python scan_and_fix.py to scan it.") print("When finished: python setup_vulnerable_demo.py --cleanup") print("=" * 60) if __name__ == '__main__': main()