updated hashview apis and tests.

This commit is contained in:
Justin Bollinger
2026-01-27 16:00:09 -05:00
parent 2dd8ca0029
commit d788f31f84
5 changed files with 295 additions and 100 deletions
+19
View File
@@ -0,0 +1,19 @@
PREFIX ?= /usr/local
BINDIR ?= $(PREFIX)/bin
SCRIPT ?= hate_crack
WRAPPER_PATH := $(BINDIR)/$(SCRIPT)
PROJECT_ROOT := $(abspath $(dir $(lastword $(MAKEFILE_LIST))))
ENTRYPOINT := $(PROJECT_ROOT)/hate_crack/hate_crack.py
UV ?= uv
.PHONY: install clean
install:
@mkdir -p $(BINDIR)
@printf '%s\n' "#!/bin/sh" "" "$(UV) run $(ENTRYPOINT) \"\$$@\"" > $(WRAPPER_PATH)
@chmod +x $(WRAPPER_PATH)
@echo "Installed $(WRAPPER_PATH)"
clean:
@rm -f $(WRAPPER_PATH)
@echo "Removed $(WRAPPER_PATH)"
+139 -99
View File
@@ -1,3 +1,42 @@
class HashviewAPI:
# ...existing code...
def upload_wordlist(self):
"""Interactive method to upload a custom wordlist to Hashview."""
print("\n" + "="*60)
print("WORDLIST UPLOAD")
print("="*60)
print("Select a wordlist file to upload to Hashview.")
print("After upload, you'll need to:")
print(" 1. Create a task in Hashview using this wordlist")
print(" 2. Manually add the task to this job via web interface")
print("\nPress TAB to autocomplete file paths.")
print("="*60)
wordlist_path = select_file_with_autocomplete(
"Enter path to wordlist file"
)
uploaded_wordlist_id = None
wordlist_name = None
if wordlist_path and os.path.isfile(wordlist_path):
# Ask for wordlist name
default_name = os.path.basename(wordlist_path)
wordlist_name = input(f"\nEnter wordlist name (default: {default_name}): ").strip() or default_name
try:
# Upload the wordlist
upload_result = self.upload_wordlist(wordlist_path, wordlist_name)
print(f"\n✓ Success: {upload_result.get('msg', 'Wordlist uploaded')}")
if 'wordlist_id' in upload_result:
uploaded_wordlist_id = upload_result['wordlist_id']
print(f" Wordlist ID: {uploaded_wordlist_id}")
print(f" Wordlist Name: {wordlist_name}")
except Exception as e:
print(f"\n✗ Error uploading wordlist: {str(e)}")
print("Continuing with job creation...")
else:
print("\n✗ No valid wordlist file selected.")
print("Continuing with job creation...")
return uploaded_wordlist_id, wordlist_name
# Methodology provided by Martin Bos (pure_hate) - https://www.trustedsec.com/team/martin-bos/
# Original script created by Larry Spohn (spoonman) - https://www.trustedsec.com/team/larry-spohn/
# Python refactoring and general fixing, Justin Bollinger (bandrel) - https://www.trustedsec.com/team/justin-bollinger/
@@ -1425,118 +1464,118 @@ def hashview_api():
print("\nFull error details:")
traceback.print_exc()
# elif choice == '2':
# # Upload hashfile and create job
# hashfile_path = select_file_with_autocomplete(
# "Enter path to hashfile (TAB to autocomplete)"
# )
# if not hashfile_path or not os.path.exists(hashfile_path):
# print(f"Error: File not found: {hashfile_path}")
# continue
elif choice == '2':
# Upload hashfile and create job
hashfile_path = select_file_with_autocomplete(
"Enter path to hashfile (TAB to autocomplete)"
)
if not hashfile_path or not os.path.exists(hashfile_path):
print(f"Error: File not found: {hashfile_path}")
continue
# customer_id = int(input("Enter customer ID: "))
# hash_type = int(input(f"Enter hash type (default: {hcatHashType}): ") or hcatHashType)
customer_id = int(input("Enter customer ID: "))
hash_type = int(input(f"Enter hash type (default: {hcatHashType}): ") or hcatHashType)
# print("\nFile formats:")
# print(" 0 = pwdump, 1 = NetNTLM, 2 = kerberos")
# print(" 3 = shadow, 4 = user:hash, 5 = hash_only")
# file_format = int(input("Enter file format (default: 5): ") or 5)
print("\nFile formats:")
print(" 0 = pwdump, 1 = NetNTLM, 2 = kerberos")
print(" 3 = shadow, 4 = user:hash, 5 = hash_only")
file_format = int(input("Enter file format (default: 5): ") or 5)
# hashfile_name = input(f"Enter hashfile name (default: {os.path.basename(hashfile_path)}): ") or None
hashfile_name = input(f"Enter hashfile name (default: {os.path.basename(hashfile_path)}): ") or None
# try:
# result = api_harness.upload_hashfile(
# hashfile_path, customer_id, hash_type, file_format, hashfile_name
# )
# print(f"\n✓ Success: {result.get('msg', 'Hashfile uploaded')}")
# if 'hashfile_id' in result:
# print(f" Hashfile ID: {result['hashfile_id']}")
# # Hash count is not returned by the upload API, so we don't display it
# if 'hash_count' in result:
# print(f" Hash count: {result['hash_count']}")
# if 'instacracked' in result:
# print(f" Insta-cracked: {result['instacracked']}")
try:
result = api_harness.upload_hashfile(
hashfile_path, customer_id, hash_type, file_format, hashfile_name
)
print(f"\n✓ Success: {result.get('msg', 'Hashfile uploaded')}")
if 'hashfile_id' in result:
print(f" Hashfile ID: {result['hashfile_id']}")
# Hash count is not returned by the upload API, so we don't display it
if 'hash_count' in result:
print(f" Hash count: {result['hash_count']}")
if 'instacracked' in result:
print(f" Insta-cracked: {result['instacracked']}")
# # Offer to create a job
# create_job = input("\nWould you like to create a job for this hashfile? (Y/n): ") or "Y"
# if create_job.upper() == 'Y':
# job_name = input("Enter job name: ")
# limit_recovered = input("Limit to recovered hashes only? (y/N): ").upper() == 'Y'
# notify_email = input("Send email notifications? (Y/n): ").upper() != 'N'
# Offer to create a job
create_job = input("\nWould you like to create a job for this hashfile? (Y/n): ") or "Y"
if create_job.upper() == 'Y':
job_name = input("Enter job name: ")
limit_recovered = input("Limit to recovered hashes only? (y/N): ").upper() == 'Y'
notify_email = input("Send email notifications? (Y/n): ").upper() != 'N'
# # Ask if user wants to upload a custom wordlist for this job
# upload_wordlist = input("\nUpload a custom wordlist to Hashview? (y/N): ").upper() == 'Y'
# uploaded_wordlist_id = None
# Ask if user wants to upload a custom wordlist for this job
upload_wordlist = input("\nUpload a custom wordlist to Hashview? (y/N): ").upper() == 'Y'
uploaded_wordlist_id = None
# if upload_wordlist:
# print("\n" + "="*60)
# print("WORDLIST UPLOAD")
# print("="*60)
# print("Select a wordlist file to upload to Hashview.")
# print("After upload, you'll need to:")
# print(" 1. Create a task in Hashview using this wordlist")
# print(" 2. Manually add the task to this job via web interface")
# print("\nPress TAB to autocomplete file paths.")
# print("="*60)
if upload_wordlist:
print("\n" + "="*60)
print("WORDLIST UPLOAD")
print("="*60)
print("Select a wordlist file to upload to Hashview.")
print("After upload, you'll need to:")
print(" 1. Create a task in Hashview using this wordlist")
print(" 2. Manually add the task to this job via web interface")
print("\nPress TAB to autocomplete file paths.")
print("="*60)
# wordlist_path = select_file_with_autocomplete(
# "Enter path to wordlist file"
# )
wordlist_path = select_file_with_autocomplete(
"Enter path to wordlist file"
)
# if wordlist_path and os.path.isfile(wordlist_path):
# # Ask for wordlist name
# default_name = os.path.basename(wordlist_path)
# wordlist_name = input(f"\nEnter wordlist name (default: {default_name}): ").strip() or default_name
if wordlist_path and os.path.isfile(wordlist_path):
# Ask for wordlist name
default_name = os.path.basename(wordlist_path)
wordlist_name = input(f"\nEnter wordlist name (default: {default_name}): ").strip() or default_name
# try:
# # Upload the wordlist
# upload_result = api_harness.upload_wordlist(wordlist_path, wordlist_name)
# print(f"\n✓ Success: {upload_result.get('msg', 'Wordlist uploaded')}")
# if 'wordlist_id' in upload_result:
# uploaded_wordlist_id = upload_result['wordlist_id']
# print(f" Wordlist ID: {uploaded_wordlist_id}")
# print(f" Wordlist Name: {wordlist_name}")
# except Exception as e:
# print(f"\n✗ Error uploading wordlist: {str(e)}")
# print("Continuing with job creation...")
# else:
# print("\n✗ No valid wordlist file selected.")
# print("Continuing with job creation...")
try:
# Upload the wordlist
upload_result = api_harness.upload_wordlist(wordlist_path, wordlist_name)
print(f"\n✓ Success: {upload_result.get('msg', 'Wordlist uploaded')}")
if 'wordlist_id' in upload_result:
uploaded_wordlist_id = upload_result['wordlist_id']
print(f" Wordlist ID: {uploaded_wordlist_id}")
print(f" Wordlist Name: {wordlist_name}")
except Exception as e:
print(f"\n✗ Error uploading wordlist: {str(e)}")
print("Continuing with job creation...")
else:
print("\n✗ No valid wordlist file selected.")
print("Continuing with job creation...")
# try:
# job_result = api_harness.create_job(
# job_name, result['hashfile_id'], customer_id,
# limit_recovered, notify_email
# )
# print(f"\n✓ Success: {job_result.get('msg', 'Job created')}")
# if 'job_id' in job_result:
# print(f" Job ID: {job_result['job_id']}")
# print(f"\nNote: Job created with automatically assigned tasks based on")
# print(f" historical effectiveness for hash type {hash_type}.")
try:
job_result = api_harness.create_job(
job_name, result['hashfile_id'], customer_id,
limit_recovered, notify_email
)
print(f"\n✓ Success: {job_result.get('msg', 'Job created')}")
if 'job_id' in job_result:
print(f" Job ID: {job_result['job_id']}")
print(f"\nNote: Job created with automatically assigned tasks based on")
print(f" historical effectiveness for hash type {hash_type}.")
# if uploaded_wordlist_id:
# print(f"\n{'='*60}")
# print("NEXT STEPS - Configure Task in Hashview Web Interface:")
# print(f"{'='*60}")
# print(f"1. Go to: {hashview_url}")
# print(f"2. Navigate to Tasks → Create New Task")
# print(f"3. Configure task with:")
# print(f" - Wordlist ID: {uploaded_wordlist_id} ({wordlist_name})")
# print(f" - Rule: (select appropriate rule)")
# print(f" - Attack mode: 0 (dictionary)")
# print(f"4. Go to Jobs → Job ID {job_result['job_id']}")
# print(f"5. Add the new task to this job")
# print(f"{'='*60}")
if uploaded_wordlist_id:
print(f"\n{'='*60}")
print("NEXT STEPS - Configure Task in Hashview Web Interface:")
print(f"{'='*60}")
print(f"1. Go to: {hashview_url}")
print(f"2. Navigate to Tasks → Create New Task")
print(f"3. Configure task with:")
print(f" - Wordlist ID: {uploaded_wordlist_id} ({wordlist_name})")
print(f" - Rule: (select appropriate rule)")
print(f" - Attack mode: 0 (dictionary)")
print(f"4. Go to Jobs → Job ID {job_result['job_id']}")
print(f"5. Add the new task to this job")
print(f"{'='*60}")
# # Offer to start the job
# start_now = input("\nStart the job now? (Y/n): ") or "Y"
# if start_now.upper() == 'Y':
# start_result = api_harness.start_job(job_result['job_id'])
# print(f"\n✓ Success: {start_result.get('msg', 'Job started')}")
# except Exception as e:
# print(f"\n✗ Error creating job: {str(e)}")
# except Exception as e:
# print(f"\n✗ Error uploading hashfile: {str(e)}")
# Offer to start the job
start_now = input("\nStart the job now? (Y/n): ") or "Y"
if start_now.upper() == 'Y':
start_result = api_harness.start_job(job_result['job_id'])
print(f"\n✓ Success: {start_result.get('msg', 'Job started')}")
except Exception as e:
print(f"\n✗ Error creating job: {str(e)}")
except Exception as e:
print(f"\n✗ Error uploading hashfile: {str(e)}")
elif choice == '3':
# List customers
@@ -1612,6 +1651,7 @@ def hashview_api():
switch = input("\nSwitch to this hashfile for cracking? (Y/n): ").strip().lower()
if switch != 'n':
hcatHashFile = download_result['output_file']
print(f"✓ Switched to hashfile: {hcatHashFile}")
print("\nReturning to main menu to start cracking...")
return # Exit hashview menu and return to main menu
+34 -1
View File
@@ -410,6 +410,39 @@ def weakpass_wordlist_menu(rank=-1):
# Hashview Integration - Real API implementation matching hate_crack.py
class HashviewAPI:
def upload_wordlist_file(self, wordlist_path, wordlist_name=None):
"""Directly upload a wordlist file to Hashview (non-interactive)."""
if wordlist_name is None:
wordlist_name = os.path.basename(wordlist_path)
with open(wordlist_path, 'rb') as f:
file_content = f.read()
url = f"{self.base_url}/v1/wordlists/add/{wordlist_name}"
headers = {'Content-Type': 'text/plain'}
resp = self.session.post(url, data=file_content, headers=headers)
resp.raise_for_status()
return resp.json()
def list_wordlists(self):
"""List available wordlists from Hashview API."""
endpoint = f"{self.base_url}/v1/wordlists"
response = self.session.get(endpoint, headers=self._auth_headers())
response.raise_for_status()
try:
data = response.json()
except Exception:
raise Exception(f"Invalid API response: {response.text}")
# The API may return a list or a dict with a key
if isinstance(data, dict) and 'wordlists' in data:
wordlists = data['wordlists']
# If wordlists is a JSON string, decode it
if isinstance(wordlists, str):
import json
wordlists = json.loads(wordlists)
return wordlists
elif isinstance(data, list):
return data
else:
return []
def __init__(self, base_url, api_key, debug=False):
self.base_url = base_url.rstrip('/')
self.api_key = api_key
@@ -639,7 +672,7 @@ class HashviewAPI:
def download_left_hashes(self, customer_id, hashfile_id, output_file=None):
import sys
url = f"{self.base_url}/v1/hashfiles/{hashfile_id}"
url = f"{self.base_url}/v1/hashfiles/{hashfile_id}/left"
resp = self.session.get(url, stream=True)
resp.raise_for_status()
if output_file is None:
+30
View File
@@ -0,0 +1,30 @@
import os
import json
import pytest
from hate_crack.api import HashviewAPI
def get_hashview_config():
config_path = os.path.join(os.path.dirname(__file__), '..', 'config.json')
with open(config_path, 'r') as f:
config = json.load(f)
hashview_url = config.get('hashview_url')
hashview_api_key = config.get('hashview_api_key')
return hashview_url, hashview_api_key
@pytest.mark.skipif(
not get_hashview_config()[0] or not get_hashview_config()[1],
reason="Requires hashview_url and hashview_api_key in config.json."
)
def test_upload_cracked_hashes_from_file():
hashview_url, hashview_api_key = get_hashview_config()
api = HashviewAPI(hashview_url, hashview_api_key)
file_path = os.path.join(os.path.dirname(__file__), '..', '1.out')
if not os.path.isfile(file_path):
pytest.skip("1.out not found in repo root.")
result = api.upload_cracked_hashes(file_path, hash_type='1000')
assert result is not None
assert result.get('type') != 'Error'
+73
View File
@@ -0,0 +1,73 @@
import os
import json
import tempfile
import pytest
from hate_crack.api import HashviewAPI
def get_hashview_config():
config_path = os.path.join(os.path.dirname(__file__), '..', 'config.json')
with open(config_path, 'r') as f:
config = json.load(f)
hashview_url = config.get('hashview_url')
hashview_api_key = config.get('hashview_api_key')
return hashview_url, hashview_api_key
def test_upload_wordlist_api_mocked(monkeypatch):
"""Test direct API upload of a wordlist file using a mocked API call."""
hashview_url, hashview_api_key = get_hashview_config()
api = HashviewAPI(hashview_url or "http://example.com", hashview_api_key or "dummy")
# Create a temp wordlist file
with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
f.write('password1\npassword2\n')
wordlist_path = f.name
wordlist_name = os.path.basename(wordlist_path)
class DummyResponse:
def raise_for_status(self):
return None
def json(self):
return {"status": 200, "type": "message", "msg": "Wordlist added", "wordlist_id": 123}
def fake_post(url, data=None, headers=None):
assert url.endswith(f"/v1/wordlists/add/{wordlist_name}")
assert headers and headers.get("Content-Type") == "text/plain"
assert data is not None
return DummyResponse()
monkeypatch.setattr(api.session, "post", fake_post)
upload_result = api.upload_wordlist_file(wordlist_path, wordlist_name)
assert upload_result is not None
assert 'wordlist_id' in upload_result
msg = upload_result.get('msg', '').lower()
assert 'uploaded' in msg or 'added' in msg
os.remove(wordlist_path)
@pytest.mark.skipif(
os.environ.get("HATE_CRACK_RUN_LIVE_HASHVIEW_TESTS") != "1",
reason="Live Hashview test disabled. Set HATE_CRACK_RUN_LIVE_HASHVIEW_TESTS=1 to run.",
)
def test_upload_wordlist_api_live():
"""Live API upload test; only runs when explicitly enabled."""
hashview_url, hashview_api_key = get_hashview_config()
if not hashview_url or not hashview_api_key:
pytest.skip("Requires hashview_url and hashview_api_key in config.json.")
api = HashviewAPI(hashview_url, hashview_api_key)
with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
f.write('password1\npassword2\n')
wordlist_path = f.name
wordlist_name = os.path.basename(wordlist_path)
try:
upload_result = api.upload_wordlist_file(wordlist_path, wordlist_name)
assert upload_result is not None
assert 'wordlist_id' in upload_result
msg = upload_result.get('msg', '').lower()
assert 'uploaded' in msg or 'added' in msg
finally:
os.remove(wordlist_path)