automatic hash_id detection via api for hashview hashes

This commit is contained in:
Justin Bollinger
2026-02-05 14:56:27 -05:00
parent 9756f83b0c
commit af5fe15ec4
3 changed files with 280 additions and 129 deletions
+148 -45
View File
@@ -101,6 +101,22 @@ def get_rules_dir():
return default
def get_hcat_tuning_args():
config_path = _resolve_config_path()
if config_path:
try:
with open(config_path) as f:
config = json.load(f)
tuning = config.get("hcatTuning")
if tuning:
import shlex
return shlex.split(tuning)
except Exception:
pass
return []
def cleanup_torrent_files(directory=None):
"""Remove stray .torrent files from the wordlists directory on graceful exit."""
if directory is None:
@@ -566,11 +582,17 @@ class HashviewAPI:
resp.raise_for_status()
try:
data = resp.json()
except Exception:
except Exception as e:
if self.debug:
print(f"[DEBUG] Failed to parse JSON from {url}: {e}")
data = None
hashtype = None
if data:
hashtype = data.get("hashtype") or data.get("hash_type")
hashtype = data.get("hashtype") or data.get("hash_type") or data.get("type")
if self.debug:
print(f"[DEBUG] get_hashfile_details({hashfile_id}): raw data={data}, hashtype={hashtype}")
elif self.debug:
print(f"[DEBUG] get_hashfile_details({hashfile_id}): no data returned. raw response: {resp.text}")
return {
"hashfile_id": hashfile_id,
"hashtype": hashtype,
@@ -612,9 +634,34 @@ class HashviewAPI:
def get_customer_hashfiles(self, customer_id):
all_hashfiles = self.list_hashfiles()
return [
customer_hfs = [
hf for hf in all_hashfiles if int(hf.get("customer_id", 0)) == customer_id
]
if self.debug:
print(f"[DEBUG] get_customer_hashfiles({customer_id}): found {len(customer_hfs)} hashfiles")
# Fetch hash types for any hashfiles missing them
for hf in customer_hfs:
if not (hf.get("hashtype") or hf.get("hash_type")):
hf_id = hf.get("id")
if hf_id is not None:
if self.debug:
print(f"[DEBUG] Fetching hash_type for hashfile {hf_id}")
try:
details = self.get_hashfile_details(hf_id)
hashtype = details.get("hashtype")
if hashtype:
hf["hash_type"] = hashtype
if self.debug:
print(f"[DEBUG] Updated hashfile {hf_id} with hash_type={hashtype}")
elif self.debug:
print(f"[DEBUG] No hashtype found in details for {hf_id}: {details}")
except Exception as e:
if self.debug:
print(f"[DEBUG] Exception fetching hash_type for {hf_id}: {e}")
return customer_hfs
def get_customer_hashfiles_with_hashtype(self, customer_id, target_hashtype="1000"):
"""Return hashfiles for a customer that match the requested hashtype."""
@@ -739,9 +786,10 @@ class HashviewAPI:
resp.raise_for_status()
return resp.json()
def download_left_hashes(self, customer_id, hashfile_id, output_file=None):
def download_left_hashes(self, customer_id, hashfile_id, output_file=None, hash_type=None):
import sys
import re
import subprocess
url = f"{self.base_url}/v1/hashfiles/{hashfile_id}/left"
resp = self.session.get(url, stream=True)
@@ -772,7 +820,7 @@ class HashviewAPI:
if total == 0:
print(f"Downloaded {downloaded} bytes.")
# Try to download found file and merge with corresponding .out file if it exists
# Try to download found file and process with hashcat
combined_count = 0
combined_file = None
out_dir = os.path.dirname(output_abs) or os.getcwd()
@@ -796,59 +844,114 @@ class HashviewAPI:
if chunk:
f.write(chunk)
# Determine the .out file (either .txt.out for regular or .nt.txt.out for pwdump)
out_file = output_abs + ".out"
if not os.path.exists(out_file):
# Check for pwdump format .nt.txt.out file if caller used the .txt name.
if out_file.endswith(".txt.out") and not out_file.endswith(".nt.txt.out"):
out_file = out_file.replace(".txt.out", ".nt.txt.out")
# Split found file into hashes and clears
found_hashes_file = os.path.join(out_dir, f"found_hashes_{customer_id}_{hashfile_id}.txt")
found_clears_file = os.path.join(out_dir, f"found_clears_{customer_id}_{hashfile_id}.txt")
# Only merge if the .out file exists
if os.path.exists(out_file):
# Read existing hashes from .out file into a set
out_hashes = set()
with open(out_file, "r", encoding="utf-8", errors="ignore") as f:
for line in f:
line = line.strip()
if line:
out_hashes.add(line)
original_count = len(out_hashes)
# Read and add hashes from the found file
hashes_count = 0
clears_count = 0
with open(found_hashes_file, "w", encoding="utf-8") as hf, \
open(found_clears_file, "w", encoding="utf-8") as cf:
with open(found_file, "r", encoding="utf-8", errors="ignore") as f:
for line in f:
line = line.strip()
if line:
out_hashes.add(line)
combined_count = len(out_hashes) - original_count
# Write combined results back to the .out file
combined_file = out_file
with open(combined_file, "w", encoding="utf-8") as f:
for line in sorted(out_hashes):
f.write(line + "\n")
print(f"Merged {combined_count} new hashes from {found_file}")
print(f"Total unique hashes in {combined_file}: {len(out_hashes)}")
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
# Delete the found file after successful merge
print(f"Split found file into {hashes_count} hashes and {clears_count} clears")
# Run hashcat to combine them
combined_file = output_abs + ".out"
try:
os.remove(found_file)
print(f"Deleted {found_file}")
# Execute hashcat: hashcat <tuning> -m hash_type found_hashes found_clears --outfile output.out --outfile-format=1,2
tuning_args = get_hcat_tuning_args()
# 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(f"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"Warning: Could not delete {found_file}: {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}")
else:
print("Debug enabled: keeping found and split files")
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
try:
if os.path.exists(found_file):
os.remove(found_file)
except Exception:
pass
if not self.debug:
try:
if os.path.exists(found_file):
os.remove(found_file)
except Exception:
pass
return {
"output_file": output_file,
+123 -68
View File
@@ -2157,90 +2157,127 @@ def hashview_api():
elif choice == "4":
# Download left hashes
try:
# 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.")
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.")
# 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
print(f" Customer ID: {customer_id}")
except Exception as e:
print(f"\n✗ Error creating customer: {str(e)}")
else:
print("\n✗ Error: Customer name cannot be empty.")
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_id = int(customer_input)
except ValueError:
print(
"\n✗ Error: Invalid ID entered. Please enter a numeric ID or N."
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
# List hashfiles for the customer
try:
customer_hashfiles = api_harness.get_customer_hashfiles(
customer_id
)
if customer_hashfiles:
print("\n" + "=" * 100)
print(f"Hashfiles for Customer ID {customer_id}:")
print("=" * 100)
print(f"{'ID':<10} {'Name':<88}")
print("-" * 100)
for hf in customer_hashfiles:
hf_id = hf.get("id", "N/A")
hf_name = hf.get("name", "N/A")
# Truncate long names to fit within 100 columns
if len(str(hf_name)) > 88:
hf_name = str(hf_name)[:85] + "..."
print(f"{hf_id:<10} {hf_name:<88}")
print("=" * 100)
print(f"Total: {len(customer_hashfiles)} hashfile(s)")
else:
print(f"\nNo hashfiles found for customer ID {customer_id}")
except Exception as e:
print(f"\nWarning: Could not list hashfiles: {e}")
print(
"You may need to manually find the hashfile ID in the web interface."
)
hashfile_id = int(input("\nEnter hashfile ID: "))
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"left_{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 left hashes
if debug_mode:
print(f"[DEBUG] Calling download_left_hashes with hash_type={selected_hash_type}")
download_result = api_harness.download_left_hashes(
customer_id, hashfile_id, output_file
customer_id, hashfile_id, output_file, hash_type=selected_hash_type
)
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}")
# Ask if user wants to switch to this hashfile
switch = (
@@ -2250,7 +2287,10 @@ def hashview_api():
)
if switch != "n":
hcatHashFile = download_result["output_file"]
hcatHashType = "1000" # Default to NTLM for Hashview downloads
if selected_hash_type:
hcatHashType = str(selected_hash_type)
else:
hcatHashType = "1000" # Default to NTLM if unavailable
print(f"✓ Switched to hashfile: {hcatHashFile}")
print("\nReturning to main menu to start cracking...")
return # Exit hashview menu and return to main menu
@@ -2654,6 +2694,9 @@ def main():
hv_download_left.add_argument(
"--hashfile-id", required=True, type=int, help="Hashfile ID"
)
hv_download_left.add_argument(
"--hash-type", default=None, help="Hash type for hashcat (e.g., 1000 for NTLM)"
)
hv_upload_hashfile_job = hashview_subparsers.add_parser(
"upload-hashfile-job",
@@ -2801,6 +2844,7 @@ def main():
download_result = api_harness.download_left_hashes(
args.customer_id,
args.hashfile_id,
hash_type=args.hash_type,
)
print(f"\n✓ Success: Downloaded {download_result['size']} bytes")
print(f" File: {download_result['output_file']}")
@@ -2993,6 +3037,17 @@ def main():
_write_field_sorted_unique(hcatHashFile, f"{hcatHashFile}.nt", 2)
hcatHashFileOrig = hcatHashFile
hcatHashFile = hcatHashFile + ".nt"
elif re.search(r"^.+::.+:.+:[a-f0-9A-F]{64}:", hcatHashFileLine):
# NetNTLMv2 format: username::domain:server_challenge:ntproofstr:blob
# NetNTLMv2-ESS format is similar, with Enhanced Session Security
pwdump_format = False
# Try to detect if it's NetNTLMv2-ESS (has specific markers)
if re.search(r"^.+::.+:.+:[a-f0-9A-F]{16}:[a-f0-9A-F]{32}:[a-f0-9A-F]+$", hcatHashFileLine):
print("NetNTLMv2-ESS format detected")
print("Note: Hash type should be 5600 for NetNTLMv2-ESS hashes")
else:
print("NetNTLMv2 format detected")
print("Note: Hash type should be 5500 for NetNTLMv2 hashes")
else:
print("unknown format....does it have usernames?")
exit(1)
+9 -16
View File
@@ -528,16 +528,12 @@ class TestHashviewAPI:
assert detect_format(hexhash_file) == 5, "hex:hash should default to hash_only"
def test_download_left_with_auto_merge(self, api, tmp_path, monkeypatch):
"""Test that download_left automatically downloads and merges found hashes"""
"""Test that download_left automatically downloads and splits found hashes for hashcat"""
# Use a different CWD than the output directory to ensure merging uses
# output_file's directory (not os.getcwd()).
other_cwd = tmp_path / "other_cwd"
other_cwd.mkdir()
monkeypatch.chdir(other_cwd)
# Create existing .out file
out_file = tmp_path / "left_1_2.txt.out"
out_file.write_text("previously_cracked_hash1:password1\npreviously_cracked_hash2:password2\n")
# Mock left hashes download
mock_left_response = Mock()
@@ -564,26 +560,23 @@ class TestHashviewAPI:
# Set up session.get to return different responses
api.session.get.side_effect = [mock_left_response, mock_found_response]
# Download left hashes (should auto-download and merge found)
# Download left hashes (should auto-download and split found for hashcat)
left_file = tmp_path / "left_1_2.txt"
result = api.download_left_hashes(1, 2, output_file=str(left_file))
# Verify left file was created
assert os.path.exists(result["output_file"])
# Verify found file was downloaded, merged, and deleted
# Verify found file was downloaded and deleted
found_file = tmp_path / "found_1_2.txt"
assert not os.path.exists(found_file), "Found file should be deleted after merge"
assert not os.path.exists(found_file), "Found file should be deleted after split"
assert not (other_cwd / "found_1_2.txt").exists()
# Verify merged content in .out file
if os.path.exists(str(out_file)):
with open(str(out_file), 'r') as f:
merged_content = f.read()
# Should contain both previous and new found hashes
assert "previously_cracked_hash1:password1" in merged_content
assert "found_hash1:found_password1" in merged_content or "found_password1" in merged_content
# 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"
assert not os.path.exists(str(found_clears_file)), "Split clears file should be deleted"
def test_download_left_id_matching(self, api, tmp_path):
"""Test that found hashes only merge when customer_id and hashfile_id match"""
# Create .out file with specific IDs