mirror of
https://github.com/trustedsec/hate_crack.git
synced 2026-07-28 22:51:14 -07:00
Merge pull request #56 from trustedsec/wordlist_downloaders
Wordlist downloaders
This commit is contained in:
Justin Bollinger
@@ -1,2 +1,3 @@
|
||||
config.json
|
||||
hashcat.pot
|
||||
hate_crack/__init__.pyc
|
||||
|
||||
+20
-25
@@ -20,12 +20,6 @@ make install
|
||||
```git clone https://github.com/trustedsec/hate_crack.git```
|
||||
* Customize binary and wordlist paths in "config.json"
|
||||
* Make sure that at least "rockyou.txt" is within your "wordlists" path
|
||||
|
||||
```$ python wordlist_optimizer.py
|
||||
usage: python wordlist_optimizer.py <input file list> <output directory>
|
||||
|
||||
$ python wordlist_optimizer.py wordlists.txt ../optimized_wordlists
|
||||
```
|
||||
-------------------------------------------------------------------
|
||||
## Project Structure
|
||||
Core logic is now split into modules under `hate_crack/`:
|
||||
@@ -48,7 +42,6 @@ usage: python hate_crack.py <hash_file> <hash_type> [options]
|
||||
```
|
||||
|
||||
Common options:
|
||||
- `--task <N>`: Run a specific menu task number (e.g., 1, 93).
|
||||
- `--download-hashview`: Download hashes from Hashview before cracking.
|
||||
- `--weakpass`: Download wordlists from Weakpass.
|
||||
- `--hashmob`: Download wordlists from Hashmob.net.
|
||||
@@ -120,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:
|
||||
```
|
||||
+60
-277
@@ -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('"', '"')
|
||||
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,10 +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('--task', metavar='TASK', help='Run a menu task by number (e.g., 1, 93)')
|
||||
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
|
||||
@@ -2000,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
|
||||
@@ -2031,13 +1815,14 @@ 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.task and not (args.hashfile and args.hashtype) and not args.download_hashview:
|
||||
print("Error: --task requires <hashfile> <hashtype> or --download-hashview to supply hashes.")
|
||||
sys.exit(1)
|
||||
if args.hashmob:
|
||||
download_hashmob_wordlists(print_fn=print)
|
||||
sys.exit(0)
|
||||
|
||||
if args.hashfile and args.hashtype:
|
||||
hcatHashFile = resolve_path(args.hashfile)
|
||||
@@ -2078,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)
|
||||
@@ -2150,34 +1935,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
|
||||
}
|
||||
if args.task:
|
||||
task = str(args.task).strip()
|
||||
if task not in options:
|
||||
print(f"Invalid task: {task}. Valid options are: {', '.join(sorted(options.keys(), key=int))}")
|
||||
sys.exit(1)
|
||||
options[task]()
|
||||
sys.exit(0)
|
||||
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")
|
||||
@@ -2192,13 +1973,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]()
|
||||
|
||||
+745
-2
@@ -1,9 +1,506 @@
|
||||
import json
|
||||
import os
|
||||
import threading
|
||||
from queue import Queue
|
||||
import shutil
|
||||
from typing import Callable, Tuple
|
||||
|
||||
from hate_crack.hashview import HashviewAPI
|
||||
from hate_crack.hashmob_wordlist import list_and_download_official_wordlists
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
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"
|
||||
}
|
||||
|
||||
# Resolve a filename even if a URL is provided.
|
||||
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(rank=-1):
|
||||
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
|
||||
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
|
||||
# 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:
|
||||
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(
|
||||
@@ -53,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
@@ -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:
|
||||
|
||||
@@ -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
|
||||
@@ -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()
|
||||
|
||||
@@ -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}")
|
||||
@@ -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
|
||||
@@ -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"
|
||||
|
||||
@@ -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'
|
||||
|
||||
@@ -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"]
|
||||
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user