From 544ee2dba3cf07bd8268c10a20f8de55500077d5 Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Mon, 26 Jan 2026 14:29:39 -0500 Subject: [PATCH] cleanup of modules --- README.md | 38 +-- hate_crack.py | 61 ++-- hate_crack/api.py | 492 ++++++++++++++++++++++++++++++++- hate_crack/hashview.py | 168 ----------- hate_crack/weakpass.py | 338 ---------------------- tests/test_hashview.py | 2 +- tests/test_hate_crack_utils.py | 38 --- tests/test_menu_snapshots.py | 53 ---- tests/test_module_imports.py | 31 --- 9 files changed, 544 insertions(+), 677 deletions(-) delete mode 100644 hate_crack/hashview.py delete mode 100644 hate_crack/weakpass.py delete mode 100644 tests/test_hate_crack_utils.py delete mode 100644 tests/test_menu_snapshots.py delete mode 100644 tests/test_module_imports.py diff --git a/README.md b/README.md index d9c70b2..19ef267 100644 --- a/README.md +++ b/README.md @@ -113,25 +113,27 @@ Tests automatically run on GitHub Actions for every push and pull request. The w ------------------------------------------------------------------- - (1) Quick Crack - (2) Extensive Pure_Hate Methodology Crack - (3) Brute Force Attack - (4) Top Mask Attack - (5) Fingerprint Attack - (6) Combinator Attack - (7) Hybrid Attack - (8) Pathwell Top 100 Mask Brute Force Crack - (9) PRINCE Attack - (10) YOLO Combinator Attack - (11) Middle Combinator Attack - (12) Thorough Combinator Attack - (13) Bandrel Methodology + (1) Quick Crack + (2) Extensive Pure_Hate Methodology Crack + (3) Brute Force Attack + (4) Top Mask Attack + (5) Fingerprint Attack + (6) Combinator Attack + (7) Hybrid Attack + (8) Pathwell Top 100 Mask Brute Force Crack + (9) PRINCE Attack + (10) YOLO Combinator Attack + (11) Middle Combinator Attack + (12) Thorough Combinator Attack + (13) Bandrel Methodology - (95) Analyze hashes with Pipal - (96) Export Output to Excel Format - (97) Display Cracked Hashes - (98) Display README - (99) Quit + (93) Download Wordlists + (94) Hashview Integration + (95) Analyze hashes with Pipal + (96) Export Output to Excel Format + (97) Display Cracked Hashes + (98) Display README + (99) Quit Select a task: ``` diff --git a/hate_crack.py b/hate_crack.py index 08a881c..753a457 100755 --- a/hate_crack.py +++ b/hate_crack.py @@ -2145,27 +2145,30 @@ def main(): # Display Options try: - options = {"1": quick_crack, - "2": extensive_crack, - "3": brute_force_crack, - "4": top_mask_crack, - "5": fingerprint_crack, - "6": combinator_crack, - "7": hybrid_crack, - "8": pathwell_crack, - "9": prince_attack, - "10": yolo_combination, - "11": middle_combinator, - "12": thorough_combinator, - "13": bandrel_method, - "93": weakpass_wordlist_menu, - "94": hashview_api, - "95": pipal, - "96": export_excel, - "97": show_results, - "98": show_readme, - "99": quit_hc - } + options = { + "1": quick_crack, + "2": extensive_crack, + "3": brute_force_crack, + "4": top_mask_crack, + "5": fingerprint_crack, + "6": combinator_crack, + "7": hybrid_crack, + "8": pathwell_crack, + "9": prince_attack, + "10": yolo_combination, + "11": middle_combinator, + "12": thorough_combinator, + "13": bandrel_method, + "92": weakpass_wordlist_menu, + "93": download_hashmob_wordlists, + "94": weakpass_wordlist_menu, + "95": hashview_api, + "96": pipal, + "97": export_excel, + "98": show_results, + "99": show_readme, + "100": quit_hc + } while 1: print("\n\t(1) Quick Crack") print("\t(2) Extensive Pure_Hate Methodology Crack") @@ -2180,13 +2183,15 @@ def main(): print("\t(11) Middle Combinator Attack") print("\t(12) Thorough Combinator Attack") print("\t(13) Bandrel Methodology") - print("\n\t(93) Download wordlists from Weakpass") - print("\t(94) Hashview") - print("\t(95) Analyze hashes with Pipal") - print("\t(96) Export Output to Excel Format") - print("\t(97) Display Cracked Hashes") - print("\t(98) Display README") - print("\t(99) Quit") + print("\n\t(92) Download wordlists from Weakpass") + print("\t(93) Download wordlists from Hashmob.net") + print("\t(94) Weakpass Wordlist Menu") + print("\t(95) Hashview API") + print("\t(96) Analyze hashes with Pipal") + print("\t(97) Export Output to Excel Format") + print("\t(98) Display Cracked Hashes") + print("\t(99) Display README") + print("\t(100) Quit") try: task = input("\nSelect a task: ") options[task]() diff --git a/hate_crack/api.py b/hate_crack/api.py index 316866a..47ac5b8 100644 --- a/hate_crack/api.py +++ b/hate_crack/api.py @@ -2,8 +2,496 @@ import json import os from typing import Callable, Tuple -from hate_crack.hashview import HashviewAPI -from hate_crack.hashmob_wordlist import list_and_download_official_wordlists + + +import threading +from queue import Queue +import requests +from bs4 import BeautifulSoup +import shutil + +def check_7z(): + import shutil + if shutil.which('7z') or shutil.which('7za'): + return True + print("\n[!] 7z (or 7za) is missing.") + print("To install on macOS: brew install p7zip") + print("To install on Ubuntu/Debian: sudo apt-get install p7zip-full") + print("Please install 7z and try again.") + return False + +def check_transmission_cli(): + import shutil + if shutil.which('transmission-cli'): + return True + print("\n[!] transmission-cli is missing.") + print("To install on macOS: brew install transmission-cli") + print("To install on Ubuntu/Debian: sudo apt-get install transmission-cli") + print("Please install transmission-cli and try again.") + return False + +def get_hcat_wordlists_dir(): + pkg_dir = os.path.dirname(os.path.abspath(__file__)) + project_root = os.path.abspath(os.path.join(pkg_dir, os.pardir)) + candidates = [ + os.path.join(pkg_dir, 'config.json'), + os.path.join(project_root, 'config.json') + ] + default = os.path.join(project_root, 'wordlists') + for config_path in candidates: + try: + if os.path.isfile(config_path): + with open(config_path) as f: + config = json.load(f) + path = config.get('hcatWordlists') + if path: + path = os.path.expanduser(path) + if not os.path.isabs(path): + path = os.path.join(project_root, path) + os.makedirs(path, exist_ok=True) + return path + except Exception: + continue + os.makedirs(default, exist_ok=True) + return default + +def fetch_all_weakpass_wordlists_multithreaded(total_pages=67, threads=10, output_file="weakpass_wordlists.json"): + wordlists = [] + lock = threading.Lock() + q = Queue() + headers = {"User-Agent": "Mozilla/5.0"} + + def worker(): + while True: + page = q.get() + if page is None: + break + try: + url = f"https://weakpass.com/wordlists?page={page}" + r = requests.get(url, headers=headers, timeout=30) + soup = BeautifulSoup(r.text, "html.parser") + app_div = soup.find("div", id="app") + if not app_div or not app_div.has_attr("data-page"): + q.task_done() + continue + data_page_val = app_div["data-page"] + if not isinstance(data_page_val, str): + data_page_val = str(data_page_val) + data = json.loads(data_page_val) + wordlists_data = data.get("props", {}).get("wordlists", {}) + if isinstance(wordlists_data, dict) and 'data' in wordlists_data: + wordlists_data = wordlists_data['data'] + with lock: + for wl in wordlists_data: + wordlists.append({ + "name": wl.get("name", ""), + "size": wl.get("size", ""), + "rank": wl.get("rank", ""), + "downloads": wl.get("downloaded", ""), + "torrent_url": wl.get("torrent_link", "") + }) + except Exception as e: + print(f"Error fetching page {page}: {e}") + q.task_done() + + for page in range(1, total_pages + 1): + q.put(page) + + threads_list = [] + for _ in range(threads): + t = threading.Thread(target=worker) + t.start() + threads_list.append(t) + + q.join() + + for _ in range(threads): + q.put(None) + for t in threads_list: + t.join() + + seen = set() + unique_wordlists = [] + for wl in wordlists: + if wl['name'] not in seen: + unique_wordlists.append(wl) + seen.add(wl['name']) + + with open(output_file, "w", encoding="utf-8") as f: + json.dump(unique_wordlists, f, indent=2) + print(f"Saved {len(unique_wordlists)} wordlists to {output_file}") + +def download_torrent_file(torrent_url, save_dir=None): + if not save_dir: + save_dir = get_hcat_wordlists_dir() + else: + save_dir = os.path.expanduser(save_dir) + if not os.path.isabs(save_dir): + save_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), save_dir) + os.makedirs(save_dir, exist_ok=True) + 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" + } + + if not torrent_url.startswith("http"): + filename = torrent_url + else: + filename = torrent_url.split("/")[-1] + + wordlist_base = filename.replace('.torrent', '').replace('.7z', '').replace('.txt', '') + wordlist_uri = f"https://weakpass.com/wordlists/{wordlist_base}" + print(f"[+] Fetching wordlist page: {wordlist_uri}") + r = requests.get(wordlist_uri, headers=headers) + if r.status_code != 200: + print(f"[!] Failed to fetch wordlist page: {wordlist_uri}") + return None + + soup = BeautifulSoup(r.text, "html.parser") + app_div = soup.find("div", id="app") + if not app_div or not app_div.has_attr("data-page"): + print(f"[!] Could not find app data on {wordlist_uri}") + return None + + data_page_val = app_div["data-page"] + if not isinstance(data_page_val, str): + data_page_val = str(data_page_val) + data_page_val = data_page_val.replace('"', '"') + try: + data = json.loads(data_page_val) + wordlist = data.get('props', {}).get('wordlist') + wordlist_id = None + torrent_link_from_data = None + if wordlist: + wordlist_id = wordlist.get('id') + torrent_link_from_data = wordlist.get('torrent_link') + else: + wordlists = data.get('props', {}).get('wordlists') + if isinstance(wordlists, dict) and 'data' in wordlists: + wordlists = wordlists['data'] + if isinstance(wordlists, list): + for wl in wordlists: + if wl.get('torrent_link') == filename or wl.get('name') == filename: + wordlist_id = wl.get('id') + torrent_link_from_data = wl.get('torrent_link') + break + if wordlist_base in wl.get('name', ''): + wordlist_id = wl.get('id') + torrent_link_from_data = wl.get('torrent_link') + break + except Exception as e: + print(f"[!] Failed to parse data-page JSON: {e}") + return None + + if not (torrent_link_from_data and wordlist_id): + print(f"[!] No torrent link or id found in wordlist data for {filename}.") + return None + if not torrent_link_from_data.startswith('http'): + torrent_link = f"https://weakpass.com/download/{wordlist_id}/{torrent_link_from_data}" + else: + torrent_link = torrent_link_from_data + + print(f"[+] Downloading .torrent file from: {torrent_link}") + r2 = requests.get(torrent_link, headers=headers, stream=True) + content_type = r2.headers.get("Content-Type", "") + local_filename = os.path.join(save_dir, filename if filename.endswith('.torrent') else filename + '.torrent') + if r2.status_code == 200 and not content_type.startswith("text/html"): + with open(local_filename, 'wb') as f: + for chunk in r2.iter_content(chunk_size=8192): + if chunk: + f.write(chunk) + print(f"Saved to {local_filename}") + else: + print(f"Failed to download a valid torrent file: {torrent_link}") + try: + html = r2.content.decode(errors="replace") + print("--- Begin HTML Debug Output ---") + print(html[:2000]) + print("--- End HTML Debug Output ---") + except Exception as e: + print(f"Could not decode response for debug: {e}") + return None + + if shutil.which("transmission-cli") is None: + print("[ERROR] transmission-cli is not installed or not in your PATH.") + print("Please install it with: brew install transmission-cli (on macOS) or your package manager.") + print(f"Torrent file saved at {local_filename}, but download will not start until transmission-cli is available.") + return local_filename + + def run_transmission(torrent_file, output_dir): + import subprocess + import glob + print(f"Starting transmission-cli for {torrent_file}...") + try: + proc = subprocess.Popen([ + "transmission-cli", + "-w", output_dir, + torrent_file + ], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, universal_newlines=True) + if proc.stdout is not None: + for line in proc.stdout: + print(line, end='') + proc.wait() + if proc.returncode != 0: + print(f"transmission-cli failed for {torrent_file} (exit {proc.returncode})") + return + else: + print(f"Download complete for {torrent_file}") + sevenz_files = glob.glob(os.path.join(output_dir, '*.7z')) + if not sevenz_files: + print("[i] No .7z files found to extract.") + return + for zfile in sevenz_files: + print(f"[+] Extracting {zfile} ...") + sevenz_bin = shutil.which('7z') or shutil.which('7za') + if not sevenz_bin: + print("[!] 7z or 7za not found in PATH. Please install p7zip-full or 7-zip to extract archives.") + continue + try: + extract_result = subprocess.run([ + sevenz_bin, 'x', '-y', zfile, f'-o{output_dir}' + ], capture_output=True, text=True) + print(extract_result.stdout) + if extract_result.returncode == 0: + print(f"[+] Extraction complete: {zfile}") + else: + print(f"[!] Extraction failed for {zfile}: {extract_result.stderr}") + except Exception as e: + print(f"[!] Error extracting {zfile}: {e}") + except Exception as e: + print(f"Error running transmission-cli: {e}") + + t = threading.Thread(target=run_transmission, args=(local_filename, save_dir)) + t.start() + print(f"transmission-cli launched in background for {local_filename}") + return local_filename + +def weakpass_wordlist_menu(): + fetch_all_weakpass_wordlists_multithreaded() + try: + with open("weakpass_wordlists.json", "r", encoding="utf-8") as f: + all_wordlists = json.load(f) + except Exception as e: + print(f"Failed to load local wordlist cache: {e}") + return + page = 0 + batch_size = 100 + while True: + filtered_wordlists = [wl for wl in all_wordlists if str(wl.get('rank', '')) == '7'] + start_idx = page * batch_size + end_idx = start_idx + batch_size + page_wordlists = filtered_wordlists[start_idx:end_idx] + if not page_wordlists: + print("No more wordlists.") + if page > 0: + page -= 1 + continue + col_width = 45 + cols = 3 + print("\nEach entry shows: [number]. [wordlist name] [effectiveness score] [rank]") + print(f"Available Wordlists (Batch {page+1}):") + rows = (len(page_wordlists) + cols - 1) // cols + lines = [''] * rows + for idx, wl in enumerate(page_wordlists): + col = idx // rows + row = idx % rows + effectiveness = wl.get('effectiveness', wl.get('downloads', '')) + rank = wl.get('rank', '') + entry = f"{start_idx+idx+1:3d}. {wl['name'][:25]:<25} {effectiveness:<8} {rank:<2}" + lines[row] += entry.ljust(col_width) + for line in lines: + print(line) + sel = input("\nEnter the number to download, 'n' for next batch, 'p' for previous, or 'q' to cancel: ") + if sel.lower() == 'q': + print("Returning to menu...") + return + if sel.lower() == 'n': + page += 1 + continue + if sel.lower() == 'p' and page > 0: + page -= 1 + continue + try: + sel_idx = int(sel) - 1 - start_idx + if 0 <= sel_idx < len(page_wordlists): + torrent_url = page_wordlists[sel_idx]['torrent_url'] + download_torrent_file(torrent_url) + else: + print("Invalid selection.") + except Exception as e: + print(f"Error: {e}") +import requests + +# Hashview Integration - Real API implementation matching hate_crack.py +class HashviewAPI: + def get_hashfile_details(self, hashfile_id): + """Get hashfile details and hashtype for a given hashfile_id.""" + url = f"{self.base_url}/v1/hashfiles/{hashfile_id}" + resp = self.session.get(url) + resp.raise_for_status() + # Try to parse JSON if available, else fallback to raw content + try: + data = resp.json() + except Exception: + data = None + # If JSON, look for hashtype + hashtype = None + if data: + hashtype = data.get('hashtype') or data.get('hash_type') + return {'hashfile_id': hashfile_id, 'hashtype': hashtype, 'details': data, 'raw': resp.content} + + FILE_FORMATS = { + 'pwdump': 0, + 'netntlm': 1, + 'kerberos': 2, + 'shadow': 3, + 'user:hash': 4, + 'hash_only': 5, + } + + def __init__(self, base_url, api_key, debug=False): + self.base_url = base_url.rstrip('/') + self.api_key = api_key + self.debug = debug + self.session = requests.Session() + self.session.cookies.set('uuid', api_key) + self.session.verify = False + import urllib3 + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + + def list_customers(self): + url = f"{self.base_url}/v1/customers" + resp = self.session.get(url) + resp.raise_for_status() + data = resp.json() + if 'users' in data: + customers = json.loads(data['users']) + return {'customers': customers} + return data + + def list_hashfiles(self): + url = f"{self.base_url}/v1/hashfiles" + resp = self.session.get(url) + resp.raise_for_status() + data = resp.json() + if 'hashfiles' in data: + if isinstance(data['hashfiles'], str): + hashfiles = json.loads(data['hashfiles']) + else: + hashfiles = data['hashfiles'] + return hashfiles + return [] + + def get_customer_hashfiles(self, customer_id): + all_hashfiles = self.list_hashfiles() + return [hf for hf in all_hashfiles if int(hf.get('customer_id', 0)) == customer_id] + + def display_customers_multicolumn(self, customers): + if not customers: + print("\nNo customers found.") + return + try: + terminal_width = os.get_terminal_size().columns + except: + terminal_width = 120 + max_id_len = max(len(str(c.get('id', ''))) for c in customers) + col_width = max_id_len + 2 + 30 + 2 + num_cols = max(1, terminal_width // col_width) + print("\n" + "="*terminal_width) + print("Available Customers:") + print("="*terminal_width) + num_customers = len(customers) + rows = (num_customers + num_cols - 1) // num_cols + for row in range(rows): + line_parts = [] + for col in range(num_cols): + idx = row + col * rows + if idx < num_customers: + customer = customers[idx] + cust_id = customer.get('id', 'N/A') + cust_name = customer.get('name', 'N/A') + name_width = col_width - max_id_len - 2 - 2 + if len(str(cust_name)) > name_width: + cust_name = str(cust_name)[:name_width-3] + "..." + entry = f"{cust_id}: {cust_name}" + line_parts.append(entry.ljust(col_width)) + print("".join(line_parts).rstrip()) + print("="*terminal_width) + print(f"Total: {len(customers)} customer(s)") + + def upload_hashfile(self, file_path, customer_id, hash_type, file_format=5, hashfile_name=None): + if hashfile_name is None: + hashfile_name = os.path.basename(file_path) + with open(file_path, 'rb') as f: + file_content = f.read() + url = ( + f"{self.base_url}/v1/hashfiles/upload/" + f"{customer_id}/{file_format}/{hash_type}/{hashfile_name}" + ) + headers = {'Content-Type': 'text/plain'} + resp = self.session.post(url, data=file_content, headers=headers) + resp.raise_for_status() + return resp.json() + + def create_job(self, name, hashfile_id, customer_id, limit_recovered=False, notify_email=True): + url = f"{self.base_url}/v1/jobs/add" + headers = {'Content-Type': 'application/json'} + data = { + "name": name, + "hashfile_id": hashfile_id, + "customer_id": customer_id, + } + resp = self.session.post(url, json=data, headers=headers) + resp.raise_for_status() + return resp.json() + + def download_left_hashes(self, customer_id, hashfile_id, output_file=None): + url = f"{self.base_url}/v1/hashfiles/{hashfile_id}" + resp = self.session.get(url) + resp.raise_for_status() + if output_file is None: + output_file = f"left_{customer_id}_{hashfile_id}.txt" + with open(output_file, 'wb') as f: + f.write(resp.content) + return {'output_file': output_file, 'size': len(resp.content)} + + def upload_cracked_hashes(self, file_path, hash_type='1000'): + valid_lines = [] + with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: + for line in f: + line = line.strip() + if '31d6cfe0d16ae931b73c59d7e0c089c0' in line: + continue + if not line or ':' not in line: + continue + parts = line.split(':', 1) + if len(parts) != 2: + break + hash_value = parts[0].strip() + plaintext = parts[1].strip() + valid_lines.append(f"{hash_value}:{plaintext}") + converted_content = '\n'.join(valid_lines) + url = f"{self.base_url}/v1/hashes/import/{hash_type}" + headers = {'Content-Type': 'text/plain'} + resp = self.session.post(url, data=converted_content, headers=headers) + resp.raise_for_status() + try: + json_response = resp.json() + if 'type' in json_response and json_response['type'] == 'Error': + raise Exception(f"Hashview API Error: {json_response.get('msg', 'Unknown error')}") + return json_response + except (json.JSONDecodeError, ValueError): + raise Exception(f"Invalid API response: {resp.text[:200]}") + + def create_customer(self, name): + url = f"{self.base_url}/v1/customers/add" + headers = {'Content-Type': 'application/json'} + data = {"name": name} + resp = self.session.post(url, json=data, headers=headers) + resp.raise_for_status() + return resp.json() + + + def download_hashes_from_hashview( diff --git a/hate_crack/hashview.py b/hate_crack/hashview.py deleted file mode 100644 index ea1388f..0000000 --- a/hate_crack/hashview.py +++ /dev/null @@ -1,168 +0,0 @@ -""" -hashview_api.py -Modularized Hashview API integration (class only). -""" - - -import os -import json -import requests - - - - -# Hashview Integration - Real API implementation matching hate_crack.py -class HashviewAPI: - """Hashview API integration for uploading/downloading hashfiles, wordlists, jobs, and customers.""" - - FILE_FORMATS = { - 'pwdump': 0, - 'netntlm': 1, - 'kerberos': 2, - 'shadow': 3, - 'user:hash': 4, - 'hash_only': 5, - } - - def __init__(self, base_url, api_key, debug=False): - self.base_url = base_url.rstrip('/') - self.api_key = api_key - self.debug = debug - self.session = requests.Session() - self.session.cookies.set('uuid', api_key) - self.session.verify = False - import urllib3 - urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - - def list_customers(self): - url = f"{self.base_url}/v1/customers" - resp = self.session.get(url) - resp.raise_for_status() - data = resp.json() - if 'users' in data: - customers = json.loads(data['users']) - return {'customers': customers} - return data - - def list_hashfiles(self): - url = f"{self.base_url}/v1/hashfiles" - resp = self.session.get(url) - resp.raise_for_status() - data = resp.json() - if 'hashfiles' in data: - if isinstance(data['hashfiles'], str): - hashfiles = json.loads(data['hashfiles']) - else: - hashfiles = data['hashfiles'] - return hashfiles - return [] - - def get_customer_hashfiles(self, customer_id): - all_hashfiles = self.list_hashfiles() - return [hf for hf in all_hashfiles if int(hf.get('customer_id', 0)) == customer_id] - - def display_customers_multicolumn(self, customers): - if not customers: - print("\nNo customers found.") - return - try: - terminal_width = os.get_terminal_size().columns - except: - terminal_width = 120 - max_id_len = max(len(str(c.get('id', ''))) for c in customers) - col_width = max_id_len + 2 + 30 + 2 - num_cols = max(1, terminal_width // col_width) - print("\n" + "="*terminal_width) - print("Available Customers:") - print("="*terminal_width) - num_customers = len(customers) - rows = (num_customers + num_cols - 1) // num_cols - for row in range(rows): - line_parts = [] - for col in range(num_cols): - idx = row + col * rows - if idx < num_customers: - customer = customers[idx] - cust_id = customer.get('id', 'N/A') - cust_name = customer.get('name', 'N/A') - name_width = col_width - max_id_len - 2 - 2 - if len(str(cust_name)) > name_width: - cust_name = str(cust_name)[:name_width-3] + "..." - entry = f"{cust_id}: {cust_name}" - line_parts.append(entry.ljust(col_width)) - print("".join(line_parts).rstrip()) - print("="*terminal_width) - print(f"Total: {len(customers)} customer(s)") - - def upload_hashfile(self, file_path, customer_id, hash_type, file_format=5, hashfile_name=None): - if hashfile_name is None: - hashfile_name = os.path.basename(file_path) - with open(file_path, 'rb') as f: - file_content = f.read() - url = ( - f"{self.base_url}/v1/hashfiles/upload/" - f"{customer_id}/{file_format}/{hash_type}/{hashfile_name}" - ) - headers = {'Content-Type': 'text/plain'} - resp = self.session.post(url, data=file_content, headers=headers) - resp.raise_for_status() - return resp.json() - - def create_job(self, name, hashfile_id, customer_id, limit_recovered=False, notify_email=True): - url = f"{self.base_url}/v1/jobs/add" - headers = {'Content-Type': 'application/json'} - data = { - "name": name, - "hashfile_id": hashfile_id, - "customer_id": customer_id, - } - resp = self.session.post(url, json=data, headers=headers) - resp.raise_for_status() - return resp.json() - - def download_left_hashes(self, customer_id, hashfile_id, output_file=None): - url = f"{self.base_url}/v1/hashfiles/{hashfile_id}" - resp = self.session.get(url) - resp.raise_for_status() - if output_file is None: - output_file = f"left_{customer_id}_{hashfile_id}.txt" - with open(output_file, 'wb') as f: - f.write(resp.content) - return {'output_file': output_file, 'size': len(resp.content)} - - def upload_cracked_hashes(self, file_path, hash_type='1000'): - valid_lines = [] - with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: - for line in f: - line = line.strip() - if '31d6cfe0d16ae931b73c59d7e0c089c0' in line: - continue - if not line or ':' not in line: - continue - parts = line.split(':', 1) - if len(parts) != 2: - break - hash_value = parts[0].strip() - plaintext = parts[1].strip() - valid_lines.append(f"{hash_value}:{plaintext}") - converted_content = '\n'.join(valid_lines) - url = f"{self.base_url}/v1/hashes/import/{hash_type}" - headers = {'Content-Type': 'text/plain'} - resp = self.session.post(url, data=converted_content, headers=headers) - resp.raise_for_status() - try: - json_response = resp.json() - if 'type' in json_response and json_response['type'] == 'Error': - raise Exception(f"Hashview API Error: {json_response.get('msg', 'Unknown error')}") - return json_response - except (json.JSONDecodeError, ValueError): - raise Exception(f"Invalid API response: {resp.text[:200]}") - - def create_customer(self, name): - url = f"{self.base_url}/v1/customers/add" - headers = {'Content-Type': 'application/json'} - data = {"name": name} - resp = self.session.post(url, json=data, headers=headers) - resp.raise_for_status() - return resp.json() - diff --git a/hate_crack/weakpass.py b/hate_crack/weakpass.py deleted file mode 100644 index 0b23da8..0000000 --- a/hate_crack/weakpass.py +++ /dev/null @@ -1,338 +0,0 @@ - -def check_7z(): - """ - Check if 7z (or 7za) is installed in PATH. Print instructions if missing. - Returns True if found, False otherwise. - """ - import shutil - if shutil.which('7z') or shutil.which('7za'): - return True - print("\n[!] 7z (or 7za) is missing.") - print("To install on macOS: brew install p7zip") - print("To install on Ubuntu/Debian: sudo apt-get install p7zip-full") - print("Please install 7z and try again.") - return False - -def check_transmission_cli(): - """ - Check if transmission-cli is installed in PATH. Print instructions if missing. - Returns True if found, False otherwise. - """ - import shutil - if shutil.which('transmission-cli'): - return True - print("\n[!] transmission-cli is missing.") - print("To install on macOS: brew install transmission-cli") - print("To install on Ubuntu/Debian: sudo apt-get install transmission-cli") - print("Please install transmission-cli and try again.") - return False -""" -weakpass.py -Modularized Weakpass integration functions. -""" - -import os -import threading -from queue import Queue -import requests -from bs4 import BeautifulSoup -import json -import shutil - -def get_hcat_wordlists_dir(): - """Return the configured `hcatWordlists` directory from config.json. - - Looks for config.json in: - 1) The package directory (hate_crack/config.json) - 2) The project root (parent of package) - Falls back to './wordlists' within the project root if not found. - """ - pkg_dir = os.path.dirname(os.path.abspath(__file__)) - project_root = os.path.abspath(os.path.join(pkg_dir, os.pardir)) - candidates = [ - os.path.join(pkg_dir, 'config.json'), - os.path.join(project_root, 'config.json') - ] - default = os.path.join(project_root, 'wordlists') - for config_path in candidates: - try: - if os.path.isfile(config_path): - with open(config_path) as f: - config = json.load(f) - path = config.get('hcatWordlists') - if path: - path = os.path.expanduser(path) - if not os.path.isabs(path): - path = os.path.join(project_root, path) - os.makedirs(path, exist_ok=True) - return path - except Exception: - continue - os.makedirs(default, exist_ok=True) - return default -def fetch_all_weakpass_wordlists_multithreaded(total_pages=67, threads=10, output_file="weakpass_wordlists.json"): - """Fetch all Weakpass wordlist pages in parallel using threads and save to a local JSON file.""" - wordlists = [] - lock = threading.Lock() - q = Queue() - headers = {"User-Agent": "Mozilla/5.0"} - - def worker(): - while True: - page = q.get() - if page is None: - break - try: - url = f"https://weakpass.com/wordlists?page={page}" - r = requests.get(url, headers=headers, timeout=30) - soup = BeautifulSoup(r.text, "html.parser") - app_div = soup.find("div", id="app") - if not app_div or not app_div.has_attr("data-page"): - q.task_done() - continue - data_page_val = app_div["data-page"] - if not isinstance(data_page_val, str): - data_page_val = str(data_page_val) - data = json.loads(data_page_val) - wordlists_data = data.get("props", {}).get("wordlists", {}) - if isinstance(wordlists_data, dict) and 'data' in wordlists_data: - wordlists_data = wordlists_data['data'] - with lock: - for wl in wordlists_data: - wordlists.append({ - "name": wl.get("name", ""), - "size": wl.get("size", ""), - "rank": wl.get("rank", ""), - "downloads": wl.get("downloaded", ""), - "torrent_url": wl.get("torrent_link", "") - }) - except Exception as e: - print(f"Error fetching page {page}: {e}") - q.task_done() - - for page in range(1, total_pages + 1): - q.put(page) - - threads_list = [] - for _ in range(threads): - t = threading.Thread(target=worker) - t.start() - threads_list.append(t) - - q.join() - - for _ in range(threads): - q.put(None) - for t in threads_list: - t.join() - - seen = set() - unique_wordlists = [] - for wl in wordlists: - if wl['name'] not in seen: - unique_wordlists.append(wl) - seen.add(wl['name']) - - with open(output_file, "w", encoding="utf-8") as f: - json.dump(unique_wordlists, f, indent=2) - print(f"Saved {len(unique_wordlists)} wordlists to {output_file}") - -def download_torrent_file(torrent_url, save_dir=None): - # Use configured hcat wordlists directory by default - if not save_dir: - save_dir = get_hcat_wordlists_dir() - else: - save_dir = os.path.expanduser(save_dir) - if not os.path.isabs(save_dir): - save_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), save_dir) - os.makedirs(save_dir, exist_ok=True) - 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" - } - - if not torrent_url.startswith("http"): - filename = torrent_url - else: - filename = torrent_url.split("/")[-1] - - wordlist_base = filename.replace('.torrent', '').replace('.7z', '').replace('.txt', '') - wordlist_uri = f"https://weakpass.com/wordlists/{wordlist_base}" - print(f"[+] Fetching wordlist page: {wordlist_uri}") - r = requests.get(wordlist_uri, headers=headers) - if r.status_code != 200: - print(f"[!] Failed to fetch wordlist page: {wordlist_uri}") - return None - - soup = BeautifulSoup(r.text, "html.parser") - app_div = soup.find("div", id="app") - if not app_div or not app_div.has_attr("data-page"): - print(f"[!] Could not find app data on {wordlist_uri}") - return None - - data_page_val = app_div["data-page"] - if not isinstance(data_page_val, str): - data_page_val = str(data_page_val) - data_page_val = data_page_val.replace('"', '"') - try: - data = json.loads(data_page_val) - wordlist = data.get('props', {}).get('wordlist') - wordlist_id = None - torrent_link_from_data = None - if wordlist: - wordlist_id = wordlist.get('id') - torrent_link_from_data = wordlist.get('torrent_link') - else: - wordlists = data.get('props', {}).get('wordlists') - if isinstance(wordlists, dict) and 'data' in wordlists: - wordlists = wordlists['data'] - if isinstance(wordlists, list): - for wl in wordlists: - if wl.get('torrent_link') == filename or wl.get('name') == filename: - wordlist_id = wl.get('id') - torrent_link_from_data = wl.get('torrent_link') - break - if wordlist_base in wl.get('name', ''): - wordlist_id = wl.get('id') - torrent_link_from_data = wl.get('torrent_link') - break - except Exception as e: - print(f"[!] Failed to parse data-page JSON: {e}") - return None - - if not (torrent_link_from_data and wordlist_id): - print(f"[!] No torrent link or id found in wordlist data for {filename}.") - return None - if not torrent_link_from_data.startswith('http'): - torrent_link = f"https://weakpass.com/download/{wordlist_id}/{torrent_link_from_data}" - else: - torrent_link = torrent_link_from_data - - print(f"[+] Downloading .torrent file from: {torrent_link}") - r2 = requests.get(torrent_link, headers=headers, stream=True) - content_type = r2.headers.get("Content-Type", "") - local_filename = os.path.join(save_dir, filename if filename.endswith('.torrent') else filename + '.torrent') - if r2.status_code == 200 and not content_type.startswith("text/html"): - with open(local_filename, 'wb') as f: - for chunk in r2.iter_content(chunk_size=8192): - if chunk: - f.write(chunk) - print(f"Saved to {local_filename}") - else: - print(f"Failed to download a valid torrent file: {torrent_link}") - try: - html = r2.content.decode(errors="replace") - print("--- Begin HTML Debug Output ---") - print(html[:2000]) - print("--- End HTML Debug Output ---") - except Exception as e: - print(f"Could not decode response for debug: {e}") - return None - - if shutil.which("transmission-cli") is None: - print("[ERROR] transmission-cli is not installed or not in your PATH.") - print("Please install it with: brew install transmission-cli (on macOS) or your package manager.") - print(f"Torrent file saved at {local_filename}, but download will not start until transmission-cli is available.") - return local_filename - - def run_transmission(torrent_file, output_dir): - import subprocess - import glob - print(f"Starting transmission-cli for {torrent_file}...") - try: - proc = subprocess.Popen([ - "transmission-cli", - "-w", output_dir, - torrent_file - ], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, universal_newlines=True) - if proc.stdout is not None: - for line in proc.stdout: - print(line, end='') - proc.wait() - if proc.returncode != 0: - print(f"transmission-cli failed for {torrent_file} (exit {proc.returncode})") - return - else: - print(f"Download complete for {torrent_file}") - sevenz_files = glob.glob(os.path.join(output_dir, '*.7z')) - if not sevenz_files: - print("[i] No .7z files found to extract.") - return - for zfile in sevenz_files: - print(f"[+] Extracting {zfile} ...") - sevenz_bin = shutil.which('7z') or shutil.which('7za') - if not sevenz_bin: - print("[!] 7z or 7za not found in PATH. Please install p7zip-full or 7-zip to extract archives.") - continue - try: - extract_result = subprocess.run([ - sevenz_bin, 'x', '-y', zfile, f'-o{output_dir}' - ], capture_output=True, text=True) - print(extract_result.stdout) - if extract_result.returncode == 0: - print(f"[+] Extraction complete: {zfile}") - else: - print(f"[!] Extraction failed for {zfile}: {extract_result.stderr}") - except Exception as e: - print(f"[!] Error extracting {zfile}: {e}") - except Exception as e: - print(f"Error running transmission-cli: {e}") - - t = threading.Thread(target=run_transmission, args=(local_filename, save_dir)) - t.start() - print(f"transmission-cli launched in background for {local_filename}") - return local_filename - -def weakpass_wordlist_menu(): - fetch_all_weakpass_wordlists_multithreaded() - try: - with open("weakpass_wordlists.json", "r", encoding="utf-8") as f: - all_wordlists = json.load(f) - except Exception as e: - print(f"Failed to load local wordlist cache: {e}") - return - page = 0 - batch_size = 100 - while True: - filtered_wordlists = [wl for wl in all_wordlists if str(wl.get('rank', '')) == '7'] - start_idx = page * batch_size - end_idx = start_idx + batch_size - page_wordlists = filtered_wordlists[start_idx:end_idx] - if not page_wordlists: - print("No more wordlists.") - if page > 0: - page -= 1 - continue - col_width = 45 - cols = 3 - print("\nEach entry shows: [number]. [wordlist name] [effectiveness score] [rank]") - print(f"Available Wordlists (Batch {page+1}):") - rows = (len(page_wordlists) + cols - 1) // cols - lines = [''] * rows - for idx, wl in enumerate(page_wordlists): - col = idx // rows - row = idx % rows - effectiveness = wl.get('effectiveness', wl.get('downloads', '')) - rank = wl.get('rank', '') - entry = f"{start_idx+idx+1:3d}. {wl['name'][:25]:<25} {effectiveness:<8} {rank:<2}" - lines[row] += entry.ljust(col_width) - for line in lines: - print(line) - sel = input("\nEnter the number to download, 'n' for next batch, 'p' for previous, or 'q' to cancel: ") - if sel.lower() == 'q': - print("Returning to menu...") - return - if sel.lower() == 'n': - page += 1 - continue - if sel.lower() == 'p' and page > 0: - page -= 1 - continue - try: - sel_idx = int(sel) - 1 - start_idx - if 0 <= sel_idx < len(page_wordlists): - torrent_url = page_wordlists[sel_idx]['torrent_url'] - download_torrent_file(torrent_url) - else: - print("Invalid selection.") - except Exception as e: - print(f"Error: {e}") diff --git a/tests/test_hashview.py b/tests/test_hashview.py index b6a2a28..f3f87db 100644 --- a/tests/test_hashview.py +++ b/tests/test_hashview.py @@ -12,7 +12,7 @@ from unittest.mock import Mock, patch, MagicMock # Add the parent directory to the path to import hate_crack sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) -from hate_crack.hashview import HashviewAPI +from hate_crack.api import HashviewAPI # Test configuration - these are mock values, not real credentials HASHVIEW_URL = 'https://hashview.example.com' diff --git a/tests/test_hate_crack_utils.py b/tests/test_hate_crack_utils.py deleted file mode 100644 index 1a97250..0000000 --- a/tests/test_hate_crack_utils.py +++ /dev/null @@ -1,38 +0,0 @@ -from pathlib import Path - - -def test_generate_session_id_sanitizes(hc_module): - hc = hc_module - hc.hcatHashFile = "/tmp/my hash@file(1).txt" - assert hc.generate_session_id() == "my_hash_file_1_" - - -def test_line_count(hc_module, tmp_path): - hc = hc_module - test_file = tmp_path / "lines.txt" - test_file.write_text("a\nb\nc\n", encoding="utf-8") - assert hc.lineCount(str(test_file)) == 3 - - -def test_verify_wordlist_dir_resolves(hc_module, tmp_path): - hc = hc_module - directory = tmp_path / "wordlists" - directory.mkdir() - wordlist = directory / "list.txt" - wordlist.write_text("one\n", encoding="utf-8") - assert hc.verify_wordlist_dir(str(directory), "list.txt") == str(wordlist) - - -def test_verify_wordlist_dir_prefers_absolute(hc_module, tmp_path): - hc = hc_module - wordlist = tmp_path / "absolute.txt" - wordlist.write_text("one\n", encoding="utf-8") - assert hc.verify_wordlist_dir("/does/not/matter", str(wordlist)) == str(wordlist) - - -def test_convert_hex(hc_module, tmp_path): - hc = hc_module - data = "$HEX[68656c6c6f]\nplain\n" - infile = tmp_path / "hex.txt" - infile.write_text(data, encoding="utf-8") - assert hc.convert_hex(str(infile)) == ["hello", "plain"] diff --git a/tests/test_menu_snapshots.py b/tests/test_menu_snapshots.py deleted file mode 100644 index af9f6ca..0000000 --- a/tests/test_menu_snapshots.py +++ /dev/null @@ -1,53 +0,0 @@ -import builtins -from pathlib import Path - - -SNAPSHOT_DIR = Path(__file__).resolve().parent / "fixtures" / "menu_outputs" - - -def _snapshot_text(out, err): - return f"STDOUT:\n{out}STDERR:\n{err}" - - -def _assert_snapshot(name, capsys): - captured = capsys.readouterr() - snapshot_path = SNAPSHOT_DIR / f"{name}.txt" - expected = snapshot_path.read_text(encoding="utf-8") - actual = _snapshot_text(captured.out, captured.err) - assert actual == expected - - -def _input_sequence(values): - iterator = iter(values) - - def _fake_input(_prompt=""): - return next(iterator) - - return _fake_input - - -def _setup_globals(hc, tmp_path): - hc.hcatHashType = "1000" - hc.hcatHashFile = "hashes" - hc.hcatHashFileOrig = hc.hcatHashFile - hc.hcatWordlists = str(tmp_path / "wordlists") - hc.hcatOptimizedWordlists = str(tmp_path / "optimized") - hc.hcatPath = str(tmp_path / "hcat") - hc.hcatHybridlist = ["hybrid1.txt", "hybrid2.txt"] - hc.hcatBruteCount = 0 - hc.hcatDictionaryCount = 0 - hc.hcatMaskCount = 0 - hc.hcatFingerprintCount = 0 - hc.hcatCombinationCount = 0 - hc.hcatHybridCount = 0 - hc.hcatExtraCount = 0 - -def test_hybrid_crack_snapshot(hc_module, monkeypatch, tmp_path, capsys): - hc = hc_module - _setup_globals(hc, tmp_path) - monkeypatch.setattr(hc, "hcatHybrid", lambda *args, **kwargs: print("hcatHybrid called")) - monkeypatch.setattr(builtins, "input", _input_sequence([""])) - - hc.hybrid_crack() - _assert_snapshot("hybrid_crack", capsys) - diff --git a/tests/test_module_imports.py b/tests/test_module_imports.py deleted file mode 100644 index d038ce1..0000000 --- a/tests/test_module_imports.py +++ /dev/null @@ -1,31 +0,0 @@ -import importlib - - -def test_import_hashview_module(): - module = importlib.import_module("hate_crack.hashview") - assert hasattr(module, "HashviewAPI") - - -def test_import_hashmob_module(): - module = importlib.import_module("hate_crack.hashmob_wordlist") - assert hasattr(module, "list_and_download_official_wordlists") - - -def test_import_weakpass_module(): - module = importlib.import_module("hate_crack.weakpass") - assert hasattr(module, "weakpass_wordlist_menu") - - -def test_import_cli_module(): - module = importlib.import_module("hate_crack.cli") - assert hasattr(module, "add_common_args") - - -def test_import_api_module(): - module = importlib.import_module("hate_crack.api") - assert hasattr(module, "download_hashes_from_hashview") - - -def test_import_attacks_module(): - module = importlib.import_module("hate_crack.attacks") - assert hasattr(module, "quick_crack")