#!/usr/bin/env python3 """ ================================================================ Public Exposure Demonstration ================================================================ Proves the vulnerable baseline is genuinely leaking data, and that remediation actually stops the leak. Uses plain HTTP GET requests with NO AWS credentials whatsoever, exactly as any person on the internet could. USAGE ----- python demonstrate_leak.py # tests all three files python demonstrate_leak.py --bucket NAME # override auto-detection python demonstrate_leak.py --save-evidence # writes proof to disk WHAT THIS DEMONSTRATES ---------------------- BEFORE remediation: - HTTP 200 responses - Full file contents returned - "LEAK CONFIRMED" verdict AFTER remediation (running scan_and_fix.py's suggested commands against the bucket): - HTTP 403 Forbidden responses - Access denied XML error body - "LEAK CLOSED" verdict Nothing about this script uses boto3 or an AWS session. It is deliberately written the way an attacker or an auditor would test exposure: from outside the account, with no credentials. ================================================================ """ import argparse import json import os import sys import urllib.request import urllib.error from datetime import datetime STATE_FILE = 'demo_resources.json' REGION = 'eu-west-2' DEMO_KEYS = ['customers.csv', 'internal_credentials.txt', 'financial_report_Q4.txt'] def load_bucket_from_state(): """Read the bucket name from the setup script's state file.""" try: with open(STATE_FILE) as f: state = json.load(f) return state.get('bucket') except FileNotFoundError: return None def fetch_without_credentials(url): """ Attempt an unauthenticated HTTP GET. Returns (status_code, body_text). Explicitly does NOT sign the request or attach any AWS credentials. """ try: req = urllib.request.Request(url) req.add_header('User-Agent', 'demonstrate_leak.py / no-credentials-test') with urllib.request.urlopen(req, timeout=10) as response: body = response.read().decode('utf-8', errors='replace') return response.getcode(), body except urllib.error.HTTPError as e: body = e.read().decode('utf-8', errors='replace') if e.fp else '' return e.code, body except urllib.error.URLError as e: return None, f'URL error: {e.reason}' def test_file(bucket, key, save_evidence): """Test a single file for public reachability.""" url = f'https://{bucket}.s3.{REGION}.amazonaws.com/{key}' print(f'\n----- Testing: {key} -----') print(f'URL: {url}') print(f'Credentials sent: NONE (this is what any internet user would see)') status, body = fetch_without_credentials(url) if status == 200: print(f'HTTP Status: 200 OK') print(f'Bytes returned: {len(body)}') print(f'') print(f'--- FILE CONTENTS (first 400 chars) ---') preview = body[:400] for line in preview.split('\n'): print(f' {line}') if len(body) > 400: print(f' ... ({len(body) - 400} more bytes)') print(f'') print(f'VERDICT: [!] LEAK CONFIRMED - file is fully readable by anyone') if save_evidence: ts = datetime.now().strftime('%Y%m%d_%H%M%S') evidence_file = f'evidence_{key.replace(".", "_")}_{ts}.txt' with open(evidence_file, 'w') as f: f.write(f'Evidence captured: {datetime.now().isoformat()}\n') f.write(f'URL: {url}\n') f.write(f'HTTP Status: 200\n') f.write(f'Credentials sent: NONE\n') f.write(f'--- Response body follows ---\n') f.write(body) print(f'Evidence saved: {evidence_file}') return 'leaked' elif status == 403: print(f'HTTP Status: 403 Forbidden') print(f'') print(f'--- SERVER RESPONSE (first 300 chars) ---') for line in body[:300].split('\n'): print(f' {line}') print(f'') print(f'VERDICT: [OK] LEAK CLOSED - AWS is denying anonymous requests') return 'closed' elif status == 404: print(f'HTTP Status: 404 Not Found') print(f'VERDICT: File does not exist (or bucket does not exist)') return 'notfound' else: print(f'HTTP Status: {status}') print(f'Response body: {body[:200]}') print(f'VERDICT: Unexpected status; investigate manually') return 'unknown' def main(): parser = argparse.ArgumentParser(description='Prove the S3 bucket is publicly leaking (or that it has been fixed)') parser.add_argument('--bucket', help='Bucket name (default: read from demo_resources.json)') parser.add_argument('--save-evidence', action='store_true', help='Write successful reads to evidence files on disk') args = parser.parse_args() bucket = args.bucket or load_bucket_from_state() if not bucket: print('ERROR: no bucket name given and no demo_resources.json found.') print(' Either run setup_vulnerable_demo.py first, or pass --bucket NAME.') sys.exit(1) print('=' * 66) print('PUBLIC EXPOSURE DEMONSTRATION') print(f'Bucket: {bucket}') print(f'Region: {REGION}') print(f'Started: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}') print('=' * 66) print('This script sends unauthenticated HTTP requests only.') print('No AWS credentials are used. No boto3 session is created.') results = {} for key in DEMO_KEYS: results[key] = test_file(bucket, key, args.save_evidence) print('\n' + '=' * 66) print('SUMMARY') print('=' * 66) leaked = [k for k, v in results.items() if v == 'leaked'] closed = [k for k, v in results.items() if v == 'closed'] for k, v in results.items(): marker = {'leaked': '[!] LEAKING ', 'closed': '[OK] SECURE ', 'notfound': '[?] MISSING ', 'unknown': '[??] UNKNOWN'}[v] print(f' {marker} {k}') print() if leaked: print(f'{len(leaked)} of {len(DEMO_KEYS)} files are publicly readable without credentials.') print('Baseline exposure is confirmed. Now run scan_and_fix.py to see the') print('standards-mapped remediation plan, apply the S3 fixes, then re-run') print('this script to confirm the leak is closed.') sys.exit(1) elif closed and not leaked: print(f'All {len(closed)} files return 403 Forbidden to unauthenticated requests.') print('Remediation has closed the public leak. Compare this output against') print('the pre-remediation run captured earlier.') sys.exit(0) else: print('No files were reachable at all - the bucket may not exist yet,') print('or the files were never uploaded. Run setup_vulnerable_demo.py.') sys.exit(2) if __name__ == '__main__': main()