cloudflair.py

#!/usr/bin/env python3

import dns_utils
import cloudflare_utils, cloudfront_utils
import os
import sys
import censys_search
import requests
import urllib3
from html_similarity import similarity
import cli
import random

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

config = {
    'http_timeout_seconds': 3,
    'response_similarity_threshold': 0.9
}

CERT_CHUNK_SIZE = 25


# Returns a legitimate looking user-agent
def get_user_agent():
    user_agents = [
        "Mozilla/5.0 (Windows NT 10.0; rv:68.0) Gecko/20100101 Firefox/68.0",
        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/76.0.3809.100 Safari/537.36",
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.115 Safari/537.36"
    ]
    return random.choice(user_agents)


# Removes any Cloudflare IPs from the given list
def filter_cloudflare_ips(ips):
    return [ ip for ip in ips if not cloudflare_utils.is_cloudflare_ip(ip) ]


# Removes any Cloudfront IPs from the given list
def filter_cloudfront_ips(ips):
    return [ ip for ip in ips if not cloudfront_utils.is_cloudfront_ip(ip) ]


def find_hosts(domain, censys_api_id, censys_api_secret, use_cloudfront):
    if not dns_utils.is_valid_domain(domain):
        sys.stderr.write('[-] The domain "%s" looks invalid.\n' % domain)
        exit(1)

    if not use_cloudfront:
        if not cloudflare_utils.uses_cloudflare(domain):
            print('[-] The domain "%s" does not seem to be behind CloudFlare.' % domain)
            exit(0)

        print('[*] The target appears to be behind CloudFlare.')

    else: 
        if not cloudfront_utils.uses_cloudfront(domain):
            print('[-] The domain "%s" does not seem to be behind CloudFront.' % domain)
            exit(0)

        print('[*] The target appears to be behind CloudFront.')

    print('[*] Looking for certificates matching "%s" using Censys' % domain)
    cert_fingerprints = censys_search.get_certificates(domain, censys_api_id, censys_api_secret)
    cert_fingerprints = list(cert_fingerprints)
    cert_fingerprints_count = len(cert_fingerprints)
    print('[*] %d certificates matching "%s" found.' % (cert_fingerprints_count, domain))

    if cert_fingerprints_count == 0:
        print('Exiting.')
        exit(0)

    chunking = (cert_fingerprints_count > CERT_CHUNK_SIZE)
    if chunking:
        print(f'[*] Splitting the list of certificates into chunks of {CERT_CHUNK_SIZE}.')

    print('[*] Looking for IPv4 hosts presenting these certificates...')
    hosts = set()
    for i in range(0, cert_fingerprints_count, CERT_CHUNK_SIZE):
        if chunking:
            print('[*] Processing chunk %d/%d' % (i/CERT_CHUNK_SIZE + 1, cert_fingerprints_count/CERT_CHUNK_SIZE))
        hosts.update(censys_search.get_hosts(cert_fingerprints[i:i+CERT_CHUNK_SIZE], censys_api_id, censys_api_secret))

    hosts = filter_cloudflare_ips(hosts) if not use_cloudfront else filter_cloudfront_ips(hosts)
    print('[*] %d IPv4 hosts presenting a certificate issued to "%s" were found.' % (len(hosts), domain))

    if len(hosts) == 0:
        print('[-] The target is most likely not vulnerable.')
        exit(0)

    return set(hosts)


def print_hosts(hosts):
    for host in hosts:
        print('  - %s' % host)
    print('')


def retrieve_original_page(domain):
    url = 'https://' + domain
    print('[*] Retrieving target homepage at %s' % url)
    try:
        headers = {'User-Agent': get_user_agent()}
        original_response = requests.get(url, timeout=config['http_timeout_seconds'], headers=headers)
    except requests.exceptions.Timeout:
        sys.stderr.write('[-] %s timed out after %d seconds.\n' % (url, config['http_timeout_seconds']))
        exit(1)
    except requests.exceptions.RequestException:
        sys.stderr.write('[-] Failed to retrieve %s\n' % url)
        exit(1)

    if original_response.status_code != 200:
        print('[-] %s responded with an unexpected HTTP status code %d' % (url, original_response.status_code))
        exit(1)

    if original_response.url != url:
        print('[*] "%s" redirected to "%s"' % (url, original_response.url))

    return original_response

def print_origins(origins):
    for origin in origins:
        print('  - %s (%s)' % (origin[0], origin[1]))

    print('')

def save_origins_to_file(origins, output_file):
    if output_file is None:
        return

    try:
        with open(output_file, 'w') as f:
            for origin in origins:
                f.write(origin[0] + '\n')
        print('[*] Wrote %d likely origins to output file %s' % (len(origins), os.path.abspath(output_file)))
    except IOError as e:
        sys.stderr.write('[-] Unable to write to output file %s : %s\n' % (output_file, e))

def find_origins(domain, candidates):
    print('\n[*] Testing candidate origin servers')
    original_response = retrieve_original_page(domain)
    host_header_value = original_response.url.replace('https://', '').split('/')[0]
    origins = []
    for host in candidates:
        try:
            print('  - %s' % host)
            url = 'https://' + host
            headers = {
                'Host': host_header_value, # only keep the TLD, without any slashes
                'User-Agent': get_user_agent()
            }
            response = requests.get(url, timeout=config['http_timeout_seconds'], headers=headers, verify=False)
        except requests.exceptions.Timeout:
            print('      timed out after %d seconds' % config['http_timeout_seconds'])
            continue
        except requests.exceptions.RequestException as e:
            print('      unable to retrieve')
            continue

        if response.status_code != 200:
            print('      responded with an unexpected HTTP status code %d' % response.status_code)
            continue

        if response.text == original_response.text:
            origins.append((host, 'HTML content identical to %s' % domain))
            continue

        if len(response.text) > 0:
            try:
                page_similarity = similarity(response.text, original_response.text)
            except:
                page_similarity = 0

            if page_similarity > config['response_similarity_threshold']:
                origins.append((host, 'HTML content is %d %% structurally similar to %s' % (round(100 *page_similarity, 2), domain)))

    return origins


def main(domain, output_file, censys_api_id, censys_api_secret, use_cloudfront):
    hosts = find_hosts(domain, censys_api_id, censys_api_secret, use_cloudfront)
    print_hosts(hosts)
    origins = find_origins(domain, hosts)

    if len(origins) == 0:
        print('[-] Did not find any origin server.')
        exit(0)

    print('')
    print('[*] Found %d likely origin servers of %s!' % (len(origins), domain))
    print_origins(origins)
    save_origins_to_file(origins, output_file)

if __name__ == "__main__":
    args = cli.parser.parse_args()

    censys_api_id = None
    censys_api_secret = None

    if 'CENSYS_API_ID' in os.environ and 'CENSYS_API_SECRET' in os.environ:
        censys_api_id = os.environ['CENSYS_API_ID']
        censys_api_secret = os.environ['CENSYS_API_SECRET']

    if args.censys_api_id and args.censys_api_secret:
        censys_api_id = args.censys_api_id
        censys_api_secret = args.censys_api_secret

    if None in [ censys_api_id, censys_api_secret ]:
        sys.stderr.write('[!] Please set your Censys API ID and secret from your environment (CENSYS_API_ID and CENSYS_API_SECRET) or from the command line.\n')
        exit(1)
    
    main(args.domain, args.output_file, censys_api_id, censys_api_secret, args.use_cloudfront)