cleanup of modules

This commit is contained in:
Justin Bollinger
2026-01-26 14:29:39 -05:00
parent 17e8298f07
commit 544ee2dba3
9 changed files with 544 additions and 677 deletions
+20 -18
View File
@@ -113,25 +113,27 @@ Tests automatically run on GitHub Actions for every push and pull request. The w
-------------------------------------------------------------------
(1) Quick Crack
(2) Extensive Pure_Hate Methodology Crack
(3) Brute Force Attack
(4) Top Mask Attack
(5) Fingerprint Attack
(6) Combinator Attack
(7) Hybrid Attack
(8) Pathwell Top 100 Mask Brute Force Crack
(9) PRINCE Attack
(10) YOLO Combinator Attack
(11) Middle Combinator Attack
(12) Thorough Combinator Attack
(13) Bandrel Methodology
(1) Quick Crack
(2) Extensive Pure_Hate Methodology Crack
(3) Brute Force Attack
(4) Top Mask Attack
(5) Fingerprint Attack
(6) Combinator Attack
(7) Hybrid Attack
(8) Pathwell Top 100 Mask Brute Force Crack
(9) PRINCE Attack
(10) YOLO Combinator Attack
(11) Middle Combinator Attack
(12) Thorough Combinator Attack
(13) Bandrel Methodology
(95) Analyze hashes with Pipal
(96) Export Output to Excel Format
(97) Display Cracked Hashes
(98) Display README
(99) Quit
(93) Download Wordlists
(94) Hashview Integration
(95) Analyze hashes with Pipal
(96) Export Output to Excel Format
(97) Display Cracked Hashes
(98) Display README
(99) Quit
Select a task:
```
+33 -28
View File
@@ -2145,27 +2145,30 @@ def main():
# Display Options
try:
options = {"1": quick_crack,
"2": extensive_crack,
"3": brute_force_crack,
"4": top_mask_crack,
"5": fingerprint_crack,
"6": combinator_crack,
"7": hybrid_crack,
"8": pathwell_crack,
"9": prince_attack,
"10": yolo_combination,
"11": middle_combinator,
"12": thorough_combinator,
"13": bandrel_method,
"93": weakpass_wordlist_menu,
"94": hashview_api,
"95": pipal,
"96": export_excel,
"97": show_results,
"98": show_readme,
"99": quit_hc
}
options = {
"1": quick_crack,
"2": extensive_crack,
"3": brute_force_crack,
"4": top_mask_crack,
"5": fingerprint_crack,
"6": combinator_crack,
"7": hybrid_crack,
"8": pathwell_crack,
"9": prince_attack,
"10": yolo_combination,
"11": middle_combinator,
"12": thorough_combinator,
"13": bandrel_method,
"92": weakpass_wordlist_menu,
"93": download_hashmob_wordlists,
"94": weakpass_wordlist_menu,
"95": hashview_api,
"96": pipal,
"97": export_excel,
"98": show_results,
"99": show_readme,
"100": quit_hc
}
while 1:
print("\n\t(1) Quick Crack")
print("\t(2) Extensive Pure_Hate Methodology Crack")
@@ -2180,13 +2183,15 @@ def main():
print("\t(11) Middle Combinator Attack")
print("\t(12) Thorough Combinator Attack")
print("\t(13) Bandrel Methodology")
print("\n\t(93) Download wordlists from Weakpass")
print("\t(94) Hashview")
print("\t(95) Analyze hashes with Pipal")
print("\t(96) Export Output to Excel Format")
print("\t(97) Display Cracked Hashes")
print("\t(98) Display README")
print("\t(99) Quit")
print("\n\t(92) Download wordlists from Weakpass")
print("\t(93) Download wordlists from Hashmob.net")
print("\t(94) Weakpass Wordlist Menu")
print("\t(95) Hashview API")
print("\t(96) Analyze hashes with Pipal")
print("\t(97) Export Output to Excel Format")
print("\t(98) Display Cracked Hashes")
print("\t(99) Display README")
print("\t(100) Quit")
try:
task = input("\nSelect a task: ")
options[task]()
+490 -2
View File
@@ -2,8 +2,496 @@ import json
import os
from typing import Callable, Tuple
from hate_crack.hashview import HashviewAPI
from hate_crack.hashmob_wordlist import list_and_download_official_wordlists
import threading
from queue import Queue
import requests
from bs4 import BeautifulSoup
import shutil
def check_7z():
import shutil
if shutil.which('7z') or shutil.which('7za'):
return True
print("\n[!] 7z (or 7za) is missing.")
print("To install on macOS: brew install p7zip")
print("To install on Ubuntu/Debian: sudo apt-get install p7zip-full")
print("Please install 7z and try again.")
return False
def check_transmission_cli():
import shutil
if shutil.which('transmission-cli'):
return True
print("\n[!] transmission-cli is missing.")
print("To install on macOS: brew install transmission-cli")
print("To install on Ubuntu/Debian: sudo apt-get install transmission-cli")
print("Please install transmission-cli and try again.")
return False
def get_hcat_wordlists_dir():
pkg_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.abspath(os.path.join(pkg_dir, os.pardir))
candidates = [
os.path.join(pkg_dir, 'config.json'),
os.path.join(project_root, 'config.json')
]
default = os.path.join(project_root, 'wordlists')
for config_path in candidates:
try:
if os.path.isfile(config_path):
with open(config_path) as f:
config = json.load(f)
path = config.get('hcatWordlists')
if path:
path = os.path.expanduser(path)
if not os.path.isabs(path):
path = os.path.join(project_root, path)
os.makedirs(path, exist_ok=True)
return path
except Exception:
continue
os.makedirs(default, exist_ok=True)
return default
def fetch_all_weakpass_wordlists_multithreaded(total_pages=67, threads=10, output_file="weakpass_wordlists.json"):
wordlists = []
lock = threading.Lock()
q = Queue()
headers = {"User-Agent": "Mozilla/5.0"}
def worker():
while True:
page = q.get()
if page is None:
break
try:
url = f"https://weakpass.com/wordlists?page={page}"
r = requests.get(url, headers=headers, timeout=30)
soup = BeautifulSoup(r.text, "html.parser")
app_div = soup.find("div", id="app")
if not app_div or not app_div.has_attr("data-page"):
q.task_done()
continue
data_page_val = app_div["data-page"]
if not isinstance(data_page_val, str):
data_page_val = str(data_page_val)
data = json.loads(data_page_val)
wordlists_data = data.get("props", {}).get("wordlists", {})
if isinstance(wordlists_data, dict) and 'data' in wordlists_data:
wordlists_data = wordlists_data['data']
with lock:
for wl in wordlists_data:
wordlists.append({
"name": wl.get("name", ""),
"size": wl.get("size", ""),
"rank": wl.get("rank", ""),
"downloads": wl.get("downloaded", ""),
"torrent_url": wl.get("torrent_link", "")
})
except Exception as e:
print(f"Error fetching page {page}: {e}")
q.task_done()
for page in range(1, total_pages + 1):
q.put(page)
threads_list = []
for _ in range(threads):
t = threading.Thread(target=worker)
t.start()
threads_list.append(t)
q.join()
for _ in range(threads):
q.put(None)
for t in threads_list:
t.join()
seen = set()
unique_wordlists = []
for wl in wordlists:
if wl['name'] not in seen:
unique_wordlists.append(wl)
seen.add(wl['name'])
with open(output_file, "w", encoding="utf-8") as f:
json.dump(unique_wordlists, f, indent=2)
print(f"Saved {len(unique_wordlists)} wordlists to {output_file}")
def download_torrent_file(torrent_url, save_dir=None):
if not save_dir:
save_dir = get_hcat_wordlists_dir()
else:
save_dir = os.path.expanduser(save_dir)
if not os.path.isabs(save_dir):
save_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), save_dir)
os.makedirs(save_dir, exist_ok=True)
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
}
if not torrent_url.startswith("http"):
filename = torrent_url
else:
filename = torrent_url.split("/")[-1]
wordlist_base = filename.replace('.torrent', '').replace('.7z', '').replace('.txt', '')
wordlist_uri = f"https://weakpass.com/wordlists/{wordlist_base}"
print(f"[+] Fetching wordlist page: {wordlist_uri}")
r = requests.get(wordlist_uri, headers=headers)
if r.status_code != 200:
print(f"[!] Failed to fetch wordlist page: {wordlist_uri}")
return None
soup = BeautifulSoup(r.text, "html.parser")
app_div = soup.find("div", id="app")
if not app_div or not app_div.has_attr("data-page"):
print(f"[!] Could not find app data on {wordlist_uri}")
return None
data_page_val = app_div["data-page"]
if not isinstance(data_page_val, str):
data_page_val = str(data_page_val)
data_page_val = data_page_val.replace('"', '"')
try:
data = json.loads(data_page_val)
wordlist = data.get('props', {}).get('wordlist')
wordlist_id = None
torrent_link_from_data = None
if wordlist:
wordlist_id = wordlist.get('id')
torrent_link_from_data = wordlist.get('torrent_link')
else:
wordlists = data.get('props', {}).get('wordlists')
if isinstance(wordlists, dict) and 'data' in wordlists:
wordlists = wordlists['data']
if isinstance(wordlists, list):
for wl in wordlists:
if wl.get('torrent_link') == filename or wl.get('name') == filename:
wordlist_id = wl.get('id')
torrent_link_from_data = wl.get('torrent_link')
break
if wordlist_base in wl.get('name', ''):
wordlist_id = wl.get('id')
torrent_link_from_data = wl.get('torrent_link')
break
except Exception as e:
print(f"[!] Failed to parse data-page JSON: {e}")
return None
if not (torrent_link_from_data and wordlist_id):
print(f"[!] No torrent link or id found in wordlist data for {filename}.")
return None
if not torrent_link_from_data.startswith('http'):
torrent_link = f"https://weakpass.com/download/{wordlist_id}/{torrent_link_from_data}"
else:
torrent_link = torrent_link_from_data
print(f"[+] Downloading .torrent file from: {torrent_link}")
r2 = requests.get(torrent_link, headers=headers, stream=True)
content_type = r2.headers.get("Content-Type", "")
local_filename = os.path.join(save_dir, filename if filename.endswith('.torrent') else filename + '.torrent')
if r2.status_code == 200 and not content_type.startswith("text/html"):
with open(local_filename, 'wb') as f:
for chunk in r2.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
print(f"Saved to {local_filename}")
else:
print(f"Failed to download a valid torrent file: {torrent_link}")
try:
html = r2.content.decode(errors="replace")
print("--- Begin HTML Debug Output ---")
print(html[:2000])
print("--- End HTML Debug Output ---")
except Exception as e:
print(f"Could not decode response for debug: {e}")
return None
if shutil.which("transmission-cli") is None:
print("[ERROR] transmission-cli is not installed or not in your PATH.")
print("Please install it with: brew install transmission-cli (on macOS) or your package manager.")
print(f"Torrent file saved at {local_filename}, but download will not start until transmission-cli is available.")
return local_filename
def run_transmission(torrent_file, output_dir):
import subprocess
import glob
print(f"Starting transmission-cli for {torrent_file}...")
try:
proc = subprocess.Popen([
"transmission-cli",
"-w", output_dir,
torrent_file
], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, universal_newlines=True)
if proc.stdout is not None:
for line in proc.stdout:
print(line, end='')
proc.wait()
if proc.returncode != 0:
print(f"transmission-cli failed for {torrent_file} (exit {proc.returncode})")
return
else:
print(f"Download complete for {torrent_file}")
sevenz_files = glob.glob(os.path.join(output_dir, '*.7z'))
if not sevenz_files:
print("[i] No .7z files found to extract.")
return
for zfile in sevenz_files:
print(f"[+] Extracting {zfile} ...")
sevenz_bin = shutil.which('7z') or shutil.which('7za')
if not sevenz_bin:
print("[!] 7z or 7za not found in PATH. Please install p7zip-full or 7-zip to extract archives.")
continue
try:
extract_result = subprocess.run([
sevenz_bin, 'x', '-y', zfile, f'-o{output_dir}'
], capture_output=True, text=True)
print(extract_result.stdout)
if extract_result.returncode == 0:
print(f"[+] Extraction complete: {zfile}")
else:
print(f"[!] Extraction failed for {zfile}: {extract_result.stderr}")
except Exception as e:
print(f"[!] Error extracting {zfile}: {e}")
except Exception as e:
print(f"Error running transmission-cli: {e}")
t = threading.Thread(target=run_transmission, args=(local_filename, save_dir))
t.start()
print(f"transmission-cli launched in background for {local_filename}")
return local_filename
def weakpass_wordlist_menu():
fetch_all_weakpass_wordlists_multithreaded()
try:
with open("weakpass_wordlists.json", "r", encoding="utf-8") as f:
all_wordlists = json.load(f)
except Exception as e:
print(f"Failed to load local wordlist cache: {e}")
return
page = 0
batch_size = 100
while True:
filtered_wordlists = [wl for wl in all_wordlists if str(wl.get('rank', '')) == '7']
start_idx = page * batch_size
end_idx = start_idx + batch_size
page_wordlists = filtered_wordlists[start_idx:end_idx]
if not page_wordlists:
print("No more wordlists.")
if page > 0:
page -= 1
continue
col_width = 45
cols = 3
print("\nEach entry shows: [number]. [wordlist name] [effectiveness score] [rank]")
print(f"Available Wordlists (Batch {page+1}):")
rows = (len(page_wordlists) + cols - 1) // cols
lines = [''] * rows
for idx, wl in enumerate(page_wordlists):
col = idx // rows
row = idx % rows
effectiveness = wl.get('effectiveness', wl.get('downloads', ''))
rank = wl.get('rank', '')
entry = f"{start_idx+idx+1:3d}. {wl['name'][:25]:<25} {effectiveness:<8} {rank:<2}"
lines[row] += entry.ljust(col_width)
for line in lines:
print(line)
sel = input("\nEnter the number to download, 'n' for next batch, 'p' for previous, or 'q' to cancel: ")
if sel.lower() == 'q':
print("Returning to menu...")
return
if sel.lower() == 'n':
page += 1
continue
if sel.lower() == 'p' and page > 0:
page -= 1
continue
try:
sel_idx = int(sel) - 1 - start_idx
if 0 <= sel_idx < len(page_wordlists):
torrent_url = page_wordlists[sel_idx]['torrent_url']
download_torrent_file(torrent_url)
else:
print("Invalid selection.")
except Exception as e:
print(f"Error: {e}")
import requests
# Hashview Integration - Real API implementation matching hate_crack.py
class HashviewAPI:
def get_hashfile_details(self, hashfile_id):
"""Get hashfile details and hashtype for a given hashfile_id."""
url = f"{self.base_url}/v1/hashfiles/{hashfile_id}"
resp = self.session.get(url)
resp.raise_for_status()
# Try to parse JSON if available, else fallback to raw content
try:
data = resp.json()
except Exception:
data = None
# If JSON, look for hashtype
hashtype = None
if data:
hashtype = data.get('hashtype') or data.get('hash_type')
return {'hashfile_id': hashfile_id, 'hashtype': hashtype, 'details': data, 'raw': resp.content}
FILE_FORMATS = {
'pwdump': 0,
'netntlm': 1,
'kerberos': 2,
'shadow': 3,
'user:hash': 4,
'hash_only': 5,
}
def __init__(self, base_url, api_key, debug=False):
self.base_url = base_url.rstrip('/')
self.api_key = api_key
self.debug = debug
self.session = requests.Session()
self.session.cookies.set('uuid', api_key)
self.session.verify = False
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
def list_customers(self):
url = f"{self.base_url}/v1/customers"
resp = self.session.get(url)
resp.raise_for_status()
data = resp.json()
if 'users' in data:
customers = json.loads(data['users'])
return {'customers': customers}
return data
def list_hashfiles(self):
url = f"{self.base_url}/v1/hashfiles"
resp = self.session.get(url)
resp.raise_for_status()
data = resp.json()
if 'hashfiles' in data:
if isinstance(data['hashfiles'], str):
hashfiles = json.loads(data['hashfiles'])
else:
hashfiles = data['hashfiles']
return hashfiles
return []
def get_customer_hashfiles(self, customer_id):
all_hashfiles = self.list_hashfiles()
return [hf for hf in all_hashfiles if int(hf.get('customer_id', 0)) == customer_id]
def display_customers_multicolumn(self, customers):
if not customers:
print("\nNo customers found.")
return
try:
terminal_width = os.get_terminal_size().columns
except:
terminal_width = 120
max_id_len = max(len(str(c.get('id', ''))) for c in customers)
col_width = max_id_len + 2 + 30 + 2
num_cols = max(1, terminal_width // col_width)
print("\n" + "="*terminal_width)
print("Available Customers:")
print("="*terminal_width)
num_customers = len(customers)
rows = (num_customers + num_cols - 1) // num_cols
for row in range(rows):
line_parts = []
for col in range(num_cols):
idx = row + col * rows
if idx < num_customers:
customer = customers[idx]
cust_id = customer.get('id', 'N/A')
cust_name = customer.get('name', 'N/A')
name_width = col_width - max_id_len - 2 - 2
if len(str(cust_name)) > name_width:
cust_name = str(cust_name)[:name_width-3] + "..."
entry = f"{cust_id}: {cust_name}"
line_parts.append(entry.ljust(col_width))
print("".join(line_parts).rstrip())
print("="*terminal_width)
print(f"Total: {len(customers)} customer(s)")
def upload_hashfile(self, file_path, customer_id, hash_type, file_format=5, hashfile_name=None):
if hashfile_name is None:
hashfile_name = os.path.basename(file_path)
with open(file_path, 'rb') as f:
file_content = f.read()
url = (
f"{self.base_url}/v1/hashfiles/upload/"
f"{customer_id}/{file_format}/{hash_type}/{hashfile_name}"
)
headers = {'Content-Type': 'text/plain'}
resp = self.session.post(url, data=file_content, headers=headers)
resp.raise_for_status()
return resp.json()
def create_job(self, name, hashfile_id, customer_id, limit_recovered=False, notify_email=True):
url = f"{self.base_url}/v1/jobs/add"
headers = {'Content-Type': 'application/json'}
data = {
"name": name,
"hashfile_id": hashfile_id,
"customer_id": customer_id,
}
resp = self.session.post(url, json=data, headers=headers)
resp.raise_for_status()
return resp.json()
def download_left_hashes(self, customer_id, hashfile_id, output_file=None):
url = f"{self.base_url}/v1/hashfiles/{hashfile_id}"
resp = self.session.get(url)
resp.raise_for_status()
if output_file is None:
output_file = f"left_{customer_id}_{hashfile_id}.txt"
with open(output_file, 'wb') as f:
f.write(resp.content)
return {'output_file': output_file, 'size': len(resp.content)}
def upload_cracked_hashes(self, file_path, hash_type='1000'):
valid_lines = []
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
for line in f:
line = line.strip()
if '31d6cfe0d16ae931b73c59d7e0c089c0' in line:
continue
if not line or ':' not in line:
continue
parts = line.split(':', 1)
if len(parts) != 2:
break
hash_value = parts[0].strip()
plaintext = parts[1].strip()
valid_lines.append(f"{hash_value}:{plaintext}")
converted_content = '\n'.join(valid_lines)
url = f"{self.base_url}/v1/hashes/import/{hash_type}"
headers = {'Content-Type': 'text/plain'}
resp = self.session.post(url, data=converted_content, headers=headers)
resp.raise_for_status()
try:
json_response = resp.json()
if 'type' in json_response and json_response['type'] == 'Error':
raise Exception(f"Hashview API Error: {json_response.get('msg', 'Unknown error')}")
return json_response
except (json.JSONDecodeError, ValueError):
raise Exception(f"Invalid API response: {resp.text[:200]}")
def create_customer(self, name):
url = f"{self.base_url}/v1/customers/add"
headers = {'Content-Type': 'application/json'}
data = {"name": name}
resp = self.session.post(url, json=data, headers=headers)
resp.raise_for_status()
return resp.json()
def download_hashes_from_hashview(
-168
View File
@@ -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()
-338
View File
@@ -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('&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:
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}")
+1 -1
View File
@@ -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'
-38
View File
@@ -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"]
-53
View File
@@ -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)
-31
View File
@@ -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")