weakpass tests and migration to api.py

This commit is contained in:
Justin Bollinger
2026-01-26 16:10:58 -05:00
parent 544ee2dba3
commit a46ea5897a
7 changed files with 367 additions and 630 deletions
+1
View File
@@ -1,2 +1,3 @@
config.json
hashcat.pot
hate_crack/__init__.pyc
+28 -238
View File
@@ -85,235 +85,6 @@ def ensure_binary(binary_path, build_dir=None, name=None):
# Weakpass wordlist menu and fetch logic moved here
import threading
from queue import Queue
# --- Weakpass Wordlist Download Logic ---
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."""
import json
from bs4 import BeautifulSoup
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="wordlists"):
import os
from bs4 import BeautifulSoup
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"
}
# Only use id from data-page JSON in <div id="app">
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
import json
data_page_val = app_div["data-page"]
if not isinstance(data_page_val, str):
data_page_val = str(data_page_val)
# Unescape HTML entities
data_page_val = data_page_val.replace('&quot;', '"')
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:
# Try to match in 'wordlists' (list page)
wordlists = data.get('props', {}).get('wordlists')
if isinstance(wordlists, dict) and 'data' in wordlists:
wordlists = wordlists['data']
if isinstance(wordlists, list):
# Try to match by exact filename or base name
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
# Try matching base name
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
import shutil
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}")
import threading
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():
from hate_crack.weakpass import weakpass_wordlist_menu as _weakpass_wordlist_menu
return _weakpass_wordlist_menu()
import os
import random
import re
@@ -325,8 +96,8 @@ import glob
import subprocess
import argparse
from hate_crack.weakpass import fetch_all_weakpass_wordlists_multithreaded, download_torrent_file, weakpass_wordlist_menu
from hate_crack.hashview import HashviewAPI
from hate_crack.api import fetch_all_weakpass_wordlists_multithreaded, download_torrent_file, weakpass_wordlist_menu
from hate_crack.api import HashviewAPI
from hate_crack.api import (
download_all_weakpass_torrents,
download_hashes_from_hashview,
@@ -335,7 +106,6 @@ from hate_crack.api import (
)
from hate_crack.cli import (
add_common_args,
apply_config_overrides,
resolve_path,
setup_logging,
)
@@ -429,7 +199,22 @@ def verify_wordlist_dir(directory, wordlist):
return directory + '/' + wordlist
else:
print('Invalid path for {0}. Please check configuration and try again.'.format(wordlist))
quit(1)
response = input(f"Wordlist '{wordlist}' not found. Would you like to download rockyou.txt automatically? (Y/n): ").strip().lower()
if response in ('', 'y', 'yes'):
try:
import urllib.request
rockyou_url = "https://weakpass.com/download/90/rockyou.txt.gz"
dest_path = os.path.join(directory, "rockyou.txt.gz")
print(f"Downloading rockyou.txt.gz to {dest_path} ...")
urllib.request.urlretrieve(rockyou_url, dest_path)
print("Download complete.")
return dest_path
except Exception as e:
print(f"Failed to download rockyou.txt.gz: {e}")
return None
else:
print('Exiting. Please check configuration and try again.')
return None
if not SKIP_INIT:
# hashcat biniary checks for systems that install hashcat binary in different location than the rest of the hashcat files
@@ -1976,9 +1761,10 @@ def main():
parser.add_argument('--download-torrent', metavar='FILENAME', help='Download a specific Weakpass torrent file')
parser.add_argument('--download-all-torrents', action='store_true', help='Download all available Weakpass torrents from cache')
parser.add_argument('--weakpass', action='store_true', help='Download wordlists from Weakpass')
parser.add_argument('--rank', type=int, default=-1, help='Only show wordlists with this rank (use 0 to show all, default: >4)')
parser.add_argument('--hashmob', action='store_true', help='Download wordlists from Hashmob.net')
parser.add_argument('--debug', action='store_true', help='Enable debug mode')
add_common_args(parser)
# Removed add_common_args(parser) since config items are now only set via config file
args = parser.parse_args()
global debug_mode
@@ -1999,8 +1785,7 @@ def main():
maxruntime=maxruntime,
bandrelbasewords=bandrelbasewords,
)
apply_config_overrides(args, config=config)
hashview_url = config.hashview_url
hashview_api_key = config.hashview_api_key
hcatPath = config.hcatPath
@@ -2030,8 +1815,13 @@ def main():
sys.exit(1)
sys.exit(0)
if args.weakpass:
weakpass_wordlist_menu()
weakpass_wordlist_menu(rank=args.rank)
sys.exit(0)
if args.hashmob:
download_hashmob_wordlists(print_fn=print)
sys.exit(0)
if args.hashfile and args.hashtype:
@@ -2073,7 +1863,7 @@ def main():
print(f"\n✗ Error downloading hashes: {str(e)}")
sys.exit(1)
elif choice == '2' or args.weakpass:
weakpass_wordlist_menu()
weakpass_wordlist_menu(rank=args.rank)
sys.exit(0)
elif choice == '3' or args.hashmob:
download_hashmob_wordlists(print_fn=print)
+305 -50
View File
@@ -1,14 +1,12 @@
import json
import os
from typing import Callable, Tuple
import threading
from queue import Queue
import shutil
from typing import Callable, Tuple
import requests
from bs4 import BeautifulSoup
import shutil
def check_7z():
import shutil
@@ -122,6 +120,7 @@ def fetch_all_weakpass_wordlists_multithreaded(total_pages=67, threads=10, outpu
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:
@@ -133,6 +132,7 @@ def download_torrent_file(torrent_url, save_dir=None):
"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"
}
# Resolve a filename even if a URL is provided.
if not torrent_url.startswith("http"):
filename = torrent_url
else:
@@ -265,7 +265,7 @@ def download_torrent_file(torrent_url, save_dir=None):
print(f"transmission-cli launched in background for {local_filename}")
return local_filename
def weakpass_wordlist_menu():
def weakpass_wordlist_menu(rank=-1):
fetch_all_weakpass_wordlists_multithreaded()
try:
with open("weakpass_wordlists.json", "r", encoding="utf-8") as f:
@@ -273,53 +273,62 @@ def weakpass_wordlist_menu():
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 rank == 0:
filtered_wordlists = all_wordlists
elif rank > 0:
filtered_wordlists = [wl for wl in all_wordlists if str(wl.get('rank', '')) == str(rank)]
else:
# Default: show all with rank > 4
filtered_wordlists = [wl for wl in all_wordlists if str(wl.get('rank', '')) > '4']
col_width = 45
cols = 3
print("\nEach entry shows: [number]. [wordlist name] [effectiveness score] [rank]")
print(f"Available Wordlists:")
rows = (len(filtered_wordlists) + cols - 1) // cols
lines = [''] * rows
for idx, wl in enumerate(filtered_wordlists):
col = idx // rows
row = idx % rows
effectiveness = wl.get('effectiveness', wl.get('downloads', ''))
rank = wl.get('rank', '')
entry = f"{idx+1:3d}. {wl['name'][:25]:<25} {effectiveness:<8} {rank:<2}"
lines[row] += entry.ljust(col_width)
for line in lines:
print(line)
try:
sel = input("\nEnter the number(s) to download (e.g. 1,3,5-7) 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
# Parse comma- and dash-separated values
def parse_indices(selection):
indices = set()
for part in selection.split(','):
part = part.strip()
if '-' in part:
try:
start, end = map(int, part.split('-', 1))
indices.update(range(start, end + 1))
except Exception:
continue
else:
try:
indices.add(int(part))
except Exception:
continue
return sorted(i for i in indices if 1 <= i <= len(filtered_wordlists))
indices = parse_indices(sel)
if not indices:
print("No valid selection.")
return
for idx in indices:
torrent_url = filtered_wordlists[idx - 1]['torrent_url']
download_torrent_file(torrent_url)
except KeyboardInterrupt:
print("\nKeyboard interrupt: Returning to main menu...")
return
except Exception as e:
print(f"Error: {e}")
# Hashview Integration - Real API implementation matching hate_crack.py
class HashviewAPI:
@@ -541,6 +550,252 @@ def download_hashes_from_hashview(
return hcat_hash_file, hcat_hash_type
def sanitize_filename(filename):
"""Sanitize a filename by replacing spaces and removing problematic characters."""
import re
filename = filename.replace(' ', '_')
filename = re.sub(r'[^A-Za-z0-9._-]', '', filename)
return filename
def get_hashmob_api_key():
"""Return hashmob_api_key from config.json in package or project root."""
pkg_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.abspath(os.path.join(pkg_dir, os.pardir))
for cfg in (os.path.join(pkg_dir, 'config.json'), os.path.join(project_root, 'config.json')):
if os.path.isfile(cfg):
try:
with open(cfg) as f:
config = json.load(f)
key = config.get('hashmob_api_key')
if key:
return key
except Exception:
continue
return None
def download_hashmob_wordlist_list():
"""Fetch available wordlists from Hashmob API v2 and print them."""
url = "https://hashmob.net/api/v2/resource"
api_key = get_hashmob_api_key()
headers = {"api-key": api_key} if api_key else {}
try:
resp = requests.get(url, headers=headers, timeout=30)
resp.raise_for_status()
data = resp.json()
wordlists = [r for r in data if r.get('type') == 'wordlist']
print("Available Hashmob Wordlists:")
for idx, wl in enumerate(wordlists):
print(f"{idx+1}. {wl.get('name', wl.get('file_name', ''))} - {wl.get('information', '')}")
return wordlists
except Exception as e:
print(f"Error fetching Hashmob wordlists: {e}")
return []
def download_hashmob_wordlist(file_name, out_path):
"""Download a wordlist file from Hashmob by file name."""
url = f"https://hashmob.net/api/v2/downloads/research/wordlists/{file_name}"
api_key = get_hashmob_api_key()
headers = {"api-key": api_key} if api_key else {}
try:
with requests.get(url, headers=headers, stream=True, timeout=60, allow_redirects=True) as r:
if r.status_code in (301, 302, 303, 307, 308):
redirect_url = r.headers.get('Location')
if redirect_url:
print(f"Following redirect to: {redirect_url}")
return download_hashmob_wordlist(redirect_url, out_path)
print("Redirect with no Location header!")
return False
r.raise_for_status()
content_type = r.headers.get('Content-Type', '')
if 'text/plain' in content_type:
html = r.content.decode(errors='replace')
import re
match = re.search(
r"<meta[^>]+http-equiv=['\"]refresh['\"][^>]+content=['\"]0;url=([^'\"]+)['\"]",
html,
re.IGNORECASE
)
if match:
real_url = match.group(1)
print(f"Found meta refresh redirect to: {real_url}")
with requests.get(real_url, stream=True, timeout=120) as r2:
r2.raise_for_status()
with open(out_path, 'wb') as f:
for chunk in r2.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
print(f"Downloaded {out_path}")
return True
print("Error: Received HTML instead of file. Possible permission or quota issue.")
return False
with open(out_path, 'wb') as f:
for chunk in r.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
print(f"Downloaded {out_path}")
return True
except Exception as e:
print(f"Error downloading wordlist: {e}")
return False
def list_official_wordlists():
"""List files in the official wordlists directory via the Hashmob API."""
url = "https://hashmob.net/api/v2/downloads/research/official/"
api_key = get_hashmob_api_key()
headers = {"api-key": api_key} if api_key else {}
try:
resp = requests.get(url, headers=headers, timeout=30)
resp.raise_for_status()
try:
data = resp.json()
print("Official Hashmob Wordlists (JSON):")
for idx, entry in enumerate(data):
print(f"{idx+1}. {entry}")
return data
except Exception:
print("Official Hashmob Wordlists (raw text):")
print(resp.text)
return resp.text
except Exception as e:
print(f"Error listing official wordlists: {e}")
return []
def list_and_download_official_wordlists():
"""List files in the official wordlists directory via the Hashmob API, prompt for selection, and download."""
url = "https://hashmob.net/api/v2/downloads/research/official/"
try:
resp = requests.get(url, timeout=30)
resp.raise_for_status()
data = resp.json()
if not isinstance(data, list):
print("Unexpected response format. Raw output:")
print(data)
return
print("Official Hashmob Wordlists:")
for idx, entry in enumerate(data):
name = entry.get('name', entry.get('file_name', str(entry)))
file_name = entry.get('file_name', name)
print(f"{idx+1}. {name} ({file_name})")
print("a. Download ALL files")
sel = input("Enter the number of the wordlist to download, or 'a' for all, or 'q' to quit: ")
if sel.lower() == 'q':
return
if sel.lower() == 'a':
try:
for entry in data:
file_name = entry.get('file_name')
if not file_name:
print("No file_name found for an entry, skipping.")
continue
out_path = entry.get('file_name', file_name)
if download_official_wordlist(file_name, out_path):
dest_dir = get_hcat_wordlists_dir()
archive_path = os.path.join(dest_dir, out_path) if not os.path.isabs(out_path) else out_path
if archive_path.endswith('.7z'):
extract_with_7z(archive_path)
except KeyboardInterrupt:
print("\nKeyboard interrupt: Returning to download menu...")
return
return
try:
idx = int(sel) - 1
if idx < 0 or idx >= len(data):
print("Invalid selection.")
return
file_name = data[idx].get('file_name')
if not file_name:
print("No file_name found for selection.")
return
out_path = data[idx].get('file_name', file_name)
if download_official_wordlist(file_name, out_path):
dest_dir = get_hcat_wordlists_dir()
archive_path = os.path.join(dest_dir, out_path) if not os.path.isabs(out_path) else out_path
if archive_path.endswith('.7z'):
extract_with_7z(archive_path)
except Exception as e:
print(f"Error: {e}")
except Exception as e:
print(f"Error listing official wordlists: {e}")
def download_official_wordlist(file_name, out_path):
"""Download a file from the official wordlists directory with a progress bar."""
import sys
url = f"https://hashmob.net/api/v2/downloads/research/official/{file_name}"
try:
with requests.get(url, stream=True, timeout=120) as r:
r.raise_for_status()
try:
total = int(r.headers.get('content-length') or 0)
except Exception:
total = 0
downloaded = 0
chunk_size = 8192
out_path = sanitize_filename(file_name)
dest_dir = get_hcat_wordlists_dir()
archive_path = os.path.join(dest_dir, out_path) if not os.path.isabs(out_path) else out_path
os.makedirs(os.path.dirname(archive_path), exist_ok=True)
with open(archive_path, 'wb') as f:
for chunk in r.iter_content(chunk_size=chunk_size):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if total:
done = int(50 * downloaded / total)
percent = 100 * downloaded / total
bar = '=' * done + ' ' * (50 - done)
sys.stdout.write(
f"\r[{bar}] {percent:6.2f}% ({downloaded // 1024} KB/{total // 1024} KB)"
)
sys.stdout.flush()
else:
sys.stdout.write(f"\rDownloaded {downloaded // 1024} KB")
sys.stdout.flush()
sys.stdout.write("\n")
print(f"Downloaded {archive_path}")
if archive_path.endswith('.7z'):
extract_with_7z(archive_path)
return True
except Exception as e:
print(f"Error downloading official wordlist: {e}")
return False
def extract_with_7z(archive_path, output_dir=None):
"""Extract a .7z archive using the 7z or 7za command."""
import subprocess
if output_dir is None:
output_dir = os.path.splitext(archive_path)[0]
os.makedirs(output_dir, exist_ok=True)
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.")
return False
try:
print(f"Extracting {archive_path} to {output_dir} ...")
result = subprocess.run(
[sevenz_bin, 'x', '-y', archive_path, f'-o{output_dir}'],
capture_output=True,
text=True
)
print(result.stdout)
if result.returncode == 0:
print(f"[+] Extraction complete: {archive_path}")
return True
print(f"[!] Extraction failed for {archive_path}: {result.stderr}")
return False
except Exception as e:
print(f"[!] Error extracting {archive_path}: {e}")
return False
def download_hashmob_wordlists(print_fn=print) -> None:
"""Download official Hashmob wordlists."""
list_and_download_official_wordlists()
+1 -31
View File
@@ -10,38 +10,8 @@ def resolve_path(value: Optional[str]) -> Optional[str]:
return os.path.abspath(os.path.expanduser(value))
def apply_config_overrides(args, config):
"""Apply CLI config overrides to the provided config object."""
if args.hashview_url:
config.hashview_url = args.hashview_url
if args.hashview_api_key is not None:
config.hashview_api_key = args.hashview_api_key
if args.hcat_path:
config.hcatPath = resolve_path(args.hcat_path)
if args.hcat_bin:
config.hcatBin = args.hcat_bin
if args.wordlists_dir:
config.hcatWordlists = resolve_path(args.wordlists_dir)
if args.optimized_wordlists_dir:
config.hcatOptimizedWordlists = resolve_path(args.optimized_wordlists_dir)
if args.pipal_path:
config.pipalPath = resolve_path(args.pipal_path)
if args.maxruntime is not None:
config.maxruntime = args.maxruntime
if args.bandrel_basewords:
config.bandrelbasewords = args.bandrel_basewords
def add_common_args(parser) -> None:
parser.add_argument('--hashview-url', dest='hashview_url', help='Override Hashview URL')
parser.add_argument('--hashview-api-key', dest='hashview_api_key', help='Override Hashview API key')
parser.add_argument('--hcat-path', dest='hcat_path', help='Override hashcat path')
parser.add_argument('--hcat-bin', dest='hcat_bin', help='Override hashcat binary name')
parser.add_argument('--wordlists-dir', dest='wordlists_dir', help='Override wordlists directory')
parser.add_argument('--optimized-wordlists-dir', dest='optimized_wordlists_dir', help='Override optimized wordlists directory')
parser.add_argument('--pipal-path', dest='pipal_path', help='Override pipal path')
parser.add_argument('--maxruntime', type=int, help='Override max runtime setting')
parser.add_argument('--bandrel-basewords', dest='bandrel_basewords', help='Override bandrel basewords setting')
pass # All config items are now set via config file only
def setup_logging(logger: logging.Logger, hate_path: str, debug_mode: bool) -> None:
-283
View File
@@ -1,283 +0,0 @@
import json
import os
import requests
def sanitize_filename(filename):
"""Sanitize a filename by replacing spaces and removing problematic characters."""
import re
filename = filename.replace(' ', '_')
filename = re.sub(r'[^A-Za-z0-9._-]', '', filename)
return filename
def get_api_key():
"""Return hashmob_api_key from config.json in package or project root."""
pkg_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.abspath(os.path.join(pkg_dir, os.pardir))
for cfg in (os.path.join(pkg_dir, 'config.json'), os.path.join(project_root, 'config.json')):
if os.path.isfile(cfg):
try:
with open(cfg) as f:
config = json.load(f)
key = config.get('hashmob_api_key')
if key:
return key
except Exception:
continue
return None
def get_hcat_wordlists_dir():
"""Return the configured `hcatWordlists` directory from config.json.
Checks both the package directory and the project root for config.json.
Falls back to project_root/wordlists when not configured.
"""
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 download_hashmob_wordlist_list():
"""Fetch available wordlists from Hashmob API v2 and print them."""
url = "https://hashmob.net/api/v2/resource"
api_key = get_api_key()
headers = {"api-key": api_key} if api_key else {}
try:
resp = requests.get(url, headers=headers, timeout=30)
resp.raise_for_status()
data = resp.json()
wordlists = [r for r in data if r.get('type') == 'wordlist']
print("Available Hashmob Wordlists:")
for idx, wl in enumerate(wordlists):
print(f"{idx+1}. {wl.get('name', wl.get('file_name', ''))} - {wl.get('information', '')}")
return wordlists
except Exception as e:
print(f"Error fetching Hashmob wordlists: {e}")
return []
def download_hashmob_wordlist(file_name, out_path):
"""Download a wordlist file from Hashmob by file name."""
url = f"https://hashmob.net/api/v2/downloads/research/wordlists/{file_name}"
api_key = get_api_key()
headers = {"api-key": api_key} if api_key else {}
try:
with requests.get(url, headers=headers, stream=True, timeout=60, allow_redirects=True) as r:
if r.status_code in (301, 302, 303, 307, 308):
redirect_url = r.headers.get('Location')
if redirect_url:
print(f"Following redirect to: {redirect_url}")
return download_hashmob_wordlist(redirect_url, out_path)
print("Redirect with no Location header!")
return False
r.raise_for_status()
content_type = r.headers.get('Content-Type', '')
if 'text/plain' in content_type:
html = r.content.decode(errors='replace')
import re
match = re.search(
r"<meta[^>]+http-equiv=['\"]refresh['\"][^>]+content=['\"]0;url=([^'\"]+)['\"]",
html,
re.IGNORECASE
)
if match:
real_url = match.group(1)
print(f"Found meta refresh redirect to: {real_url}")
with requests.get(real_url, stream=True, timeout=120) as r2:
r2.raise_for_status()
with open(out_path, 'wb') as f:
for chunk in r2.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
print(f"Downloaded {out_path}")
return True
print("Error: Received HTML instead of file. Possible permission or quota issue.")
return False
with open(out_path, 'wb') as f:
for chunk in r.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
print(f"Downloaded {out_path}")
return True
except Exception as e:
print(f"Error downloading wordlist: {e}")
return False
def list_official_wordlists():
"""List files in the official wordlists directory via the Hashmob API."""
url = "https://hashmob.net/api/v2/downloads/research/official/"
api_key = get_api_key()
headers = {"api-key": api_key} if api_key else {}
try:
resp = requests.get(url, headers=headers, timeout=30)
resp.raise_for_status()
try:
data = resp.json()
print("Official Hashmob Wordlists (JSON):")
for idx, entry in enumerate(data):
print(f"{idx+1}. {entry}")
return data
except Exception:
print("Official Hashmob Wordlists (raw text):")
print(resp.text)
return resp.text
except Exception as e:
print(f"Error listing official wordlists: {e}")
return []
def list_and_download_official_wordlists():
"""List files in the official wordlists directory via the Hashmob API, prompt for selection, and download."""
url = "https://hashmob.net/api/v2/downloads/research/official/"
api_key = get_api_key()
headers = {"api-key": api_key} if api_key else {}
try:
resp = requests.get(url, headers=headers, timeout=30)
resp.raise_for_status()
data = resp.json()
if not isinstance(data, list):
print("Unexpected response format. Raw output:")
print(data)
return
print("Official Hashmob Wordlists:")
for idx, entry in enumerate(data):
name = entry.get('name', entry.get('file_name', str(entry)))
file_name = entry.get('file_name', name)
info = entry.get('information', '')
print(f"{idx+1}. {name} ({file_name}) - {info}")
print("a. Download ALL files")
sel = input("Enter the number of the wordlist to download, or 'a' for all, or 'q' to quit: ")
if sel.lower() == 'q':
return
if sel.lower() == 'a':
for entry in data:
file_name = entry.get('file_name')
if not file_name:
print("No file_name found for an entry, skipping.")
continue
out_path = entry.get('name', file_name)
if download_official_wordlist(file_name, out_path):
dest_dir = get_hcat_wordlists_dir()
archive_path = os.path.join(dest_dir, out_path) if not os.path.isabs(out_path) else out_path
if archive_path.endswith('.7z'):
extract_with_7z(archive_path)
return
try:
idx = int(sel) - 1
if idx < 0 or idx >= len(data):
print("Invalid selection.")
return
file_name = data[idx].get('file_name')
if not file_name:
print("No file_name found for selection.")
return
out_path = data[idx].get('name', file_name)
if download_official_wordlist(file_name, out_path):
dest_dir = get_hcat_wordlists_dir()
archive_path = os.path.join(dest_dir, out_path) if not os.path.isabs(out_path) else out_path
if archive_path.endswith('.7z'):
extract_with_7z(archive_path)
except Exception as e:
print(f"Error: {e}")
except Exception as e:
print(f"Error listing official wordlists: {e}")
def download_official_wordlist(file_name, out_path):
"""Download a file from the official wordlists directory with a progress bar."""
import sys
url = f"https://hashmob.net/api/v2/downloads/research/official/{file_name}"
api_key = get_api_key()
headers = {"api-key": api_key} if api_key else {}
try:
with requests.get(url, headers=headers, stream=True, timeout=120) as r:
r.raise_for_status()
try:
total = int(r.headers.get('content-length') or 0)
except Exception:
total = 0
downloaded = 0
chunk_size = 8192
out_path = sanitize_filename(out_path)
dest_dir = get_hcat_wordlists_dir()
archive_path = os.path.join(dest_dir, out_path) if not os.path.isabs(out_path) else out_path
os.makedirs(os.path.dirname(archive_path), exist_ok=True)
with open(archive_path, 'wb') as f:
for chunk in r.iter_content(chunk_size=chunk_size):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if total:
done = int(50 * downloaded / total)
percent = 100 * downloaded / total
bar = '=' * done + ' ' * (50 - done)
sys.stdout.write(
f"\r[{bar}] {percent:6.2f}% ({downloaded // 1024} KB/{total // 1024} KB)"
)
sys.stdout.flush()
else:
sys.stdout.write(f"\rDownloaded {downloaded // 1024} KB")
sys.stdout.flush()
sys.stdout.write("\n")
print(f"Downloaded {archive_path}")
if archive_path.endswith('.7z'):
extract_with_7z(archive_path)
return True
except Exception as e:
print(f"Error downloading official wordlist: {e}")
return False
def extract_with_7z(archive_path, output_dir=None):
"""Extract a .7z archive using the 7z or 7za command."""
import shutil
import subprocess
if output_dir is None:
output_dir = os.path.splitext(archive_path)[0]
os.makedirs(output_dir, exist_ok=True)
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.")
return False
try:
print(f"Extracting {archive_path} to {output_dir} ...")
result = subprocess.run(
[sevenz_bin, 'x', '-y', archive_path, f'-o{output_dir}'],
capture_output=True,
text=True
)
print(result.stdout)
if result.returncode == 0:
print(f"[+] Extraction complete: {archive_path}")
return True
print(f"[!] Extraction failed for {archive_path}: {result.stderr}")
return False
except Exception as e:
print(f"[!] Error extracting {archive_path}: {e}")
return False
+14
View File
@@ -0,0 +1,14 @@
import sys
import pytest
def test_cli_weakpass_exits(hc_module, monkeypatch, capsys):
hc = hc_module
monkeypatch.setattr(hc, "weakpass_wordlist_menu", lambda **kwargs: print("weakpass_wordlist_menu called"))
monkeypatch.setattr(sys, "argv", ["hate_crack.py", "--weakpass"])
with pytest.raises(SystemExit) as excinfo:
hc.main()
assert excinfo.value.code == 0
captured = capsys.readouterr()
assert "weakpass_wordlist_menu called" in captured.out
+18 -28
View File
@@ -1,31 +1,21 @@
def test_hashmob_connectivity_mocked(monkeypatch, capsys):
from hate_crack import hashmob_wordlist as hm
import re
import pytest
from hate_crack.api import download_hashmob_wordlist_list
class FakeResp:
def raise_for_status(self):
return None
def json(self):
return [
{"type": "wordlist", "name": "List A", "information": "Info A"},
{"type": "wordlist", "name": "List B", "information": "Info B"},
{"type": "other", "name": "Ignore", "information": "Nope"},
]
def fake_get(url, headers=None, timeout=None):
assert url == "https://hashmob.net/api/v2/resource"
assert headers == {"api-key": "test-key"}
return FakeResp()
monkeypatch.setattr(hm, "get_api_key", lambda: "test-key")
monkeypatch.setattr(hm.requests, "get", fake_get)
result = hm.download_hashmob_wordlist_list()
assert len(result) == 2
assert result[0]["name"] == "List A"
assert result[1]["name"] == "List B"
def test_hashmob_connectivity_real(capsys):
try:
result = download_hashmob_wordlist_list()
except Exception as e:
pytest.skip(f"Network or API unavailable: {e}")
assert isinstance(result, list)
assert any('name' in wl for wl in result)
captured = capsys.readouterr()
assert "Available Hashmob Wordlists:" in captured.out
assert "List A" in captured.out
assert "List B" in captured.out
# Check for at least one wordlist name in output using regex
names = [wl['name'] for wl in result if 'name' in wl]
found = False
for name in names:
if re.search(re.escape(name), captured.out):
found = True
break
assert found, "No wordlist name found in output"