HostAnalyzer.py

import re
import sys
import time
import random
import urllib3
import argparse
from hashlib import sha256
from functools import partial
from urllib.parse import urlparse, urljoin, quote_plus
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib3.exceptions import NameResolutionError

import requests
import difflib
from colorama import Fore, Style, init


urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
init(autoreset=True)

class Logger:
    _LEVELS = {'debug': 0, 'info': 1, 'warn': 2, 'error': 3}
    _COLORS = {'debug': Fore.WHITE, 'info': Fore.CYAN, 'warn': Fore.YELLOW, 'error': Fore.RED}
    
    def __init__(self, level='info'):
        self.level = self._LEVELS[level]
    
    def _log(self, message, level, **kwargs):
        if self._LEVELS[level] >= self.level:
            timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
            print(f"{self._COLORS[level]}[{timestamp}][{level.upper()}] {message}{Style.RESET_ALL}", **kwargs)

    def debug(self, message): self._log(message, 'debug')
    def info(self, message): self._log(message, 'info')
    def warn(self, message): self._log(message, 'warn')
    def error(self, message): self._log(message, 'error')

logger = Logger()

class RequestEngine:
    def __init__(self, timeout=10, retries=3, jitter=0.5):
        self.timeout = timeout
        self.retries = retries
        self.jitter = jitter
        self.session = requests.Session()
        self.session.verify = False
    
    def _request_with_retry(self, method, url, **kwargs):
        for attempt in range(self.retries):
            try:
                delay = self.timeout * (attempt + 1) * random.uniform(1 - self.jitter, 1 + self.jitter)
                time.sleep(delay)
                return self.session.request(method, url, timeout=self.timeout, **kwargs)
            except Exception as e:
                if attempt == self.retries - 1:
                    raise e
    
    def get(self, url, **kwargs):
        return self._request_with_retry('GET', url, **kwargs)

class ResponseAnalyzer:
    @staticmethod
    def calculate_body_fingerprint(text):
        return sha256(text.encode()).hexdigest()[:16]
    
    @staticmethod
    def advanced_similarity_analysis(a, b):
        simple_ratio = difflib.SequenceMatcher(None, a, b).ratio()
        
        def _ngram_ratio(n=3):
            a_ngrams = set(a[i:i+n] for i in range(len(a)-n+1))
            b_ngrams = set(b[i:i+n] for i in range(len(b)-n+1))
            intersection = a_ngrams & b_ngrams
            return len(intersection) / max(len(a_ngrams), len(b_ngrams))
        
        return (simple_ratio + _ngram_ratio(3) + _ngram_ratio(5)) / 3

class HostInjectionTester:
    CLOUD_METADATA_ENDPOINTS = {
        'aws': ['169.254.169.254'],
        'gcp': ['metadata.google.internal'],
        'azure': ['169.254.169.254', 'metadata.azure.com']
    }
    
    def __init__(self, target_url, config):
        self.target_url = target_url
        self.config = config
        self.engine = RequestEngine(timeout=config.timeout)
        self.baseline = None
        self._validate_target()
    
    def _validate_target(self):
        if not re.match(r'^https?://[^/]+', self.target_url):
            raise ValueError("Invalid target URL format")
        
    def _generate_host_variants(self, base_host):
        variants = []
        encoded_host = quote_plus(base_host)
        for template in [
            'evil.com', 'localhost', '127.0.0.1', '0000::1',
            '0.0.0.0', '10.0.0.1', 'example.com', 'attacker.local',
            f'{base_host}.evil.com', base_host.replace('.', '-') + '.attacker.net',
            f'{base_host}:8080', 'localhost:1337', '[::]', '0x7f.0x0.0x0.0x1',
            base_host + '%23', encoded_host, base_host.upper(), 
            ' ' * 10 + base_host + ' ' * 10
        ]:
            variants.extend(self._generate_cloud_metadata_variants(template))
        return list(set(variants))
    
    def _generate_cloud_metadata_variants(self, host):
        return [host] + [
            f"{host}.{cloud}.internal" 
            for cloud in self.CLOUD_METADATA_ENDPOINTS.keys()
        ]
    
    def _get_headers(self, host):
        return {
            "User-Agent": f"Terminator/1.3.3.7 (Test; {host})",
            "Accept": "*/*",
            "Host": host,
            "X-Forwarded-Host": host,
            "X-Original-Host": host
        }
    
    def establish_baseline(self):
        try:
            response = self.engine.get(
                self.target_url,
                headers=self._get_headers(urlparse(self.target_url).netloc)
            )
            return {
                "status": response.status_code,
                "length": len(response.text),
                "body": response.text,
                "headers": response.headers,
                "fingerprint": ResponseAnalyzer.calculate_body_fingerprint(response.text)
            }
        except Exception as e:
            logger.error(f"Baseline establishment failed: {str(e)}")
            sys.exit(1)
    
    def _analyze_headers(self, response, expected_host):
        reflection_points = []
        for header in ['Location', 'Content-Location', 'X-Forwarded-Host']:
            if header in response.headers:
                if expected_host.lower() in response.headers[header].lower():
                    reflection_points.append(header)
        return reflection_points
    
    def _detect_cloud_metadata(self, response_text):
        cloud_indicators = {
            'aws': ['aws', 'ec2', 's3.amazonaws.com'],
            'gcp': ['google', 'gcp', 'gserviceaccount.com'],
            'azure': ['azure', 'windows.net', 'microsoftonline.com']
        }
        matches = []
        for cloud, keywords in cloud_indicators.items():
            if sum(1 for kw in keywords if kw in response_text.lower()) >= 2:
                matches.append(cloud)
        return matches
    
    def test_host(self, host):
        results = []
        with ThreadPoolExecutor(max_workers=self.config.iterations) as executor:
            futures = []
            for i in range(self.config.iterations):
                futures.append(executor.submit(
                    self._test_host_iteration, 
                    host,
                    i + 1
                ))
            
            for future in as_completed(futures):
                result = future.result()
                if result['issues']:
                    results.append(result)
        return results
    
    def _test_host_iteration(self, host, iteration):
        try:
            response = self.engine.get(
                self.target_url,
                headers=self._get_headers(host))
            
            issues = []
            if response.status_code >= 400:
                return {'iteration': iteration, 'issues': issues}
            
            status_diff = response.status_code != self.baseline['status']
            length_diff = abs(len(response.text) - self.baseline['length']) / self.baseline['length'] > 0.15
            similarity = ResponseAnalyzer.advanced_similarity_analysis(
                self.baseline['body'], response.text)
            content_diff = similarity < self.config.similarity
            
            if status_diff:
                issues.append(f"Status code changed: {self.baseline['status']} → {response.status_code}")
            if length_diff:
                issues.append(f"Content length changed: {self.baseline['length']} → {len(response.text)}")
            if content_diff:
                issues.append(f"Content similarity {similarity:.2f} < {self.config.similarity}")
            
            reflection_headers = self._analyze_headers(response, host)
            if reflection_headers:
                issues.append(f"Host reflection in headers: {', '.join(reflection_headers)}")
            
            if host.lower() in response.text.lower():
                issues.append("Host reflection in response body")
            
            cloud_matches = self._detect_cloud_metadata(response.text)
            if cloud_matches:
                issues.append(f"Detected cloud metadata: {', '.join(cloud_matches)}")
            
            return {'iteration': iteration, 'issues': issues}
        
        except Exception as e:
            logger.debug(f"Iteration {iteration} failed: {str(e)}")
            return {'iteration': iteration, 'issues': ["Request failed"]}

def execute_scan(target, config):
    logger.info(f"Starting scan for {target}")
    tester = HostInjectionTester(target, config)
    tester.baseline = tester.establish_baseline()
    
    logger.info(f"Baseline established: Status {tester.baseline['status']}, "
                f"Length {tester.baseline['length']}, "
                f"Fingerprint {tester.baseline['fingerprint']}")
    
    base_host = urlparse(target).netloc
    hosts_to_test = tester._generate_host_variants(base_host)
    
    for host in hosts_to_test:
        logger.info(f"Testing host: {host}")
        results = tester.test_host(host)
        
        anomalous = sum(1 for r in results if r['issues'])
        total = len(results)
        
        if anomalous > 0:
            logger.warn(f"Host {host}: {anomalous}/{total} anomalous iterations")
            for result in results:
                if result['issues']:
                    logger.warn(f"Iteration {result['iteration']}: " + "; ".join(result['issues']))
            
            if anomalous / total >= 0.25:
                logger.error(f"Potential vulnerability detected with host: {host}")
                logger.info(f"CURL POC: curl -k -H 'Host: {host}' '{target}'")
        else:
            logger.info(f"No anomalies detected for host: {host}")

class Config:
    def __init__(self, args):
        self.timeout = args.timeout
        self.threads = args.threads
        self.iterations = args.iterations
        self.similarity = args.similarity

def main():
    parser = argparse.ArgumentParser(description="Advanced Host Header Injection Scanner")
    parser.add_argument("-u", "--url", help="Target URL")
    parser.add_argument("-f", "--file", help="File with target URLs")
    parser.add_argument("--timeout", type=int, default=10)
    parser.add_argument("--threads", type=int, default=15)
    parser.add_argument("--iterations", type=int, default=12)
    parser.add_argument("--similarity", type=float, default=0.92)
    args = parser.parse_args()
    
    config = Config(args)
    
    targets = []
    if args.url:
        targets.append(args.url.strip())
    if args.file:
        try:
            with open(args.file) as f:
                targets.extend([line.strip() for line in f if line.strip()])
        except Exception as e:
            logger.error(f"Error reading file: {str(e)}")
            sys.exit(1)
    
    with ThreadPoolExecutor(max_workers=config.threads) as executor:
        futures = [executor.submit(execute_scan, target, config) for target in targets]
        for future in as_completed(futures):
            try:
                future.result()
            except Exception as e:
                logger.error(f"Scan failed: {str(e)}")

if __name__ == "__main__":
    main()