nulldir.py

#!/usr/bin/env python3
"""
nulldir - Fast & Multi-Threaded CLI Web Directory & Endpoint Fuzzer
Author: 0xcan
"""

import sys
import time
import argparse
import urllib3
from queue import Queue
from threading import Thread, Lock
import requests
from colorama import init, Fore, Style

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

print_lock = Lock()

BANNER = f"""{Fore.CYAN}
               _ _     _ _      
              | | |   | (_)     
  _ __  _   _ | | | __| |_ _ __ 
 | '_ \| | | || | |/ _` | | '__|
 | | | | |_| || | | (_| | | |   
 |_| |_|\__,_||_|_|\__,_|_|_|   
{Fore.YELLOW}  [ Fast & Multi-Threaded Web Endpoint Fuzzer ]
{Style.RESET_ALL}"""

STATUS_COLORS = {
    200: Fore.GREEN,
    204: Fore.GREEN,
    301: Fore.CYAN,
    302: Fore.CYAN,
    307: Fore.CYAN,
    401: Fore.YELLOW,
    403: Fore.MAGENTA,
    500: Fore.RED,
}

def scan_worker(target_url, word_queue, extensions, user_agent, timeout, hide_status, progress_stats):
    headers = {'User-Agent': user_agent}

    while not word_queue.empty():
        word = word_queue.get()
        paths_to_test = [word]
        
        if extensions:
            for ext in extensions:
                ext_clean = ext if ext.startswith('.') else f".{ext}"
                paths_to_test.append(f"{word}{ext_clean}")

        for path in paths_to_test:
            url = f"{target_url}/{path}".replace("([^:])//+", r"\1/")
            try:
                resp = requests.get(
                    url,
                    headers=headers,
                    timeout=timeout,
                    verify=False,
                    allow_redirects=False
                )
                
                status = resp.status_code
                content_len = len(resp.content)

                if status not in hide_status:
                    color = STATUS_COLORS.get(status, Fore.WHITE)
                    with print_lock:
                        print(f"[{color}{status}{Style.RESET_ALL}] {url.ljust(55)} {Fore.BLUE}[Size: {content_len}B]{Style.RESET_ALL}")
                        
            except requests.RequestException:
                pass
            finally:
                with print_lock:
                    progress_stats['scanned'] += 1

        word_queue.task_done()

def main():
    print(BANNER)
    parser = argparse.ArgumentParser(description="nulldir - Fast CLI Web Directory Fuzzer")
    parser.add_argument("-u", "--url", required=True, help="Target URL (e.g., http://example.com)")
    parser.add_argument("-w", "--wordlist", required=True, help="Path to wordlist file")
    parser.add_argument("-t", "--threads", type=int, default=25, help="Number of concurrent threads (Default: 25)")
    parser.add_argument("-x", "--extensions", type=str, help="Extensions to fuzz (e.g. php,html,txt,json)")
    parser.add_argument("--timeout", type=int, default=5, help="Request timeout in seconds (Default: 5)")
    parser.add_argument("--hide", type=str, default="404", help="Status codes to hide, comma separated (Default: 404)")
    parser.add_argument("--ua", type=str, default="nulldir/1.0", help="Custom User-Agent")

    args = parser.parse_args()

    target = args.url.rstrip('/')
    if not target.startswith("http://") and not target.startswith("https://"):
        target = "http://" + target

    try:
        with open(args.wordlist, "r", encoding="utf-8", errors="ignore") as f:
            words = [line.strip().lstrip('/') for line in f if line.strip() and not line.startswith('#')]
    except FileNotFoundError:
        print(f"{Fore.RED}[!] Wordlist file not found: {args.wordlist}{Style.RESET_ALL}")
        sys.exit(1)

    extensions = [e.strip() for e in args.extensions.split(",")] if args.extensions else []
    hide_codes = [int(c.strip()) for c in args.hide.split(",") if c.strip().isdigit()]

    q = Queue()
    for w in words:
        q.put(w)

    stats = {'scanned': 0}

    print(f"{Fore.GREEN}[*] Target:     {Style.RESET_ALL}{target}")
    print(f"{Fore.GREEN}[*] Wordlist:   {Style.RESET_ALL}{args.wordlist} ({len(words)} words)")
    print(f"{Fore.GREEN}[*] Threads:    {Style.RESET_ALL}{args.threads}")
    if extensions:
        print(f"{Fore.GREEN}[*] Extensions: {Style.RESET_ALL}{', '.join(extensions)}")
    print(f"{Fore.GREEN}[*] Filtering:  {Style.RESET_ALL}Hiding status [{', '.join(map(str, hide_codes))}]")
    print(f"{Fore.YELLOW}{'-' * 70}{Style.RESET_ALL}")

    start_time = time.time()

    for _ in range(min(args.threads, len(words))):
        t = Thread(target=scan_worker, args=(target, q, extensions, args.ua, args.timeout, hide_codes, stats))
        t.daemon = True
        t.start()

    try:
        while not q.empty():
            time.sleep(0.5)
        q.join()
    except KeyboardInterrupt:
        print(f"\n{Fore.RED}[!] Aborted by user.{Style.RESET_ALL}")
        sys.exit(0)

    elapsed = round(time.time() - start_time, 2)
    print(f"{Fore.YELLOW}{'-' * 70}{Style.RESET_ALL}")
    print(f"{Fore.GREEN}[+] Scan completed in {elapsed}s | Total Requests: {stats['scanned']}{Style.RESET_ALL}\n")

if __name__ == "__main__":
    main()