gitea-mirror/hate_crack
1
0
Fork 0
mirror of https://github.com/trustedsec/hate_crack.git synced 2026-07-28 22:51:14 -07:00
Code Issues Packages Projects Releases Wiki Activity

fix: overhaul Hashview download flow and fix hashcat --show stderr pollution

Browse Source 
- Merge download_left and download_found into single "Download Hashes" menu option
- Append found hash:clear pairs to potfile instead of running broken hashcat re-crack
- Append found hashes to left file so hashcat --show returns full results
- Clean up found_ temp files after merge
- Split found file on first colon (not last) to handle passwords containing colons
- Filter hashcat parse errors from --show stdout in _run_hashcat_show
- Add get_hcat_potfile_path() helper to api.py for potfile resolution
- Remove obsolete download_found_hashes API method and CLI subcommand
- Fix ollama tests to match current 4-arg hcatOllama signature and rule loop

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Justin Bollinger
2026-02-13 19:11:51 -05:00
co-authored by Claude Opus 4.6
parent ca1d71da6c
commit f9db54e84c
6 changed files with 346 additions and 1004 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hate_crack/api.py
+53 -267
View File
@@ -117,6 +117,29 @@ def get_hcat_tuning_args():
return []
def get_hcat_potfile_path():
"""Return the resolved potfile path from config, or the default."""
config_path = _resolve_config_path()
if config_path:
try:
with open(config_path) as f:
config = json.load(f)
raw = (config.get("hcatPotfilePath") or "").strip()
if raw:
return os.path.expanduser(raw)
except Exception:
pass
return os.path.expanduser("~/.hashcat/hashcat.potfile")
def get_hcat_potfile_args():
"""Return potfile args list for hashcat, e.g. ['--potfile-path=/path']."""
pot = get_hcat_potfile_path()
if pot:
return [f"--potfile-path={pot}"]
return []
def cleanup_torrent_files(directory=None):
"""Remove stray .torrent files from the wordlists directory on graceful exit."""
if directory is None:
@@ -805,7 +828,6 @@ class HashviewAPI:
self, customer_id, hashfile_id, output_file=None, hash_type=None
):
import sys
import subprocess
url = f"{self.base_url}/v1/hashfiles/{hashfile_id}/left"
resp = self.session.get(url, headers=self._auth_headers(), stream=True)
@@ -882,7 +904,7 @@ class HashviewAPI:
for line in f:
line = line.strip()
if line:
parts = line.rsplit(":", 1) # Split on last colon
parts = line.split(":", 1) # Split on first colon
if len(parts) == 2:
hash_part, clear_part = parts
hf.write(hash_part + "\n")
@@ -894,277 +916,41 @@ class HashviewAPI:
f"Split found file into {hashes_count} hashes and {clears_count} clears"
)
# Run hashcat to combine them
combined_file = output_abs + ".out"
try:
# Execute hashcat: hashcat <tuning> -m hash_type found_hashes found_clears --outfile output.out --outfile-format=1,2
tuning_args = get_hcat_tuning_args()
# Append found hashes to the left file
with open(output_abs, "a", encoding="utf-8") as lf:
with open(found_hashes_file, "r", encoding="utf-8", errors="ignore") as fhf:
for line in fhf:
line = line.strip()
if line:
lf.write(line + "\n")
print(f"✓ Appended {hashes_count} found hashes to {output_abs}")
# Create temporary outfile for hashcat
temp_outfile = output_abs + ".tmp"
if self.debug:
print(
f"[DEBUG] download_left_hashes: hash_type={hash_type}, type={type(hash_type)}"
)
# Build command with hash type if provided
cmd = ["hashcat", *tuning_args]
if hash_type:
cmd.extend(["-m", str(hash_type)])
cmd.extend(
[
found_hashes_file,
found_clears_file,
"--outfile",
temp_outfile,
"--outfile-format=1,2",
]
)
if self.debug:
print(f"[DEBUG] Running command: {' '.join(cmd)}")
print(f"Running: {' '.join(cmd)}")
result = subprocess.run(
cmd,
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
text=True,
timeout=300,
)
if result.returncode != 0:
print(f"Warning: hashcat exited with code {result.returncode}")
if result.stderr:
print(f" stderr: {result.stderr}")
# Append the output to the combined file
if os.path.exists(temp_outfile):
with open(
temp_outfile, "r", encoding="utf-8", errors="ignore"
) as tmp_f:
with open(combined_file, "a", encoding="utf-8") as out_f:
out_f.write(tmp_f.read())
# Count lines appended
with open(
combined_file, "r", encoding="utf-8", errors="ignore"
) as f:
combined_count = len(f.readlines())
print(
f"✓ Appended cracked hashes to {combined_file} (total lines: {combined_count})"
)
# Clean up temp file
try:
os.remove(temp_outfile)
except Exception:
pass
else:
print("Note: No cracked hashes found")
except FileNotFoundError:
print("✗ Error: hashcat not found in PATH")
except subprocess.TimeoutExpired:
print("✗ Error: hashcat execution timed out")
except Exception as e:
print(f"✗ Error running hashcat: {e}")
# Clean up temporary files (keep when debug is enabled)
if not self.debug:
files_to_delete = [found_file, found_hashes_file, found_clears_file]
for temp_file in files_to_delete:
try:
if os.path.exists(temp_file):
os.remove(temp_file)
print(f"Deleted {temp_file}")
except Exception as e:
print(f"Warning: Could not delete {temp_file}: {e}")
# Append found hash:clear pairs to the potfile
potfile_path = get_hcat_potfile_path()
if potfile_path:
appended = 0
with open(potfile_path, "a", encoding="utf-8") as pf:
with open(found_file, "r", encoding="utf-8", errors="ignore") as ff:
for line in ff:
line = line.strip()
if line and ":" in line:
pf.write(line + "\n")
appended += 1
combined_count = appended
print(f"✓ Appended {appended} found hashes to potfile: {potfile_path}")
else:
print("Debug enabled: keeping found and split files")
print("Warning: No potfile path configured, skipping potfile update")
# Clean up the two found_ files
for f_path in (found_file, found_hashes_file, found_clears_file):
try:
os.remove(f_path)
except OSError:
pass
except Exception as e:
# If there's any error downloading found file, just skip it
print(f"Note: Could not download found hashes: {e}")
# Clean up found file if it was partially written
if not self.debug:
try:
if os.path.exists(found_file):
os.remove(found_file)
except Exception:
pass
return {
"output_file": output_file,
"size": downloaded,
"combined_count": combined_count,
"combined_file": combined_file,
}
def download_found_hashes(
self, customer_id, hashfile_id, output_file=None, hash_type=None
):
import sys
import subprocess
url = f"{self.base_url}/v1/hashfiles/{hashfile_id}/found"
resp = self.session.get(url, headers=self._auth_headers(), stream=True)
resp.raise_for_status()
if output_file is None:
output_file = f"found_{customer_id}_{hashfile_id}.txt"
# Convert to absolute path to ensure file is preserved if CWD changes
output_file = os.path.abspath(output_file)
total = int(resp.headers.get("content-length", 0))
downloaded = 0
chunk_size = 8192
with open(output_file, "wb") as f:
for chunk in resp.iter_content(chunk_size=chunk_size):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if total > 0:
done = int(50 * downloaded / total)
bar = "[" + "=" * done + " " * (50 - done) + "]"
percent = 100 * downloaded / total
sys.stdout.write(
f"\rDownloading: {bar} {percent:5.1f}% ({downloaded}/{total} bytes)"
)
sys.stdout.flush()
if total > 0:
sys.stdout.write("\n")
# If content-length is not provided, just print size at end
if total == 0:
print(f"Downloaded {downloaded} bytes.")
# Split found file into hashes and clears
output_dir = os.path.dirname(os.path.abspath(output_file)) or os.getcwd()
found_hashes_file = os.path.join(
output_dir, f"found_hashes_{customer_id}_{hashfile_id}.txt"
)
found_clears_file = os.path.join(
output_dir, f"found_clears_{customer_id}_{hashfile_id}.txt"
)
hashes_count = 0
clears_count = 0
combined_count = 0
combined_file = None
try:
with (
open(found_hashes_file, "w", encoding="utf-8") as hf,
open(found_clears_file, "w", encoding="utf-8") as cf,
):
with open(output_file, "r", encoding="utf-8", errors="ignore") as f:
for line in f:
line = line.strip()
if line:
parts = line.rsplit(":", 1) # Split on last colon
if len(parts) == 2:
hash_part, clear_part = parts
hf.write(hash_part + "\n")
cf.write(clear_part + "\n")
hashes_count += 1
clears_count += 1
print(
f"✓ Split found file into {hashes_count} hashes and {clears_count} clears"
)
# Run hashcat to combine them into an output file
combined_file = output_file + ".out"
try:
tuning_args = get_hcat_tuning_args()
# Create temporary outfile for hashcat
temp_outfile = output_file + ".tmp"
if self.debug:
print(
f"[DEBUG] download_found_hashes: hash_type={hash_type}, type={type(hash_type)}"
)
# Build command with hash type if provided
cmd = ["hashcat", *tuning_args]
if hash_type:
cmd.extend(["-m", str(hash_type)])
cmd.extend(
[
found_hashes_file,
found_clears_file,
"--outfile",
temp_outfile,
"--outfile-format=1,2",
]
)
if self.debug:
print(f"[DEBUG] Running command: {' '.join(cmd)}")
print(f"Running: {' '.join(cmd)}")
result = subprocess.run(
cmd,
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
text=True,
timeout=300,
)
if result.returncode != 0:
print(f"Warning: hashcat exited with code {result.returncode}")
if result.stderr:
print(f" stderr: {result.stderr}")
# Write the output file
if os.path.exists(temp_outfile):
with open(
temp_outfile, "r", encoding="utf-8", errors="ignore"
) as tmp_f:
with open(combined_file, "w", encoding="utf-8") as out_f:
out_f.write(tmp_f.read())
# Count lines in output
with open(
combined_file, "r", encoding="utf-8", errors="ignore"
) as f:
combined_count = len(f.readlines())
print(f"✓ Created {combined_file} (total lines: {combined_count})")
# Clean up temp file
try:
os.remove(temp_outfile)
except Exception:
pass
else:
print("Note: No cracked hashes generated")
except FileNotFoundError:
print("✗ Error: hashcat not found in PATH")
except subprocess.TimeoutExpired:
print("✗ Error: hashcat execution timed out")
except Exception as e:
print(f"✗ Error running hashcat: {e}")
# Clean up temporary files (keep when debug is enabled)
if not self.debug:
files_to_delete = [found_hashes_file, found_clears_file]
for temp_file in files_to_delete:
try:
if os.path.exists(temp_file):
os.remove(temp_file)
except Exception as e:
if self.debug:
print(f"Warning: Could not delete {temp_file}: {e}")
else:
print("Debug enabled: keeping split files")
except Exception as e:
print(f"✗ Error splitting found file: {e}")
return {
"output_file": output_file,
hate_crack/main.py
+87 -296
View File
@@ -9,7 +9,6 @@
import sys
import os
import json
import lzma
import shutil
import logging
import binascii
@@ -454,15 +453,7 @@ except KeyError as e:
default_config.get("hcatDebugLogPath", "./hashcat_debug")
)
try:
ollamaUrl = config_parser["ollamaUrl"]
except KeyError as e:
print(
"{0} is not defined in config.json using defaults from config.json.example".format(
e
)
)
ollamaUrl = default_config.get("ollamaUrl", "http://localhost:11434")
ollamaUrl = "http://" + os.environ.get("OLLAMA_HOST", "localhost:11434")
try:
ollamaModel = config_parser["ollamaModel"]
except KeyError as e:
@@ -473,23 +464,23 @@ except KeyError as e:
)
ollamaModel = default_config.get("ollamaModel", "llama3.2")
try:
markovCandidateCount = int(config_parser["markovCandidateCount"])
ollamaCandidateCount = int(config_parser["ollamaCandidateCount"])
except KeyError as e:
print(
"{0} is not defined in config.json using defaults from config.json.example".format(
e
)
)
markovCandidateCount = int(default_config.get("markovCandidateCount", 5000))
ollamaCandidateCount = int(default_config.get("ollamaCandidateCount", 5000))
try:
markovWordlist = config_parser["markovWordlist"]
ollamaWordlist = config_parser["ollamaWordlist"]
except KeyError as e:
print(
"{0} is not defined in config.json using defaults from config.json.example".format(
e
)
)
markovWordlist = default_config.get("markovWordlist", "rockyou.txt")
ollamaWordlist = default_config.get("ollamaWordlist", "rockyou.txt")
hcatExpanderBin = "expander.bin"
hcatCombinatorBin = "combinator.bin"
@@ -629,7 +620,7 @@ hcatGoodMeasureBaseList = _normalize_wordlist_setting(
hcatGoodMeasureBaseList, wordlists_dir
)
hcatPrinceBaseList = _normalize_wordlist_setting(hcatPrinceBaseList, wordlists_dir)
markovWordlist = _normalize_wordlist_setting(markovWordlist, wordlists_dir)
ollamaWordlist = _normalize_wordlist_setting(ollamaWordlist, wordlists_dir)
if not SKIP_INIT:
# Verify hashcat binary is available
@@ -695,7 +686,7 @@ if not SKIP_INIT:
except SystemExit:
print("PRINCE attacks will not be available.")
# Verify hcstat2gen binary (optional, for Markov attacks)
# Verify hcstat2gen binary (optional, for LLM attacks)
# Note: hcstat2gen is part of hashcat-utils, already in hate_crack repo
hcstat2gen_path = (
hate_path + "/hashcat-utils/bin/" + hcatHcstat2genBin
@@ -707,7 +698,7 @@ if not SKIP_INIT:
name="hcstat2gen",
)
except SystemExit:
print("LLM Markov attacks will not be available.")
print("LLM attacks will not be available.")
except Exception as e:
print(f"Module initialization error: {e}")
@@ -938,20 +929,25 @@ def _write_field_sorted_unique(input_path, output_path, field_index, delimiter="
def _run_hashcat_show(hash_type, hash_file, output_path):
result = subprocess.run(
[
hcatBin,
"--show",
# Use hashcat's built-in potfile unless configured otherwise.
*([f"--potfile-path={hcatPotfilePath}"] if hcatPotfilePath else []),
"-m",
str(hash_type),
hash_file,
],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
check=False,
)
with open(output_path, "w") as out:
subprocess.run(
[
hcatBin,
"--show",
# Use hashcat's built-in potfile unless configured otherwise.
*([f"--potfile-path={hcatPotfilePath}"] if hcatPotfilePath else []),
"-m",
str(hash_type),
hash_file,
],
stdout=out,
check=False,
)
for line in result.stdout.decode("utf-8", errors="ignore").splitlines():
# hashcat --show prints parse errors to stdout; skip non-result lines
if ":" in line and not line.startswith(("Hash parsing error", "* ")):
out.write(line + "\n")
# Brute Force Attack
@@ -1539,22 +1535,10 @@ def _pull_ollama_model(url, model):
return True
# LLM Markov Attack
def hcatMarkov(hcatHashType, hcatHashFile, mode, context_data, candidate_count):
# LLM Ollama Attack
def hcatOllama(hcatHashType, hcatHashFile, mode, context_data):
global hcatProcess
candidates_path = f"{hcatHashFile}.markov_candidates"
hcstat2_raw_path = f"{hcatHashFile}.hcstat2_raw"
hcstat2_path = f"{hcatHashFile}.hcstat2"
hcstat2gen_path = os.path.join(
hate_path, "hashcat-utils", "bin", hcatHcstat2genBin
)
if not os.path.isfile(hcstat2gen_path):
print(
f"Error: hcstat2gen not found at {hcstat2gen_path}. "
"LLM Markov attacks are not available."
)
return
candidates_path = f"{hcatHashFile}.ollama_candidates"
# Step A: Build LLM prompt based on mode
if mode == "wordlist":
@@ -1577,19 +1561,18 @@ def hcatMarkov(hcatHashType, hcatHashFile, mode, context_data, candidate_count):
wordlist_sample = "\n".join(sample_lines)
prompt = (
"You are a password generation expert. Below is a sample of real passwords. "
"Study the patterns, character choices, and structures. Generate exactly "
f"{candidate_count} new unique passwords that follow similar patterns but "
"are NOT copies of the input. Output ONLY passwords, one per line, no "
"numbering or explanation.\n\n"
f"Sample passwords:\n{wordlist_sample}"
"Study the patterns, character choices, and structures. Then generate hashcat rules" \
" that could transform common base words into similar passwords. Focus on patterns like " \
"capitalization, leetspeak, suffixes, and common substitutions. Here are the sample passwords:\n" \
f"{wordlist_sample}"
)
elif mode == "target":
company = context_data.get("company", "")
industry = context_data.get("industry", "")
location = context_data.get("location", "")
prompt = (
f"Generate exactly {candidate_count} realistic passwords that employees at "
f"{company} in the {industry} industry located in {location} might use. "
"Generate baseword to be used in a denylist for keeping users from setting their passwords with these basewords. We are protecting the employees of "
f"{company} in the {industry} industry located in {location}.\n\n"
"Include variations with:\n"
"- Company name variations and abbreviations\n"
"- Common password patterns (Season+Year, Name+Numbers)\n"
@@ -1599,11 +1582,11 @@ def hcatMarkov(hcatHashType, hcatHashFile, mode, context_data, candidate_count):
"Output ONLY the passwords, one per line, no numbering or explanation."
)
else:
print(f"Error: Unknown Markov generation mode: {mode}")
print(f"Error: Unknown LLM generation mode: {mode}")
return
# Step B: Call Ollama API to generate candidates
print(f"Generating {candidate_count} password candidates via Ollama ({ollamaModel})...")
print(f"Generating password candidates via Ollama ({ollamaModel})...")
api_url = f"{ollamaUrl}/api/generate"
payload = json.dumps({
"model": ollamaModel,
@@ -1642,7 +1625,7 @@ def hcatMarkov(hcatHashType, hcatHashFile, mode, context_data, candidate_count):
print(f"Error calling Ollama API after pull: {retry_err}")
return
else:
print(f"Could not pull model '{ollamaModel}'. Aborting Markov attack.")
print(f"Could not pull model '{ollamaModel}'. Aborting LLM attack.")
return
else:
print(f"Error: Could not connect to Ollama at {ollamaUrl}: {e}")
@@ -1688,48 +1671,8 @@ def hcatMarkov(hcatHashType, hcatHashFile, mode, context_data, candidate_count):
filtered_count = len(raw_lines) - len(candidates)
print(f"[DEBUG] Filtered out {filtered_count} lines from Ollama response ({len(raw_lines)} raw -> {len(candidates)} candidates)")
# Step C: Run hcstat2gen to build Markov stats
print("Building Markov statistics with hcstat2gen...")
try:
with open(candidates_path, "rb") as candidates_file:
hcatProcess = subprocess.Popen(
[hcstat2gen_path, hcstat2_raw_path],
stdin=candidates_file,
)
try:
hcatProcess.wait()
except KeyboardInterrupt:
print("Killing PID {0}...".format(str(hcatProcess.pid)))
hcatProcess.kill()
return
except Exception as e:
print(f"Error running hcstat2gen: {e}")
return
if not os.path.isfile(hcstat2_raw_path):
print("Error: hcstat2gen did not produce output file.")
return
# Step D: LZMA compress the raw hcstat2 file
print("Compressing hcstat2 file with LZMA...")
try:
with open(hcstat2_raw_path, "rb") as raw_file:
raw_data = raw_file.read()
compressed = lzma.compress(
raw_data,
format=lzma.FORMAT_RAW,
filters=[{"id": lzma.FILTER_LZMA1}],
)
with open(hcstat2_path, "wb") as out_file:
out_file.write(compressed)
except Exception as e:
print(f"Error compressing hcstat2 file: {e}")
return
print(f"Markov statistics file created -> {hcstat2_path}")
# Step E: Run hashcat mask attack with custom Markov stats
print("Running Markov-enhanced mask attack...")
# Step C: Run hashcat wordlist attack with LLM-generated candidates (no rules)
print("Running wordlist attack with LLM-generated candidates...")
cmd = [
hcatBin,
"-m",
@@ -1739,13 +1682,7 @@ def hcatMarkov(hcatHashType, hcatHashFile, mode, context_data, candidate_count):
generate_session_id(),
"-o",
f"{hcatHashFile}.out",
"-a",
"3",
f"--markov-hcstat2={hcstat2_path}",
"--increment",
"--increment-min=1",
"--increment-max=14",
"?a?a?a?a?a?a?a?a?a?a?a?a?a?a",
candidates_path,
]
cmd.extend(shlex.split(hcatTuning))
_append_potfile_arg(cmd)
@@ -1755,14 +1692,43 @@ def hcatMarkov(hcatHashType, hcatHashFile, mode, context_data, candidate_count):
except KeyboardInterrupt:
print("Killing PID {0}...".format(str(hcatProcess.pid)))
hcatProcess.kill()
return
# Step F: Cleanup temp files
for temp_file in [candidates_path, hcstat2_raw_path]:
# Step D: Run hashcat with LLM candidates against every rule in the rules directory
rule_files = sorted(
f for f in os.listdir(rulesDirectory) if f != ".DS_Store"
)
if not rule_files:
print("No rule files found in rules directory. Skipping rule-based attacks.")
return
print(f"\nRunning LLM candidates with {len(rule_files)} rule file(s) from {rulesDirectory}...")
for rule in rule_files:
rule_path = os.path.join(rulesDirectory, rule)
print(f"\n\tRunning with rule: {rule}")
cmd = [
hcatBin,
"-m",
hcatHashType,
hcatHashFile,
"--session",
generate_session_id(),
"-o",
f"{hcatHashFile}.out",
"-r",
rule_path,
candidates_path,
]
cmd.extend(shlex.split(hcatTuning))
_append_potfile_arg(cmd)
hcatProcess = subprocess.Popen(cmd)
try:
if os.path.isfile(temp_file):
os.remove(temp_file)
except Exception:
pass
hcatProcess.wait()
except KeyboardInterrupt:
print("Killing PID {0}...".format(str(hcatProcess.pid)))
hcatProcess.kill()
return
# Middle fast Combinator Attack
@@ -2280,13 +2246,10 @@ def hashview_api():
menu_options.append(("download_wordlist", "Download Wordlist"))
menu_options.append(
(
"download_left",
"Download Left Hashes (with automatic merge if found)",
"download_hashes",
"Download Hashes (left + found to potfile)",
)
)
menu_options.append(
("download_found", "Download Found Hashes (with automatic split)")
)
if hcatHashFile:
menu_options.append(
("upload_hashfile_job", "Upload Hashfile and Create Job")
@@ -2683,7 +2646,7 @@ def hashview_api():
except Exception as e:
print(f"\n✗ Error uploading hashfile: {str(e)}")
elif option_key == "download_left":
elif option_key == "download_hashes":
# Download left hashes
try:
while True:
@@ -2852,157 +2815,6 @@ def hashview_api():
except Exception as e:
print(f"\n✗ Error downloading hashes: {str(e)}")
elif option_key == "download_found":
# Download found hashes
try:
while True:
# First, list customers to help user select
customers_result = api_harness.list_customers()
customers = (
customers_result.get("customers", [])
if isinstance(customers_result, dict)
else customers_result
)
if customers:
api_harness.display_customers_multicolumn(customers)
else:
print("\nNo customers found.")
# Select or create customer
customer_input = input(
"\nEnter customer ID or N to create new: "
).strip()
if customer_input.lower() == "n":
customer_name = input("Enter customer name: ").strip()
if customer_name:
try:
result = api_harness.create_customer(customer_name)
print(
f"\n✓ Success: {result.get('msg', 'Customer created')}"
)
customer_id = result.get(
"customer_id"
) or result.get("id")
if not customer_id:
print("\n✗ Error: Customer ID not returned.")
continue
print(f" Customer ID: {customer_id}")
except Exception as e:
print(f"\n✗ Error creating customer: {str(e)}")
continue
else:
print("\n✗ Error: Customer name cannot be empty.")
continue
else:
try:
customer_id = int(customer_input)
except ValueError:
print(
"\n✗ Error: Invalid ID entered. Please enter a numeric ID or N."
)
continue
# List hashfiles for the customer
try:
customer_hashfiles = api_harness.get_customer_hashfiles(
customer_id
)
if not customer_hashfiles:
print(
f"\nNo hashfiles found for customer ID {customer_id}"
)
continue
print("\n" + "=" * 120)
print(f"Hashfiles for Customer ID {customer_id}:")
print("=" * 120)
print(f"{'ID':<10} {'Hash Type':<10} {'Name':<96}")
print("-" * 120)
hashfile_map = {}
for hf in customer_hashfiles:
hf_id = hf.get("id")
hf_name = hf.get("name", "N/A")
hf_type = (
hf.get("hash_type") or hf.get("hashtype") or "N/A"
)
if hf_id is None:
continue
# Truncate long names to fit within 120 columns
if len(str(hf_name)) > 96:
hf_name = str(hf_name)[:93] + "..."
if debug_mode:
print(
f"[DEBUG] Hashfile {hf_id}: hash_type={hf.get('hash_type')}, hashtype={hf.get('hashtype')}, combined={hf_type}"
)
print(f"{hf_id:<10} {hf_type:<10} {hf_name:<96}")
hashfile_map[int(hf_id)] = hf_type
print("=" * 120)
print(f"Total: {len(hashfile_map)} hashfile(s)")
except Exception as e:
print(f"\nWarning: Could not list hashfiles: {e}")
continue
while True:
try:
hashfile_id_input = input(
"\nEnter hashfile ID: "
).strip()
hashfile_id = int(hashfile_id_input)
except ValueError:
print(
"\n✗ Error: Invalid ID entered. Please enter a numeric ID."
)
continue
if hashfile_id not in hashfile_map:
print(
"\n✗ Error: Hashfile ID not in the list. Please try again."
)
continue
break
break
# Set output filename automatically
output_file = f"found_{customer_id}_{hashfile_id}.txt"
# Get hash type for hashcat from the hashfile map
selected_hash_type = hashfile_map.get(hashfile_id)
if debug_mode:
print(
f"[DEBUG] selected_hash_type from map: {selected_hash_type}"
)
if not selected_hash_type or selected_hash_type == "N/A":
try:
details = api_harness.get_hashfile_details(hashfile_id)
selected_hash_type = details.get("hashtype")
if debug_mode:
print(
f"[DEBUG] selected_hash_type from get_hashfile_details: {selected_hash_type}"
)
except Exception as e:
if debug_mode:
print(f"[DEBUG] Error fetching hashfile details: {e}")
selected_hash_type = None
# Download the found hashes
if debug_mode:
print(
f"[DEBUG] Calling download_found_hashes with hash_type={selected_hash_type}"
)
download_result = api_harness.download_found_hashes(
customer_id, hashfile_id, output_file
)
print(f"\n✓ Success: Downloaded {download_result['size']} bytes")
print(f" File: {download_result['output_file']}")
if selected_hash_type:
print(f" Hash mode: {selected_hash_type}")
print("\nFound hashes downloaded successfully. These are already cracked hashes.")
except ValueError:
print("\n✗ Error: Invalid ID entered. Please enter a numeric ID.")
except Exception as e:
print(f"\n✗ Error downloading found hashes: {str(e)}")
elif option_key == "back":
break
@@ -3075,8 +2887,8 @@ def loopback_attack():
return _attacks.loopback_attack(_attack_ctx())
def markov_attack():
return _attacks.markov_attack(_attack_ctx())
def ollama_attack():
return _attacks.ollama_attack(_attack_ctx())
# convert hex words for recycling
@@ -3306,7 +3118,7 @@ def get_main_menu_options():
"12": thorough_combinator,
"13": bandrel_method,
"14": loopback_attack,
"15": markov_attack,
"15": ollama_attack,
"90": download_hashmob_rules,
"91": analyze_rules,
"92": download_hashmob_wordlists,
@@ -3448,8 +3260,8 @@ def main():
)
hv_download_left = hashview_subparsers.add_parser(
"download-left",
help="Download left hashes for a hashfile",
"download-hashes",
help="Download left hashes and append found hashes to potfile",
)
hv_download_left.add_argument(
"--customer-id", required=True, type=int, help="Customer ID"
@@ -3463,17 +3275,6 @@ def main():
help="Hash type for hashcat (e.g., 1000 for NTLM)",
)
hv_download_found = hashview_subparsers.add_parser(
"download-found",
help="Download found hashes for a hashfile",
)
hv_download_found.add_argument(
"--customer-id", required=True, type=int, help="Customer ID"
)
hv_download_found.add_argument(
"--hashfile-id", required=True, type=int, help="Hashfile ID"
)
hv_upload_hashfile_job = hashview_subparsers.add_parser(
"upload-hashfile-job",
help="Upload a hashfile and create a job",
@@ -3514,8 +3315,7 @@ def main():
hashview_subcommands = [
"upload-cracked",
"upload-wordlist",
"download-left",
"download-found",
"download-hashes",
"upload-hashfile-job",
]
has_hashview_flag = "--hashview" in argv
@@ -3639,7 +3439,7 @@ def main():
print(f" Wordlist ID: {result['wordlist_id']}")
sys.exit(0)
if args.hashview_command == "download-left":
if args.hashview_command == "download-hashes":
download_result = api_harness.download_left_hashes(
args.customer_id,
args.hashfile_id,
@@ -3649,15 +3449,6 @@ def main():
print(f" File: {download_result['output_file']}")
sys.exit(0)
if args.hashview_command == "download-found":
download_result = api_harness.download_found_hashes(
args.customer_id,
args.hashfile_id,
)
print(f"\n✓ Success: Downloaded {download_result['size']} bytes")
print(f" File: {download_result['output_file']}")
sys.exit(0)
if args.hashview_command == "upload-hashfile-job":
hashfile_path = resolve_path(args.file)
if not hashfile_path or not os.path.isfile(hashfile_path):
@@ -3887,7 +3678,7 @@ def main():
print("\t(12) Thorough Combinator Attack")
print("\t(13) Bandrel Methodology")
print("\t(14) Loopback Attack")
print("\t(15) LLM Markov Attack")
print("\t(15) LLM Attack")
print("\n\t(90) Download rules from Hashmob.net")
print("\n\t(91) Analyze Hashcat Rules")
print("\t(92) Download wordlists from Hashmob.net")
tests/test_hashview.py
+4 -54
View File
@@ -270,52 +270,6 @@ class TestHashviewAPI:
assert "Cookie" in auth_headers or "uuid" in str(auth_headers)
assert HASHVIEW_API_KEY in str(auth_headers)
def test_download_found_hashes(self, api, tmp_path):
"""Test downloading found hashes: real API if possible, else mock."""
hashview_url, hashview_api_key = self._get_hashview_config()
customer_id = os.environ.get("HASHVIEW_CUSTOMER_ID")
hashfile_id = os.environ.get("HASHVIEW_HASHFILE_ID")
if all([hashview_url, hashview_api_key, customer_id, hashfile_id]):
# Real API test
real_api = HashviewAPI(hashview_url, hashview_api_key)
output_file = tmp_path / f"found_{customer_id}_{hashfile_id}.txt"
result = real_api.download_found_hashes(
int(customer_id), int(hashfile_id), output_file=str(output_file)
)
assert os.path.exists(result["output_file"])
with open(result["output_file"], "rb") as f:
content = f.read()
print(f"[DEBUG] Downloaded {len(content)} bytes to {result['output_file']}")
assert result["size"] == len(content)
else:
# Mock test
mock_response = Mock()
mock_response.content = b"hash1:pass1\nhash2:pass2\n"
mock_response.raise_for_status = Mock()
mock_response.headers = {"content-length": "0"}
def iter_content(chunk_size=8192):
yield mock_response.content
mock_response.iter_content = iter_content
api.session.get.return_value = mock_response
output_file = tmp_path / "found_1_2.txt"
result = api.download_found_hashes(1, 2, output_file=str(output_file))
assert os.path.exists(result["output_file"])
with open(result["output_file"], "rb") as f:
content = f.read()
assert content == b"hash1:pass1\nhash2:pass2\n"
assert result["size"] == len(content)
# Verify auth headers were passed in the found hashes download call
call_args_list = api.session.get.call_args_list
found_call = [c for c in call_args_list if "found" in str(c)][0]
assert found_call.kwargs.get("headers") is not None
auth_headers = found_call.kwargs.get("headers")
assert "Cookie" in auth_headers or "uuid" in str(auth_headers)
assert HASHVIEW_API_KEY in str(auth_headers)
def test_download_wordlist(self, api, tmp_path):
"""Test downloading a wordlist: real API if possible, else mock."""
hashview_url, hashview_api_key = self._get_hashview_config()
@@ -640,21 +594,17 @@ class TestHashviewAPI:
# Verify left file was created
assert os.path.exists(result["output_file"])
# Verify found file was downloaded and deleted
# Verify found files are cleaned up after merge
found_file = tmp_path / "found_1_2.txt"
assert not os.path.exists(found_file), (
"Found file should be deleted after split"
)
assert not (other_cwd / "found_1_2.txt").exists()
assert not os.path.exists(found_file), "Found file should be deleted after merge"
# Verify split files were created and deleted
found_hashes_file = tmp_path / "found_hashes_1_2.txt"
found_clears_file = tmp_path / "found_clears_1_2.txt"
assert not os.path.exists(str(found_hashes_file)), (
"Split hashes file should be deleted"
"Split hashes file should be deleted after merge"
)
assert not os.path.exists(str(found_clears_file)), (
"Split clears file should be deleted"
"Split clears file should be deleted after merge"
)
def test_download_left_id_matching(self, api, tmp_path):
tests/test_hashview_cli_subcommands.py
-9
View File
@@ -28,15 +28,6 @@ class DummyHashviewAPI:
"size": 10,
}
def download_found_hashes(self, customer_id, hashfile_id, output_file=None):
self.calls.append(
("download_found_hashes", customer_id, hashfile_id, output_file)
)
return {
"output_file": output_file or f"found_{customer_id}_{hashfile_id}.txt",
"size": 12,
}
def upload_hashfile(
self, file_path, customer_id, hash_type, file_format=5, hashfile_name=None
):
tests/test_hashview_cli_subcommands_subprocess.py
+9 -30
View File
@@ -77,8 +77,7 @@ def _ensure_customer_one():
"--name",
"TestWordlist",
],
["hashview", "download-left", "--customer-id", "1", "--hashfile-id", "2"],
["hashview", "download-found", "--customer-id", "1", "--hashfile-id", "2"],
["hashview", "download-hashes", "--customer-id", "1", "--hashfile-id", "2"],
[
"hashview",
"upload-hashfile-job",
@@ -142,45 +141,25 @@ def test_hashview_subcommands_live_downloads():
base_cmd = [sys.executable, HATE_CRACK_SCRIPT, "hashview"]
customer_id = _ensure_customer_one()
left_cmd = base_cmd + [
"download-left",
dl_cmd = base_cmd + [
"download-hashes",
"--customer-id",
str(customer_id),
"--hashfile-id",
os.environ["HASHVIEW_HASHFILE_ID"],
]
left = subprocess.run(
left_cmd,
dl = subprocess.run(
dl_cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
cwd=REPO_ROOT,
env=env,
)
left_out = left.stdout + left.stderr
assert left.returncode == 0, left_out
assert "Downloaded" in left_out
assert "left_" in left_out
found_cmd = base_cmd + [
"download-found",
"--customer-id",
str(customer_id),
"--hashfile-id",
os.environ["HASHVIEW_HASHFILE_ID"],
]
found = subprocess.run(
found_cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
cwd=REPO_ROOT,
env=env,
)
found_out = found.stdout + found.stderr
assert found.returncode == 0, found_out
assert "Downloaded" in found_out
assert "found_" in found_out
dl_out = dl.stdout + dl.stderr
assert dl.returncode == 0, dl_out
assert "Downloaded" in dl_out
assert "left_" in dl_out
@pytest.mark.skipif(
tests/test_pull_ollama_model.py
+193 -348
View File
@@ -1,8 +1,7 @@
"""Unit tests for _pull_ollama_model helper, hcatMarkov steps A-F, and markov_attack handler."""
"""Unit tests for _pull_ollama_model helper, hcatOllama steps A-C, and ollama_attack handler."""
import io
import json
import lzma
import os
import urllib.error
import urllib.request
@@ -25,36 +24,31 @@ MODEL = "llama3.2"
@pytest.fixture
def markov_env(tmp_path):
"""Create the filesystem layout hcatMarkov expects."""
def ollama_env(tmp_path):
"""Create the filesystem layout hcatOllama expects."""
hash_file = tmp_path / "hashes.txt"
hash_file.touch()
hcutil_bin = tmp_path / "hashcat-utils" / "bin"
hcutil_bin.mkdir(parents=True)
hcstat2gen = hcutil_bin / hc_main.hcatHcstat2genBin
hcstat2gen.touch()
wordlist = tmp_path / "sample.txt"
wordlist.write_text("password\n123456\nletmein\n")
return SimpleNamespace(
tmp_path=tmp_path,
hash_file=str(hash_file),
hcstat2gen=str(hcstat2gen),
wordlist=str(wordlist),
)
@contextmanager
def markov_globals(tmp_path, tuning="", potfile=""):
"""Patch the hc_main globals that hcatMarkov reads."""
def ollama_globals(tmp_path, tuning="", potfile=""):
"""Patch the hc_main globals that hcatOllama reads."""
with mock.patch.object(hc_main, "ollamaUrl", OLLAMA_URL), \
mock.patch.object(hc_main, "ollamaModel", MODEL), \
mock.patch.object(hc_main, "hcatBin", "/usr/bin/hashcat"), \
mock.patch.object(hc_main, "hcatTuning", tuning), \
mock.patch.object(hc_main, "hcatPotfilePath", potfile), \
mock.patch.object(hc_main, "hate_path", str(tmp_path)):
mock.patch.object(hc_main, "hate_path", str(tmp_path)), \
mock.patch("hate_crack.main.generate_session_id", return_value="test_session"):
yield
@@ -184,11 +178,11 @@ class TestPullOllamaModel:
# ---------------------------------------------------------------------------
# hcatMarkov 404-retry integration tests
# hcatOllama 404-retry integration tests
# ---------------------------------------------------------------------------
class TestHcatMarkov404Retry:
"""Test that hcatMarkov auto-pulls on 404 and retries the generate call."""
class TestHcatOllama404Retry:
"""Test that hcatOllama auto-pulls on 404 and retries the generate call."""
OLLAMA_URL = "http://localhost:11434"
MODEL = "llama3.2"
@@ -198,16 +192,10 @@ class TestHcatMarkov404Retry:
return io.BytesIO(body)
def _setup_env(self, tmp_path):
"""Create the files/dirs hcatMarkov needs before it hits the API."""
"""Create the files/dirs hcatOllama needs before it hits the API."""
hash_file = str(tmp_path / "hashes.txt")
open(hash_file, "w").close()
# hcatMarkov checks for hcstat2gen binary at startup
hcutil_bin = tmp_path / "hashcat-utils" / "bin"
hcutil_bin.mkdir(parents=True, exist_ok=True)
hcstat2gen = hcutil_bin / hc_main.hcatHcstat2genBin
hcstat2gen.touch()
# Create a small wordlist for "wordlist" mode
wordlist = str(tmp_path / "sample.txt")
with open(wordlist, "w") as f:
@@ -245,12 +233,13 @@ class TestHcatMarkov404Retry:
mock.patch.object(hc_main, "hcatBin", "/usr/bin/hashcat"), \
mock.patch.object(hc_main, "hcatTuning", ""), \
mock.patch.object(hc_main, "hate_path", str(tmp_path)), \
mock.patch("hate_crack.main.generate_session_id", return_value="test_session"), \
mock.patch("subprocess.Popen") as mock_popen:
proc = _make_proc()
mock_popen.return_value = proc
hc_main.hcatMarkov("0", hash_file, "wordlist", wordlist, 10)
hc_main.hcatOllama("0", hash_file, "wordlist", wordlist)
mock_pull.assert_called_once_with(self.OLLAMA_URL, self.MODEL)
# urlopen called twice: first 404, then retry
@@ -259,7 +248,7 @@ class TestHcatMarkov404Retry:
@mock.patch("hate_crack.main._pull_ollama_model")
@mock.patch("hate_crack.main.urllib.request.urlopen")
def test_404_pull_fails_aborts(self, mock_urlopen, mock_pull, capsys, tmp_path):
"""If pull fails after 404, hcatMarkov should abort gracefully."""
"""If pull fails after 404, hcatOllama should abort gracefully."""
hash_file, wordlist = self._setup_env(tmp_path)
mock_urlopen.side_effect = urllib.error.HTTPError(
@@ -275,7 +264,7 @@ class TestHcatMarkov404Retry:
mock.patch.object(hc_main, "ollamaModel", self.MODEL), \
mock.patch.object(hc_main, "hate_path", str(tmp_path)):
hc_main.hcatMarkov("0", hash_file, "wordlist", wordlist, 10)
hc_main.hcatOllama("0", hash_file, "wordlist", wordlist)
captured = capsys.readouterr()
assert "Could not pull model" in captured.out
@@ -298,7 +287,7 @@ class TestHcatMarkov404Retry:
mock.patch.object(hc_main, "ollamaModel", self.MODEL), \
mock.patch.object(hc_main, "hate_path", str(tmp_path)):
hc_main.hcatMarkov("0", hash_file, "wordlist", wordlist, 10)
hc_main.hcatOllama("0", hash_file, "wordlist", wordlist)
captured = capsys.readouterr()
# HTTPError is a subclass of URLError, so it hits the URLError handler
@@ -310,51 +299,36 @@ class TestHcatMarkov404Retry:
# Step A: Mode routing, prompt construction, early-return error paths
# ---------------------------------------------------------------------------
class TestHcatMarkovModeRouting:
class TestHcatOllamaModeRouting:
"""Test mode selection, prompt building, and early-return errors."""
def test_unknown_mode_prints_error(self, markov_env, capsys):
def test_unknown_mode_prints_error(self, ollama_env, capsys):
"""Bad mode string → error message, no API call."""
with markov_globals(markov_env.tmp_path), \
with ollama_globals(ollama_env.tmp_path), \
mock.patch("hate_crack.main.urllib.request.urlopen") as mock_url:
hc_main.hcatMarkov("0", markov_env.hash_file, "bogus", "", 10)
hc_main.hcatOllama("0", ollama_env.hash_file, "bogus", "")
captured = capsys.readouterr()
assert "Unknown Markov generation mode" in captured.out
assert "Unknown LLM generation mode" in captured.out
mock_url.assert_not_called()
def test_missing_wordlist_prints_error(self, markov_env, capsys):
def test_missing_wordlist_prints_error(self, ollama_env, capsys):
"""Non-existent wordlist path → error, no API call."""
with markov_globals(markov_env.tmp_path), \
with ollama_globals(ollama_env.tmp_path), \
mock.patch("hate_crack.main.urllib.request.urlopen") as mock_url:
hc_main.hcatMarkov(
"0", markov_env.hash_file, "wordlist",
"/no/such/wordlist.txt", 10,
hc_main.hcatOllama(
"0", ollama_env.hash_file, "wordlist",
"/no/such/wordlist.txt",
)
captured = capsys.readouterr()
assert "Wordlist not found" in captured.out
mock_url.assert_not_called()
def test_missing_hcstat2gen_prints_error(self, tmp_path, capsys):
"""No hcstat2gen binary → error, no API call."""
hash_file = tmp_path / "hashes.txt"
hash_file.touch()
wordlist = tmp_path / "sample.txt"
wordlist.write_text("password\n")
with markov_globals(tmp_path), \
mock.patch("hate_crack.main.urllib.request.urlopen") as mock_url:
hc_main.hcatMarkov("0", str(hash_file), "wordlist", str(wordlist), 10)
captured = capsys.readouterr()
assert "hcstat2gen not found" in captured.out
mock_url.assert_not_called()
def test_wordlist_mode_reads_up_to_500_lines(self, markov_env, capsys):
def test_wordlist_mode_reads_up_to_500_lines(self, ollama_env, capsys):
"""Only the first 500 non-blank lines should appear in the prompt payload."""
# Write 600 lines to the wordlist
big_wordlist = markov_env.tmp_path / "big.txt"
big_wordlist = ollama_env.tmp_path / "big.txt"
big_wordlist.write_text("\n".join(f"pass{i}" for i in range(600)) + "\n")
captured_payload = {}
@@ -363,13 +337,13 @@ class TestHcatMarkovModeRouting:
captured_payload["data"] = json.loads(req.data.decode())
return _generate_response(["Password1"])
with markov_globals(markov_env.tmp_path), \
with ollama_globals(ollama_env.tmp_path), \
mock.patch("hate_crack.main.urllib.request.urlopen", side_effect=fake_urlopen), \
mock.patch("subprocess.Popen") as mock_popen:
proc = _make_proc()
mock_popen.return_value = proc
hc_main.hcatMarkov(
"0", markov_env.hash_file, "wordlist", str(big_wordlist), 10,
hc_main.hcatOllama(
"0", ollama_env.hash_file, "wordlist", str(big_wordlist),
)
prompt_text = captured_payload["data"]["prompt"]
@@ -377,7 +351,7 @@ class TestHcatMarkovModeRouting:
assert "pass499" in prompt_text
assert "pass500" not in prompt_text
def test_target_mode_includes_context_in_prompt(self, markov_env, capsys):
def test_target_mode_includes_context_in_prompt(self, ollama_env, capsys):
"""Company, industry, and location should appear in the prompt."""
captured_payload = {}
@@ -390,13 +364,13 @@ class TestHcatMarkovModeRouting:
"industry": "Finance",
"location": "New York",
}
with markov_globals(markov_env.tmp_path), \
with ollama_globals(ollama_env.tmp_path), \
mock.patch("hate_crack.main.urllib.request.urlopen", side_effect=fake_urlopen), \
mock.patch("subprocess.Popen") as mock_popen:
proc = _make_proc()
mock_popen.return_value = proc
hc_main.hcatMarkov(
"0", markov_env.hash_file, "target", target_info, 10,
hc_main.hcatOllama(
"0", ollama_env.hash_file, "target", target_info,
)
prompt_text = captured_payload["data"]["prompt"]
@@ -409,33 +383,37 @@ class TestHcatMarkovModeRouting:
# Step B: Candidate filtering / post-processing
# ---------------------------------------------------------------------------
class TestHcatMarkovCandidateFiltering:
class TestHcatOllamaCandidateFiltering:
"""Test regex stripping, blank-line removal, 128-char limit, empty-result handling."""
def _run_with_response(self, markov_env, response_text):
"""Run hcatMarkov with a canned Ollama response_text, return candidates file content."""
def _run_with_response(self, ollama_env, response_text):
"""Run hcatOllama with a canned Ollama response_text, return candidates file content."""
body = json.dumps({"response": response_text}).encode()
resp = mock.MagicMock()
resp.__enter__ = mock.Mock(return_value=io.BytesIO(body))
resp.__exit__ = mock.Mock(return_value=False)
with markov_globals(markov_env.tmp_path), \
candidates_path = f"{ollama_env.hash_file}.ollama_candidates"
captured = {}
def capture_popen(cmd, **kwargs):
# Read the candidates file before hashcat "runs" (cleanup happens after)
if os.path.isfile(candidates_path):
with open(candidates_path) as f:
captured["content"] = f.read()
return _make_proc()
with ollama_globals(ollama_env.tmp_path), \
mock.patch("hate_crack.main.urllib.request.urlopen", return_value=resp), \
mock.patch("subprocess.Popen") as mock_popen:
proc = _make_proc()
mock_popen.return_value = proc
hc_main.hcatMarkov(
"0", markov_env.hash_file, "wordlist", markov_env.wordlist, 10,
mock.patch("subprocess.Popen", side_effect=capture_popen):
hc_main.hcatOllama(
"0", ollama_env.hash_file, "wordlist", ollama_env.wordlist,
)
candidates_path = f"{markov_env.hash_file}.markov_candidates"
if os.path.isfile(candidates_path):
with open(candidates_path) as f:
return f.read()
return None
return captured.get("content")
def test_strips_numeric_dot_prefix(self, markov_env):
content = self._run_with_response(markov_env, "1. Password1\n2. Summer2024")
def test_strips_numeric_dot_prefix(self, ollama_env):
content = self._run_with_response(ollama_env, "1. Password1\n2. Summer2024")
assert content is not None
lines = [l for l in content.strip().split("\n") if l]
assert "Password1" in lines
@@ -445,51 +423,51 @@ class TestHcatMarkovCandidateFiltering:
assert not line.startswith("1.")
assert not line.startswith("2.")
def test_strips_dash_and_asterisk_prefix(self, markov_env):
content = self._run_with_response(markov_env, "- foo\n* bar")
def test_strips_dash_and_asterisk_prefix(self, ollama_env):
content = self._run_with_response(ollama_env, "- foo\n* bar")
assert content is not None
lines = [l for l in content.strip().split("\n") if l]
assert "foo" in lines
assert "bar" in lines
def test_skips_blank_lines(self, markov_env):
content = self._run_with_response(markov_env, "alpha\n\n\nbeta\n\n")
def test_skips_blank_lines(self, ollama_env):
content = self._run_with_response(ollama_env, "alpha\n\n\nbeta\n\n")
assert content is not None
lines = [l for l in content.strip().split("\n") if l]
assert lines == ["alpha", "beta"]
def test_rejects_over_128_chars(self, markov_env):
def test_rejects_over_128_chars(self, ollama_env):
long_pw = "A" * 129
content = self._run_with_response(markov_env, f"short\n{long_pw}\nkeep")
content = self._run_with_response(ollama_env, f"short\n{long_pw}\nkeep")
assert content is not None
lines = [l for l in content.strip().split("\n") if l]
assert "short" in lines
assert "keep" in lines
assert long_pw not in lines
def test_accepts_exactly_128_chars(self, markov_env):
def test_accepts_exactly_128_chars(self, ollama_env):
exact_pw = "B" * 128
content = self._run_with_response(markov_env, f"{exact_pw}\nother")
content = self._run_with_response(ollama_env, f"{exact_pw}\nother")
assert content is not None
lines = [l for l in content.strip().split("\n") if l]
assert exact_pw in lines
def test_empty_response_prints_error(self, markov_env, capsys):
self._run_with_response(markov_env, "")
def test_empty_response_prints_error(self, ollama_env, capsys):
self._run_with_response(ollama_env, "")
captured = capsys.readouterr()
assert "no usable" in captured.out.lower()
def test_missing_response_key(self, markov_env, capsys):
def test_missing_response_key(self, ollama_env, capsys):
"""If the JSON has no 'response' key, treat as empty."""
body = json.dumps({"other": "stuff"}).encode()
resp = mock.MagicMock()
resp.__enter__ = mock.Mock(return_value=io.BytesIO(body))
resp.__exit__ = mock.Mock(return_value=False)
with markov_globals(markov_env.tmp_path), \
with ollama_globals(ollama_env.tmp_path), \
mock.patch("hate_crack.main.urllib.request.urlopen", return_value=resp):
hc_main.hcatMarkov(
"0", markov_env.hash_file, "wordlist", markov_env.wordlist, 10,
hc_main.hcatOllama(
"0", ollama_env.hash_file, "wordlist", ollama_env.wordlist,
)
captured = capsys.readouterr()
@@ -500,37 +478,37 @@ class TestHcatMarkovCandidateFiltering:
# Step B: API error paths (non-404)
# ---------------------------------------------------------------------------
class TestHcatMarkovApiErrors:
class TestHcatOllamaApiErrors:
"""Test connection errors and generic exceptions during the generate call."""
def test_url_error_prints_connection_error(self, markov_env, capsys):
with markov_globals(markov_env.tmp_path), \
def test_url_error_prints_connection_error(self, ollama_env, capsys):
with ollama_globals(ollama_env.tmp_path), \
mock.patch(
"hate_crack.main.urllib.request.urlopen",
side_effect=urllib.error.URLError("Connection refused"),
):
hc_main.hcatMarkov(
"0", markov_env.hash_file, "wordlist", markov_env.wordlist, 10,
hc_main.hcatOllama(
"0", ollama_env.hash_file, "wordlist", ollama_env.wordlist,
)
captured = capsys.readouterr()
assert "Could not connect" in captured.out
assert "Ensure Ollama is running" in captured.out
def test_generic_exception_prints_error(self, markov_env, capsys):
with markov_globals(markov_env.tmp_path), \
def test_generic_exception_prints_error(self, ollama_env, capsys):
with ollama_globals(ollama_env.tmp_path), \
mock.patch(
"hate_crack.main.urllib.request.urlopen",
side_effect=RuntimeError("boom"),
):
hc_main.hcatMarkov(
"0", markov_env.hash_file, "wordlist", markov_env.wordlist, 10,
hc_main.hcatOllama(
"0", ollama_env.hash_file, "wordlist", ollama_env.wordlist,
)
captured = capsys.readouterr()
assert "Error calling Ollama API" in captured.out
def test_generate_request_payload(self, markov_env):
def test_generate_request_payload(self, ollama_env):
"""Verify the /api/generate request has correct URL, model, stream=false."""
captured_req = {}
@@ -539,13 +517,13 @@ class TestHcatMarkovApiErrors:
captured_req["body"] = json.loads(req.data.decode())
return _generate_response(["Password1"])
with markov_globals(markov_env.tmp_path), \
with ollama_globals(ollama_env.tmp_path), \
mock.patch("hate_crack.main.urllib.request.urlopen", side_effect=fake_urlopen), \
mock.patch("subprocess.Popen") as mock_popen:
proc = _make_proc()
mock_popen.return_value = proc
hc_main.hcatMarkov(
"0", markov_env.hash_file, "wordlist", markov_env.wordlist, 10,
hc_main.hcatOllama(
"0", ollama_env.hash_file, "wordlist", ollama_env.wordlist,
)
assert captured_req["url"] == f"{OLLAMA_URL}/api/generate"
@@ -555,269 +533,136 @@ class TestHcatMarkovApiErrors:
# ---------------------------------------------------------------------------
# Steps C + D: hcstat2gen execution and LZMA compression
# Step C: Hashcat command construction
# ---------------------------------------------------------------------------
class TestHcatMarkovHcstat2gen:
"""Test hcstat2gen subprocess step."""
class TestHcatOllamaHashcatCommand:
"""Test hashcat command flags and process handling."""
def test_hcstat2gen_correct_args(self, markov_env):
"""hcstat2gen should be called with [binary_path, raw_output_path] and stdin=file."""
hcstat2_raw_path = f"{markov_env.hash_file}.hcstat2_raw"
hcstat2gen_path = markov_env.hcstat2gen
call_args_list = []
def _run_full(self, ollama_env, tuning="", potfile=""):
"""Run hcatOllama through all steps, returning the list of Popen calls."""
popen_calls = []
def track_popen(cmd, **kwargs):
call_args_list.append((list(cmd), dict(kwargs)))
proc = _make_proc()
# Create raw hcstat2 file so the LZMA compression step continues
if cmd[0] == hcstat2gen_path:
with open(hcstat2_raw_path, "wb") as f:
f.write(b"\x00" * 100)
return proc
popen_calls.append((list(cmd), dict(kwargs)))
return _make_proc()
with markov_globals(markov_env.tmp_path), \
with ollama_globals(ollama_env.tmp_path, tuning=tuning, potfile=potfile), \
_urlopen_with_response(["Password1"]), \
mock.patch("subprocess.Popen", side_effect=track_popen), \
mock.patch("hate_crack.main.generate_session_id", return_value="test_session"):
hc_main.hcatMarkov(
"0", markov_env.hash_file, "wordlist", markov_env.wordlist, 10,
hc_main.hcatOllama(
"1000", ollama_env.hash_file, "wordlist", ollama_env.wordlist,
)
# First Popen call should be hcstat2gen writing to the raw path
first_cmd, first_kwargs = call_args_list[0]
assert first_cmd[0] == hcstat2gen_path
assert first_cmd[1] == hcstat2_raw_path
assert "stdin" in first_kwargs
return popen_calls
def test_hcstat2gen_keyboard_interrupt(self, markov_env, capsys):
"""KeyboardInterrupt during hcstat2gen should kill the process and return."""
def test_hashcat_command_structure(self, ollama_env):
"""First Popen call should be a straight wordlist attack with candidates file,
followed by rule-based attacks."""
calls = self._run_full(ollama_env)
assert len(calls) >= 1, "Expected at least one Popen call (hashcat)"
# First call: base wordlist attack (no rules)
hashcat_cmd = calls[0][0]
assert hashcat_cmd[0] == "/usr/bin/hashcat"
assert "-m" in hashcat_cmd
m_idx = hashcat_cmd.index("-m")
assert hashcat_cmd[m_idx + 1] == "1000"
# Should use the candidates wordlist file
candidates_path = f"{ollama_env.hash_file}.ollama_candidates"
assert candidates_path in hashcat_cmd
# First call should NOT have mask attack flags
assert "-a" not in hashcat_cmd
assert "--markov-hcstat2" not in " ".join(hashcat_cmd)
assert "--increment" not in hashcat_cmd
# First call should NOT have -r (rule) flag
assert "-r" not in hashcat_cmd
# Subsequent calls: rule-based attacks
if len(calls) > 1:
for rule_cmd, _ in calls[1:]:
assert "-r" in rule_cmd
assert candidates_path in rule_cmd
def test_hashcat_includes_tuning(self, ollama_env):
"""-w 3 tuning flag should be appended to hashcat cmd."""
calls = self._run_full(ollama_env, tuning="-w 3")
hashcat_cmd = calls[0][0]
assert "-w" in hashcat_cmd
w_idx = hashcat_cmd.index("-w")
assert hashcat_cmd[w_idx + 1] == "3"
def test_hashcat_includes_potfile(self, ollama_env):
"""--potfile-path should be present when hcatPotfilePath is set."""
calls = self._run_full(ollama_env, potfile="/tmp/test.potfile")
hashcat_cmd = calls[0][0]
potfile_flags = [f for f in hashcat_cmd if "--potfile-path=" in f]
assert len(potfile_flags) == 1
assert potfile_flags[0] == "--potfile-path=/tmp/test.potfile"
def test_hashcat_keyboard_interrupt(self, ollama_env, capsys):
"""KeyboardInterrupt during hashcat should kill the process."""
proc = mock.MagicMock()
proc.communicate.side_effect = KeyboardInterrupt()
proc.pid = 12345
proc.pid = 99999
proc.wait.side_effect = KeyboardInterrupt()
with markov_globals(markov_env.tmp_path), \
with ollama_globals(ollama_env.tmp_path), \
_urlopen_with_response(["Password1"]), \
mock.patch("subprocess.Popen", return_value=proc):
hc_main.hcatMarkov(
"0", markov_env.hash_file, "wordlist", markov_env.wordlist, 10,
mock.patch("subprocess.Popen", return_value=proc), \
mock.patch("hate_crack.main.generate_session_id", return_value="test_session"):
hc_main.hcatOllama(
"0", ollama_env.hash_file, "wordlist", ollama_env.wordlist,
)
captured = capsys.readouterr()
assert "Killing PID" in captured.out
proc.kill.assert_called_once()
def test_hcstat2gen_no_output_file(self, markov_env, capsys):
"""If hcstat2gen doesn't produce a raw output file, abort with error."""
proc = mock.MagicMock()
proc.communicate.return_value = (b"", b"")
proc.returncode = 0
# Don't create hcstat2_raw_path → triggers the "did not produce" error
with markov_globals(markov_env.tmp_path), \
_urlopen_with_response(["Password1"]), \
mock.patch("subprocess.Popen", return_value=proc):
hc_main.hcatMarkov(
"0", markov_env.hash_file, "wordlist", markov_env.wordlist, 10,
)
captured = capsys.readouterr()
assert "did not produce output file" in captured.out
def test_lzma_compression_error(self, markov_env, capsys):
"""LZMAError during compression → error message, no hashcat run."""
hcstat2_raw_path = f"{markov_env.hash_file}.hcstat2_raw"
def track_popen(cmd, **kwargs):
proc = _make_proc()
if cmd[0] == markov_env.hcstat2gen:
with open(hcstat2_raw_path, "wb") as f:
f.write(b"\x00" * 100)
return proc
with markov_globals(markov_env.tmp_path), \
_urlopen_with_response(["Password1"]), \
mock.patch("subprocess.Popen", side_effect=track_popen) as mock_popen, \
mock.patch("hate_crack.main.lzma.compress", side_effect=lzma.LZMAError("bad data")):
hc_main.hcatMarkov(
"0", markov_env.hash_file, "wordlist", markov_env.wordlist, 10,
)
captured = capsys.readouterr()
assert "Error compressing" in captured.out
# Should only have 1 Popen call (hcstat2gen), NOT a second for hashcat
assert mock_popen.call_count == 1
# ---------------------------------------------------------------------------
# Step E: Hashcat command construction
# ---------------------------------------------------------------------------
class TestHcatMarkovHashcatCommand:
"""Test hashcat command flags and process handling."""
def _run_full(self, markov_env, tuning="", potfile=""):
"""Run hcatMarkov through all steps, returning the list of Popen calls."""
hcstat2_raw_path = f"{markov_env.hash_file}.hcstat2_raw"
popen_calls = []
def track_popen(cmd, **kwargs):
popen_calls.append((list(cmd), dict(kwargs)))
proc = _make_proc()
# hcstat2gen: create the raw output file (LZMA2 compression step will follow)
if cmd[0] == markov_env.hcstat2gen:
with open(hcstat2_raw_path, "wb") as f:
f.write(b"\x00" * 100)
return proc
with markov_globals(markov_env.tmp_path, tuning=tuning, potfile=potfile), \
_urlopen_with_response(["Password1"]), \
mock.patch("subprocess.Popen", side_effect=track_popen), \
mock.patch("hate_crack.main.generate_session_id", return_value="test_session"):
hc_main.hcatMarkov(
"1000", markov_env.hash_file, "wordlist", markov_env.wordlist, 10,
)
return popen_calls
def test_hashcat_command_structure(self, markov_env):
"""Hashcat cmd should include -m, -a 3, --markov-hcstat2, --increment, mask."""
calls = self._run_full(markov_env)
assert len(calls) >= 2, "Expected at least hcstat2gen + hashcat Popen calls"
hashcat_cmd = calls[1][0]
assert hashcat_cmd[0] == "/usr/bin/hashcat"
assert "-m" in hashcat_cmd
m_idx = hashcat_cmd.index("-m")
assert hashcat_cmd[m_idx + 1] == "1000"
assert "-a" in hashcat_cmd
a_idx = hashcat_cmd.index("-a")
assert hashcat_cmd[a_idx + 1] == "3"
# Check markov hcstat2 flag
hcstat2_flags = [f for f in hashcat_cmd if "--markov-hcstat2=" in f]
assert len(hcstat2_flags) == 1
assert hcstat2_flags[0].endswith(".hcstat2")
assert "--increment" in hashcat_cmd
# Mask should be the last positional arg
assert "?a?a?a?a?a?a?a?a?a?a?a?a?a?a" in hashcat_cmd
def test_hashcat_includes_tuning(self, markov_env):
"""-w 3 tuning flag should be appended to hashcat cmd."""
calls = self._run_full(markov_env, tuning="-w 3")
hashcat_cmd = calls[1][0]
assert "-w" in hashcat_cmd
w_idx = hashcat_cmd.index("-w")
assert hashcat_cmd[w_idx + 1] == "3"
def test_hashcat_includes_potfile(self, markov_env):
"""--potfile-path should be present when hcatPotfilePath is set."""
calls = self._run_full(markov_env, potfile="/tmp/test.potfile")
hashcat_cmd = calls[1][0]
potfile_flags = [f for f in hashcat_cmd if "--potfile-path=" in f]
assert len(potfile_flags) == 1
assert potfile_flags[0] == "--potfile-path=/tmp/test.potfile"
def test_hashcat_keyboard_interrupt(self, markov_env, capsys):
"""KeyboardInterrupt during hashcat should kill the process."""
hcstat2_raw_path = f"{markov_env.hash_file}.hcstat2_raw"
call_count = [0]
def track_popen(cmd, **kwargs):
call_count[0] += 1
if call_count[0] == 1:
# hcstat2gen: succeeds and creates raw file
proc = _make_proc()
proc.pid = 99999
with open(hcstat2_raw_path, "wb") as f:
f.write(b"\x00" * 100)
else:
# hashcat: KeyboardInterrupt
proc = mock.MagicMock()
proc.pid = 99999
proc.wait.side_effect = KeyboardInterrupt()
return proc
with markov_globals(markov_env.tmp_path), \
_urlopen_with_response(["Password1"]), \
mock.patch("subprocess.Popen", side_effect=track_popen), \
mock.patch("hate_crack.main.generate_session_id", return_value="test_session"):
hc_main.hcatMarkov(
"0", markov_env.hash_file, "wordlist", markov_env.wordlist, 10,
)
captured = capsys.readouterr()
assert "Killing PID" in captured.out
# ---------------------------------------------------------------------------
# Step F: Cleanup
# ---------------------------------------------------------------------------
class TestHcatMarkovCleanup:
"""Test that temp files are removed after a successful run."""
class TestHcatOllamaWordlistPersistence:
"""Test that the generated wordlist file persists after the run."""
def _run_full(self, markov_env):
"""Run hcatMarkov through all steps successfully."""
hcstat2_raw_path = f"{markov_env.hash_file}.hcstat2_raw"
def track_popen(cmd, **kwargs):
proc = _make_proc()
if cmd[0] == markov_env.hcstat2gen:
with open(hcstat2_raw_path, "wb") as f:
f.write(b"\x00" * 100)
return proc
with markov_globals(markov_env.tmp_path), \
def test_candidates_file_persists(self, ollama_env):
"""ollama_candidates wordlist should remain after the run."""
with ollama_globals(ollama_env.tmp_path), \
_urlopen_with_response(["Password1"]), \
mock.patch("subprocess.Popen", side_effect=track_popen), \
mock.patch("subprocess.Popen", return_value=_make_proc()), \
mock.patch("hate_crack.main.generate_session_id", return_value="test_session"):
hc_main.hcatMarkov(
"0", markov_env.hash_file, "wordlist", markov_env.wordlist, 10,
hc_main.hcatOllama(
"0", ollama_env.hash_file, "wordlist", ollama_env.wordlist,
)
def test_temp_files_removed(self, markov_env):
"""markov_candidates and hcstat2_raw should be deleted; hcstat2 should remain."""
self._run_full(markov_env)
candidates_path = f"{markov_env.hash_file}.markov_candidates"
hcstat2_raw_path = f"{markov_env.hash_file}.hcstat2_raw"
hcstat2_path = f"{markov_env.hash_file}.hcstat2"
assert not os.path.exists(candidates_path), "candidates temp file should be removed"
assert not os.path.exists(hcstat2_raw_path), "raw hcstat2 temp file should be removed"
assert os.path.isfile(hcstat2_path), "compressed hcstat2 file should remain"
def test_cleanup_ignores_missing_files(self, markov_env):
"""No exception if temp files are already gone before cleanup."""
# This just verifies the run completes without error even though
# we don't interfere with normal cleanup
self._run_full(markov_env) # should not raise
candidates_path = f"{ollama_env.hash_file}.ollama_candidates"
assert os.path.isfile(candidates_path), "candidates wordlist should persist"
# ---------------------------------------------------------------------------
# attacks.py: markov_attack() UI handler
# attacks.py: ollama_attack() UI handler
# ---------------------------------------------------------------------------
class TestMarkovAttackHandler:
"""Test the markov_attack(ctx) menu handler in attacks.py."""
class TestOllamaAttackHandler:
"""Test the ollama_attack(ctx) menu handler in attacks.py."""
def _make_ctx(self, **overrides):
"""Build a mock ctx with the attributes markov_attack() reads."""
"""Build a mock ctx with the attributes ollama_attack() reads."""
ctx = mock.MagicMock()
ctx.hcatHashType = "0"
ctx.hcatHashFile = "/tmp/hashes.txt"
ctx.markovWordlist = "/tmp/wordlist.txt"
ctx.markovCandidateCount = 5000
ctx.ollamaWordlist = "/tmp/wordlist.txt"
ctx.ollamaCandidateCount = 5000
ctx.hcatWordlists = "/tmp/wordlists"
for k, v in overrides.items():
setattr(ctx, k, v)
return ctx
def test_wordlist_mode_default(self, tmp_path):
"""When cracked output exists, it becomes the default wordlist."""
from hate_crack.attacks import markov_attack
"""Default wordlist is always ollamaWordlist from config."""
from hate_crack.attacks import ollama_attack
hash_file = str(tmp_path / "hashes.txt")
cracked_out = hash_file + ".out"
@@ -826,51 +671,51 @@ class TestMarkovAttackHandler:
ctx = self._make_ctx(hcatHashFile=hash_file)
with mock.patch("builtins.input", side_effect=["1", "n", ""]):
markov_attack(ctx)
ollama_attack(ctx)
ctx.hcatMarkov.assert_called_once_with(
"0", hash_file, "wordlist", cracked_out, 5000,
ctx.hcatOllama.assert_called_once_with(
"0", hash_file, "wordlist", "/tmp/wordlist.txt",
)
def test_wordlist_mode_falls_back_to_config(self):
"""When cracked output does not exist, fall back to markovWordlist."""
from hate_crack.attacks import markov_attack
"""When cracked output does not exist, fall back to ollamaWordlist."""
from hate_crack.attacks import ollama_attack
ctx = self._make_ctx(hcatHashFile="/tmp/nonexistent_hashes.txt")
with mock.patch("builtins.input", side_effect=["1", "n", ""]):
markov_attack(ctx)
ollama_attack(ctx)
ctx.hcatMarkov.assert_called_once_with(
"0", "/tmp/nonexistent_hashes.txt", "wordlist", "/tmp/wordlist.txt", 5000,
ctx.hcatOllama.assert_called_once_with(
"0", "/tmp/nonexistent_hashes.txt", "wordlist", "/tmp/wordlist.txt",
)
def test_wordlist_mode_custom_path(self):
"""Selection '1', override wordlist → resolved path passed to hcatMarkov."""
from hate_crack.attacks import markov_attack
"""Selection '1', override wordlist → resolved path passed to hcatOllama."""
from hate_crack.attacks import ollama_attack
ctx = self._make_ctx()
ctx.select_file_with_autocomplete.return_value = "custom.txt"
ctx._resolve_wordlist_path.return_value = "/resolved/custom.txt"
with mock.patch("builtins.input", side_effect=["1", "y", ""]):
markov_attack(ctx)
ollama_attack(ctx)
ctx.select_file_with_autocomplete.assert_called_once()
ctx._resolve_wordlist_path.assert_called_once_with("custom.txt", "/tmp/wordlists")
ctx.hcatMarkov.assert_called_once_with(
"0", "/tmp/hashes.txt", "wordlist", "/resolved/custom.txt", 5000,
ctx.hcatOllama.assert_called_once_with(
"0", "/tmp/hashes.txt", "wordlist", "/resolved/custom.txt",
)
def test_target_mode(self):
"""Selection '2' → hcatMarkov('target', {company, industry, location})."""
from hate_crack.attacks import markov_attack
"""Selection '2' → hcatOllama('target', {company, industry, location})."""
from hate_crack.attacks import ollama_attack
ctx = self._make_ctx()
with mock.patch("builtins.input", side_effect=["2", "AcmeCorp", "Finance", "NYC", ""]):
markov_attack(ctx)
ollama_attack(ctx)
ctx.hcatMarkov.assert_called_once()
call_args = ctx.hcatMarkov.call_args
ctx.hcatOllama.assert_called_once()
call_args = ctx.hcatOllama.call_args
assert call_args[0][2] == "target"
target_info = call_args[0][3]
assert target_info["company"] == "AcmeCorp"
@@ -878,28 +723,28 @@ class TestMarkovAttackHandler:
assert target_info["location"] == "NYC"
def test_invalid_selection(self, capsys):
"""Selection '3' → 'Invalid selection.', no hcatMarkov call."""
from hate_crack.attacks import markov_attack
"""Selection '3' → 'Invalid selection.', no hcatOllama call."""
from hate_crack.attacks import ollama_attack
ctx = self._make_ctx()
with mock.patch("builtins.input", side_effect=["3"]):
markov_attack(ctx)
ollama_attack(ctx)
captured = capsys.readouterr()
assert "Invalid selection" in captured.out
ctx.hcatMarkov.assert_not_called()
ctx.hcatOllama.assert_not_called()
def test_wordlist_list_uses_first(self):
"""When markovWordlist is a list and no cracked output exists, the first element is used."""
from hate_crack.attacks import markov_attack
"""When ollamaWordlist is a list and no cracked output exists, the first element is used."""
from hate_crack.attacks import ollama_attack
ctx = self._make_ctx(
hcatHashFile="/tmp/nonexistent_hashes.txt",
markovWordlist=["/tmp/first.txt", "/tmp/second.txt"],
ollamaWordlist=["/tmp/first.txt", "/tmp/second.txt"],
)
with mock.patch("builtins.input", side_effect=["1", "n", ""]):
markov_attack(ctx)
ollama_attack(ctx)
ctx.hcatMarkov.assert_called_once()
call_args = ctx.hcatMarkov.call_args
ctx.hcatOllama.assert_called_once()
call_args = ctx.hcatOllama.call_args
assert call_args[0][3] == "/tmp/first.txt"