utils.py

#!/usr/bin/env python3

import argparse  # noqa: F401
import random
import re  # noqa: F401
import os
import socket
import ssl
import string
import sys
import time
import json
import threading
import traceback  # noqa: F401
from urllib.parse import (
    urljoin,  # noqa: F401
    urlparse,
)
from collections import defaultdict
import requests
import urllib3
from bs4 import BeautifulSoup
from bs4 import XMLParsedAsHTMLWarning

from modules.logging_config import configure_logger  # noqa: F401
from utils.style import spinner


class WafAbortError(Exception):
    """Raised when WAF is still active after max retries — thread should stop, not sys.exit()."""
    pass


# Domain-level WAF lock: when one thread detects WAF on a domain,
# ALL threads scanning the same domain wait instead of hammering it.
_waf_domain_locks: dict[str, threading.Lock] = {}
_waf_meta_lock = threading.Lock()

import requests.utils

import warnings
warnings.filterwarnings("ignore", category=XMLParsedAsHTMLWarning)

def _noop_check_header_validity(header, value=None):
    return None

requests.utils.check_header_validity = _noop_check_header_validity

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

logger = configure_logger(__name__)
from modules.lists import user_agents_list

CONTENT_DELTA_RANGE = 500
BIG_CONTENT_DELTA_RANGE = 5000

## url tranformation ##
def get_domain_from_url(url: str) -> str:
    domain = urlparse(url).netloc
    return domain


def get_ip_from_url(url: str) -> str:
    domain = get_domain_from_url(url)
    ip = socket.gethostbyname(domain)
    return ip



## Timing requests ##

def human_time(human: str) -> None:
    if human.isdigit():
        time.sleep(int(human))
    elif human.lower() == "r" or human.lower() == "random":
        time.sleep(random.uniform(0.5, 4.0))
    else:
        pass


def _get_domain_waf_lock(url):
    """Get or create a per-domain lock for WAF waiting."""
    domain = urlparse(url).netloc
    with _waf_meta_lock:
        if domain not in _waf_domain_locks:
            _waf_domain_locks[domain] = threading.Lock()
        return _waf_domain_locks[domain]


def verify_waf(url, s, initialResponse, payload=None, wait_count=0):
    # Acquire domain lock so all threads on same domain wait together
    domain_lock = _get_domain_waf_lock(url)
    
    # Use non-blocking try: if another thread already holds the lock (= already waiting),
    # we just wait for it to release instead of doing our own WAF check
    acquired = domain_lock.acquire(blocking=False)
    if not acquired:
        # Another thread is already handling WAF wait for this domain — just wait
        domain_lock.acquire()
        domain_lock.release()
        return False

    try:
        return _verify_waf_inner(url, s, initialResponse, payload, wait_count)
    finally:
        domain_lock.release()


def _verify_waf_inner(url, s, initialResponse, payload=None, wait_count=0):
    random_param = random.randint(0, 99)

    test_url = f"{url}areuawaaff{random_param}" if "?" in url else f"{url}?cb=areuawaaff{random_param}"
    req_areuawaf = requests.get(test_url, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'}, verify=False, timeout=10)

    html = req_areuawaf.text
    soup = BeautifulSoup(html, "html.parser")
    title = soup.title.string if soup.title else False
    amz_waf = req_areuawaf.headers.get("x-amzn-waf-action", "")
    
    waf_detected = False
    waf_message = ""
    waiting_time = 60
    max_retries=6

    if wait_count >= max_retries:
        print(f" └── [!] WAF still active after {max_retries} retries, wait and rescan with -hu option")
        raise WafAbortError(f"WAF still active on {url} after {max_retries} retries")

    if title and title.lower() == "human verification":
        waf_detected = True
        waf_message = "human verification"
    elif amz_waf and amz_waf.lower() == "captcha":
        waf_detected = True
        waf_message = "Amzn Rate limit"
    elif "verify that you are a real person" in html:
        waf_detected = True
        waf_message = "verify that you are a real person"
    elif "Generated by cloudfront" in html and req_areuawaf.status_code == 403:
        waf_detected = True
        waf_message = "Generated by cloudfront"
    elif initialResponse.status_code not in [405, 403, 412, 429] and req_areuawaf.status_code in [405, 403, 412, 429]:
        waf_detected = True
        waf_message = f"{req_areuawaf.status_code}"
    elif "Your IP has been banned" in html and req_areuawaf.status_code not in [200, 301, 302]:
        waf_detected = True
        waf_message = f"Your IP has been banned"
    
    if waf_detected:
        if wait_count == 0:
            print(f" └── [i] [{waf_message}] WAF activated with {payload} payload, wait a moment or try with -hu option")
        spinner(waiting_time, f"        Waiting {waiting_time}s...", wait_count)
        return _verify_waf_inner(url, s, initialResponse, payload=payload, wait_count=wait_count+1)
    else:
        return False


## Anti FP ##

def range_exclusion(main_len):
    range_exlusion = (
        range(main_len - CONTENT_DELTA_RANGE, main_len + CONTENT_DELTA_RANGE)
        if main_len < 10000
        else range(
            main_len - BIG_CONTENT_DELTA_RANGE,
            main_len + BIG_CONTENT_DELTA_RANGE,
        )
    )
    return range_exlusion


def fp_baseline(url, s):
    uri = f"{url}{random.randint(1, 99)}"
    headers = {"X-Random-azertyuiop":"azertyuiop"}
    req = s.get(uri, headers=headers, allow_redirects=False, timeout=10)

    fp_base_status = req.status_code
    fp_base_len = len(req.content)
    return fp_base_status, fp_base_len
    

## requests settings ##

def parse_headers(header_list: list[str] | None) -> dict[str, str]:
    headers = {}
    if header_list:
        for header in header_list:
            if ":" in header:
                key, value = header.split(":", 1)
                headers[key.strip()] = value.strip()
    return headers
    

def check_auth(auth: str, url: str) -> tuple[str, str] | None:
    try:
        authent = (auth.split(":")[0], auth.split(":")[1])
        r = requests.get(
            url,
            allow_redirects=False,
            verify=False,
            auth=authent,
            timeout=10,  # nosec B501
        )
        if r.status_code in [200, 302, 301]:
            print("\n+ Authentication successful\n")
            return authent
        else:
            print("\nAuthentication error")
            continue_error = input("The authentication seems bad, continue ? [y/N]")
            if continue_error not in ["y", "Y"]:
                print("Exiting")
                sys.exit()
            else:
                return None
    except Exception as e:
        print('Error, the authentication format need to be "user:pass"')
        logger.exception(e)
        sys.exit()


def random_ua():
    random_user_agent = {"User-Agent": random.choice(user_agents_list)}
    return(random_user_agent)


def new_session(base_session=None):
    # Preserve stealth (curl_cffi TLS impersonation) when the base session is a
    # StealthSession — a plain requests.Session would silently drop it.
    if base_session is not None:
        from utils.configure_session import StealthSession
        if isinstance(base_session, StealthSession):
            s = StealthSession()
            s.headers.update(dict(base_session.headers))
            if base_session.proxies:
                s.proxies = dict(base_session.proxies)
            if base_session.cookies:
                s.cookies.update(base_session.cookies)
            return s

    s = requests.Session()

    if base_session:
        s.verify = base_session.verify
        s.max_redirects = base_session.max_redirects

        s.headers.update(base_session.headers)

        s.proxies.update(base_session.proxies)

        for prefix, adapter in base_session.adapters.items():
            s.mount(prefix, adapter)

    return s