#!/usr/bin/env python
import requests
import hashlib
import argparse
from termcolor import colored
from tabulate import tabulate
from urllib.parse import urlparse

"""
Script to detect the presence of cases for which the HTTP request header "Host" can be abused.

Based on the following training and references: 
- https://portswigger.net/web-security/host-header/exploiting
- https://twitter.com/albinowax/status/1402555916482908160
- https://github.com/p0dalirius/ipsourcebypass

Requests module and CURL do not support sending 2 occurence of the "Host" header so the following tests 
must be performed manually via BURP:
- Inject duplicate Host headers
- Add line wrapping

Dependencies:
    pip3 install requests termcolor tabulate
"""

# Config
# Disable TLS warning when validation is disabled when requests is used
requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning)

# Constants
# Define proxy to debug request sent using requests
PROXIES = {}
# Collection of headers used to override the "Host" header
OVERRIDE_HEADERS = ["X-Forwarded-Host", "X-Host", "X-Forwarded-Server", "X-HTTP-Host-Override",
                    "Forwarded", "X-Original-Host", "X-Original-Remote-Addr", "X-Remote-Addr"
                    ]
# Headers taken AS IS (and updated) from the project "https://github.com/p0dalirius/ipsourcebypass"
BYPASS_HEADERS = [
    'Access-Control-Allow-Origin', 'Client-IP', 'Forwarded', 'Forwarded-For', 'Forwarded-For-IP', 'Origin',
    'X-Client-IP', 'X-Custom-IP-Authorization', 'X-Forwarded', 'X-Forwarded-By', 'X-Forwarded-For',
    'X-Forwarded-For-Original', 'X-Forwarded-Host', 'X-Forwarder-For', 'X-Originating-IP', 'X-Remote-Addr',
    'X-Remote-IP', "CF-Connecting-Ip", "X-Real-IP", "True-Client-IP"
]
OVERRIDE_HEADERS.extend(BYPASS_HEADERS)
OVERRIDE_HEADERS = list(dict.fromkeys(OVERRIDE_HEADERS))
OVERRIDE_HEADERS.sort()


class ResponseData:
    def __init__(self, response, comment=""):
        self.return_code = response.status_code
        self.body_length = len(response.text)
        self.body_content_hash = hashlib.sha1(response.text.encode("utf-8")).hexdigest()
        self.comment = comment
        content = ""
        for header in response.headers:
            if header.lower() not in ["date", "set-cookie", "expires", "etag", "last-modified", "content-length"]:
                content += f"{header}={response.headers[header]}"
        self.headers_hash = hashlib.sha1(content.encode("utf-8")).hexdigest()


def baseline(full_url):
    response = requests.get(full_url, verify=False, proxies=PROXIES, allow_redirects=False, timeout=10)
    return [ResponseData(response, "Baseline")]


def case_arbitrary_host_header(full_url, test_value, test_internal_value):
    h = {"Host": test_value}
    response1 = requests.get(full_url, headers=h, verify=False, proxies=PROXIES, allow_redirects=False, timeout=10)
    h = {"Host": f"{test_value}@{test_internal_value}"}
    response2 = requests.get(full_url, headers=h, verify=False, proxies=PROXIES, allow_redirects=False, timeout=10)
    return [ResponseData(response1, f"Host: {test_value}"), ResponseData(response2, f"Host: {test_value}@{test_internal_value}")]


def case_connection_state_attacks(full_url, test_value, test_internal_value):
    # See https://requests.readthedocs.io/en/latest/user/advanced/#keep-alive
    session = requests.Session()
    session.headers.update({"Connection": "keep-alive"})
    response1 = session.get(full_url, verify=False, proxies=PROXIES, allow_redirects=False, timeout=10)
    h = {"Host": test_value}
    response2 = session.get(full_url, headers=h, verify=False, proxies=PROXIES, allow_redirects=False, timeout=10)
    h = {"Host": f"{test_value}@{test_internal_value}"}
    response3 = session.get(full_url, headers=h, verify=False, proxies=PROXIES, allow_redirects=False, timeout=10)
    return [ResponseData(response1, f"Connection state attacks => No host override"), ResponseData(response2, f"Connection state attacks => Host: {test_value}"), ResponseData(response3, f"Connection state attacks => Host: {test_value}@{test_internal_value}")]


def case_host_override_headers(full_url, test_value, test_internal_value):
    results = []
    for override_header in OVERRIDE_HEADERS:
        h = {override_header: test_value}
        response = requests.get(full_url, headers=h, verify=False, proxies=PROXIES, allow_redirects=False, timeout=10)
        results.append(ResponseData(response, f"{override_header}: {test_value}"))
        h = {override_header: f"{test_value}@{test_internal_value}"}
        response = requests.get(full_url, headers=h, verify=False, proxies=PROXIES, allow_redirects=False, timeout=10)
        results.append(ResponseData(response, f"{override_header}: {test_value}@{test_internal_value}"))
    return results


def main(full_url, normal_value, test_value, test_internal_value):
    cases_result = []
    print(colored(f"[+] Execution context:", "yellow"))
    print(f"Full URL                 : {full_url}")
    print(f"Host normal value        : {normal_value}")
    print(f"Host test value          : {test_value}")
    print(f"Host test internal value : {test_internal_value}")
    print(f"Proxy                    : {PROXIES}")
    print(colored(f"[+] Send crafted requests for the different cases...", "yellow"))
    cases_result.extend(baseline(full_url))
    cases_result.extend(case_arbitrary_host_header(full_url, test_value, test_internal_value))
    cases_result.extend(case_arbitrary_host_header(full_url, test_value + ":test", test_internal_value))
    cases_result.extend(case_host_override_headers(full_url, test_value, test_internal_value))
    cases_result.extend(case_connection_state_attacks(full_url, test_value, test_internal_value))
    print(colored(f"[+] Results:", "yellow"))
    data = []
    data.append(["Case", "HTTP Code", "Body length", "Body hash", "Headers hash"])
    data.append([cases_result[0].comment, cases_result[0].return_code, cases_result[0].body_length, cases_result[0].body_content_hash, cases_result[0].headers_hash])
    data.append([cases_result[1].comment, cases_result[1].return_code, cases_result[1].body_length, cases_result[1].body_content_hash, cases_result[1].headers_hash])
    data.append([cases_result[2].comment, cases_result[2].return_code, cases_result[2].body_length, cases_result[2].body_content_hash, cases_result[2].headers_hash])
    for i in range(3, len(cases_result)):
        data.append([cases_result[i].comment, cases_result[i].return_code, cases_result[i].body_length, cases_result[i].body_content_hash, cases_result[i].headers_hash])
    print(tabulate(data, headers="firstrow", tablefmt="github"))
    print(colored(f"[!] REMINDER:", "cyan"))
    print("As Requests module and CURL do not support sending 2 occurences of the 'Host' header then the following tests must be performed manually via BURP:")
    print("- Inject duplicate Host headers")
    print("- Add line wrapping")
    print("=> https://portswigger.net/web-security/host-header/exploiting")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Detect the presence of cases for which the HTTP request header 'Host' can be abused.")
    required_params = parser.add_argument_group("required named arguments")
    required_params.add_argument("-b", action="store", dest="full_url", help="Base URL (ex: 'https://righettod.eu').", required=True)
    parser.add_argument("-t", action="store", dest="test_value", help="Test value for the 'Host' header (default to '5.5.5.5').", required=False, default="5.5.5.5")
    parser.add_argument("-i", action="store", dest="test_internal_value", help="Internal IP to use as test value for abuse using 'external_host@internal_ip' syntax (default to '10.10.10.10').", required=False, default="10.10.10.10")
    parser.add_argument("-x", action="store", dest="proxy", help="Proxy to use for all probe requests (ex: 'http://127.0.0.1:8080', default to no proxy).", required=False, default=None)
    args = parser.parse_args()
    if args.proxy is not None:
        PROXIES = {"http": args.proxy, "https": args.proxy}
    main(args.full_url, urlparse(args.full_url).netloc, args.test_value, args.test_internal_value)
