cloudflair.py

#!/usr/bin/env python3

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

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

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

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

    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.')

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

    if len(cert_fingerprints) is 0:
        print('Exiting.')
        exit(0)

    print('[*] Looking for IPv4 hosts presenting these certificates...')
    hosts = censys_search.get_hosts(cert_fingerprints, censys_api_id, censys_api_secret)
    hosts = filter_cloudflare_ips(hosts)
    print('[*] %d IPv4 hosts presenting a certificate issued to "%s" were found.' % (len(hosts), domain))

    if len(hosts) is 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:
        original_response = requests.get(url, timeout=config['http_timeout_seconds'])
    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 as e:
        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))

# 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) ]

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
            }
            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

        page_similarity = similarity(response.text, original_response.text)
        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):
    hosts = find_hosts(domain, censys_api_id, censys_api_secret)
    print_hosts(hosts)
    origins = find_origins(domain, hosts)

    if len(origins) is 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)