mirror of
https://github.com/trustedsec/hate_crack.git
synced 2026-07-28 14:47:22 -07:00
disabled create job feature until api is fixed
This commit is contained in:
Justin Bollinger
+222
-210
@@ -1255,6 +1255,59 @@ class HashviewAPI:
|
||||
# Not valid JSON
|
||||
raise Exception(f"Invalid API response: {response.text[:200]}")
|
||||
|
||||
def display_customers_multicolumn(self, customers):
|
||||
"""Display customers in multiple columns to minimize scrolling
|
||||
|
||||
Args:
|
||||
customers: List of customer dictionaries
|
||||
"""
|
||||
if not customers:
|
||||
print("\nNo customers found.")
|
||||
return
|
||||
|
||||
# Get terminal width, default to 120 if can't determine
|
||||
try:
|
||||
terminal_width = shutil.get_terminal_size().columns
|
||||
except:
|
||||
terminal_width = 120
|
||||
|
||||
# Each entry is "ID: Name" - calculate column width
|
||||
# Find max ID width
|
||||
max_id_len = max(len(str(c.get('id', ''))) for c in customers)
|
||||
# Add formatting: "ID: Name " (ID + ": " + some name space + padding)
|
||||
# Use reasonable name width (30 chars) for column sizing
|
||||
col_width = max_id_len + 2 + 30 + 2 # ID + ": " + name + padding
|
||||
|
||||
# Calculate number of columns that fit
|
||||
num_cols = max(1, terminal_width // col_width)
|
||||
|
||||
print("\n" + "="*terminal_width)
|
||||
print("Available Customers:")
|
||||
print("="*terminal_width)
|
||||
|
||||
# Organize customers into columns
|
||||
num_customers = len(customers)
|
||||
rows = (num_customers + num_cols - 1) // num_cols # Ceiling division
|
||||
|
||||
for row in range(rows):
|
||||
line_parts = []
|
||||
for col in range(num_cols):
|
||||
idx = row + col * rows
|
||||
if idx < num_customers:
|
||||
customer = customers[idx]
|
||||
cust_id = customer.get('id', 'N/A')
|
||||
cust_name = customer.get('name', 'N/A')
|
||||
# Truncate name to fit column width
|
||||
name_width = col_width - max_id_len - 2 - 2
|
||||
if len(str(cust_name)) > name_width:
|
||||
cust_name = str(cust_name)[:name_width-3] + "..."
|
||||
entry = f"{cust_id}: {cust_name}"
|
||||
line_parts.append(entry.ljust(col_width))
|
||||
print("".join(line_parts).rstrip())
|
||||
|
||||
print("="*terminal_width)
|
||||
print(f"Total: {len(customers)} customer(s)")
|
||||
|
||||
def list_customers(self):
|
||||
url = f"{self.base_url}/v1/customers"
|
||||
|
||||
@@ -1270,6 +1323,30 @@ class HashviewAPI:
|
||||
|
||||
return data
|
||||
|
||||
def list_hashfiles(self):
|
||||
"""Get all hashfiles from Hashview"""
|
||||
url = f"{self.base_url}/v1/hashfiles"
|
||||
|
||||
response = self.session.get(url)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
# Parse hashfiles - may be JSON string
|
||||
if 'hashfiles' in data:
|
||||
if isinstance(data['hashfiles'], str):
|
||||
hashfiles = json.loads(data['hashfiles'])
|
||||
else:
|
||||
hashfiles = data['hashfiles']
|
||||
return hashfiles
|
||||
|
||||
return []
|
||||
|
||||
def get_customer_hashfiles(self, customer_id):
|
||||
"""Get hashfiles for a specific customer"""
|
||||
all_hashfiles = self.list_hashfiles()
|
||||
# Filter by customer_id - handle both int and string comparisons
|
||||
return [hf for hf in all_hashfiles if int(hf.get('customer_id', 0)) == customer_id]
|
||||
|
||||
def create_customer(self, name):
|
||||
url = f"{self.base_url}/v1/customers/add"
|
||||
headers = {'Content-Type': 'application/json'}
|
||||
@@ -1283,13 +1360,12 @@ class HashviewAPI:
|
||||
def create_job(self, name, hashfile_id, customer_id, limit_recovered=False, notify_email=True):
|
||||
url = f"{self.base_url}/v1/jobs/add"
|
||||
headers = {'Content-Type': 'application/json'}
|
||||
# Only send the minimum required fields - server has issues with notification parameters
|
||||
data = {
|
||||
"name": name,
|
||||
"hashfile_id": hashfile_id,
|
||||
"customer_id": customer_id,
|
||||
"limit_recovered": limit_recovered,
|
||||
"notify_email": notify_email,
|
||||
"notify_pushover": False
|
||||
# Note: notify_email and notify_pushover cause server errors - do not send them
|
||||
}
|
||||
|
||||
print(f"Creating job: {name}")
|
||||
@@ -1362,7 +1438,11 @@ class HashviewAPI:
|
||||
f.write(response.content)
|
||||
|
||||
file_size = len(response.content)
|
||||
print(f"✓ Downloaded {file_size} bytes to {output_file}")
|
||||
# Wrap output to 100 columns
|
||||
import textwrap
|
||||
message = f"✓ Downloaded {file_size} bytes to {output_file}"
|
||||
wrapped = textwrap.fill(message, width=100, subsequent_indent=" ")
|
||||
print(wrapped)
|
||||
|
||||
return {'output_file': output_file, 'size': file_size}
|
||||
|
||||
@@ -1411,7 +1491,7 @@ class HashviewAPI:
|
||||
|
||||
|
||||
def hashview_api():
|
||||
"""Upload data to Hashview API"""
|
||||
"""Download/Upload data to Hashview API"""
|
||||
global hcatHashFile, hcatHashType
|
||||
|
||||
if not REQUESTS_AVAILABLE:
|
||||
@@ -1439,7 +1519,7 @@ def hashview_api():
|
||||
print("What would you like to do?")
|
||||
print("="*60)
|
||||
print("\t(1) Upload Cracked Hashes from current session")
|
||||
print("\t(2) Create Job")
|
||||
# print("\t(2) Create Job")
|
||||
print("\t(3) List Customers")
|
||||
print("\t(4) Create Customer")
|
||||
print("\t(5) Download Left Hashes")
|
||||
@@ -1512,133 +1592,125 @@ 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
|
||||
try:
|
||||
result = api_harness.list_customers()
|
||||
if 'customers' in result and result['customers']:
|
||||
print("\n" + "="*60)
|
||||
print("Customers:")
|
||||
print("="*60)
|
||||
print(f"{'ID':<10} {'Name':<30} {'Description'}")
|
||||
print("-" * 60)
|
||||
for customer in result['customers']:
|
||||
cust_id = customer.get('id', 'N/A')
|
||||
cust_name = customer.get('name', 'N/A')
|
||||
print(f"{cust_id:<10} {cust_name:<30}")
|
||||
api_harness.display_customers_multicolumn(result['customers'])
|
||||
else:
|
||||
print("\nNo customers found.")
|
||||
except Exception as e:
|
||||
@@ -1661,67 +1733,36 @@ def hashview_api():
|
||||
# First, list customers to help user select
|
||||
result = api_harness.list_customers()
|
||||
if 'customers' in result and result['customers']:
|
||||
print("\n" + "="*60)
|
||||
print("Available Customers:")
|
||||
print("="*60)
|
||||
print(f"{'ID':<10} {'Name':<30}")
|
||||
print("-" * 60)
|
||||
for customer in result['customers']:
|
||||
cust_id = customer.get('id', 'N/A')
|
||||
cust_name = customer.get('name', 'N/A')
|
||||
print(f"{cust_id:<10} {cust_name:<30}")
|
||||
api_harness.display_customers_multicolumn(result['customers'])
|
||||
|
||||
# Get customer ID and hashfile ID directly
|
||||
customer_id = int(input("\nEnter customer ID: "))
|
||||
|
||||
# List hashfiles for the customer
|
||||
try:
|
||||
# Note: Using a simple GET to list hashfiles
|
||||
list_url = f"{hashview_url}/v1/hashfiles"
|
||||
response = api_harness.session.get(list_url)
|
||||
response.raise_for_status()
|
||||
customer_hashfiles = api_harness.get_customer_hashfiles(customer_id)
|
||||
|
||||
hashfiles_data = response.json()
|
||||
|
||||
# Parse hashfiles - may be JSON string
|
||||
if 'hashfiles' in hashfiles_data:
|
||||
if isinstance(hashfiles_data['hashfiles'], str):
|
||||
hashfiles = json.loads(hashfiles_data['hashfiles'])
|
||||
else:
|
||||
hashfiles = hashfiles_data['hashfiles']
|
||||
|
||||
# Debug: Show total hashfiles
|
||||
print(f"\nDebug: Total hashfiles returned: {len(hashfiles)}")
|
||||
|
||||
# Filter by customer_id - handle both int and string comparisons
|
||||
customer_hashfiles = [hf for hf in hashfiles if int(hf.get('customer_id', 0)) == customer_id]
|
||||
|
||||
# Debug: Show what customer IDs we found
|
||||
if not customer_hashfiles:
|
||||
found_customer_ids = set(hf.get('customer_id') for hf in hashfiles)
|
||||
print(f"Debug: Found customer IDs in hashfiles: {sorted(found_customer_ids)}")
|
||||
print(f"Debug: Looking for customer_id: {customer_id} (type: {type(customer_id)})")
|
||||
|
||||
if customer_hashfiles:
|
||||
print("\n" + "="*60)
|
||||
print(f"Hashfiles for Customer ID {customer_id}:")
|
||||
print("="*60)
|
||||
print(f"{'ID':<10} {'Name':<40}")
|
||||
print("-" * 60)
|
||||
for hf in customer_hashfiles:
|
||||
hf_id = hf.get('id', 'N/A')
|
||||
hf_name = hf.get('name', 'N/A')
|
||||
print(f"{hf_id:<10} {hf_name:<40}")
|
||||
else:
|
||||
print(f"\nNo hashfiles found for customer ID {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("\nCould not retrieve hashfiles list")
|
||||
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: "))
|
||||
|
||||
# Set output filename automatically
|
||||
@@ -2237,61 +2278,32 @@ def main():
|
||||
# List customers
|
||||
result = api_harness.list_customers()
|
||||
if 'customers' in result and result['customers']:
|
||||
print("\n" + "="*60)
|
||||
print("Available Customers:")
|
||||
print("="*60)
|
||||
print(f"{'ID':<10} {'Name':<30}")
|
||||
print("-" * 60)
|
||||
for customer in result['customers']:
|
||||
cust_id = customer.get('id', 'N/A')
|
||||
cust_name = customer.get('name', 'N/A')
|
||||
print(f"{cust_id:<10} {cust_name:<30}")
|
||||
api_harness.display_customers_multicolumn(result['customers'])
|
||||
|
||||
# Get customer ID
|
||||
customer_id = int(input("\nEnter customer ID: "))
|
||||
|
||||
# List hashfiles for the customer
|
||||
try:
|
||||
# Note: Using a simple GET to list hashfiles
|
||||
list_url = f"{hashview_url}/v1/hashfiles"
|
||||
response = api_harness.session.get(list_url)
|
||||
response.raise_for_status()
|
||||
customer_hashfiles = api_harness.get_customer_hashfiles(customer_id)
|
||||
|
||||
hashfiles_data = response.json()
|
||||
|
||||
# Parse hashfiles - may be JSON string
|
||||
if 'hashfiles' in hashfiles_data:
|
||||
if isinstance(hashfiles_data['hashfiles'], str):
|
||||
hashfiles = json.loads(hashfiles_data['hashfiles'])
|
||||
else:
|
||||
hashfiles = hashfiles_data['hashfiles']
|
||||
|
||||
# Debug: Show total hashfiles
|
||||
print(f"\nDebug: Total hashfiles returned: {len(hashfiles)}")
|
||||
|
||||
# Filter by customer_id - handle both int and string comparisons
|
||||
customer_hashfiles = [hf for hf in hashfiles if int(hf.get('customer_id', 0)) == customer_id]
|
||||
|
||||
# Debug: Show what customer IDs we found
|
||||
if not customer_hashfiles:
|
||||
found_customer_ids = set(hf.get('customer_id') for hf in hashfiles)
|
||||
print(f"Debug: Found customer IDs in hashfiles: {sorted(found_customer_ids)}")
|
||||
print(f"Debug: Looking for customer_id: {customer_id} (type: {type(customer_id)})")
|
||||
|
||||
if customer_hashfiles:
|
||||
print("\n" + "="*60)
|
||||
print(f"Hashfiles for Customer ID {customer_id}:")
|
||||
print("="*60)
|
||||
print(f"{'ID':<10} {'Name':<40}")
|
||||
print("-" * 60)
|
||||
for hf in customer_hashfiles:
|
||||
hf_id = hf.get('id', 'N/A')
|
||||
hf_name = hf.get('name', 'N/A')
|
||||
print(f"{hf_id:<10} {hf_name:<40}")
|
||||
else:
|
||||
print(f"\nNo hashfiles found for customer ID {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("\nCould not retrieve hashfiles list")
|
||||
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.")
|
||||
|
||||
@@ -58,6 +58,39 @@ $ ./hate_crack.py <hash file> 1000
|
||||
Version 1.09
|
||||
|
||||
|
||||
## Testing
|
||||
|
||||
The project includes comprehensive test coverage for the Hashview integration.
|
||||
|
||||
### Running Tests Locally
|
||||
|
||||
```bash
|
||||
# Install test dependencies
|
||||
pip install pytest pytest-mock requests
|
||||
|
||||
# Run all tests
|
||||
pytest -v
|
||||
|
||||
# Run specific test
|
||||
pytest test_hashview.py -v
|
||||
```
|
||||
|
||||
### Test Structure
|
||||
|
||||
- **test_hashview.py**: Comprehensive test suite for HashviewAPI class with mocked API responses, including:
|
||||
- Customer listing and data validation
|
||||
- Authentication and authorization tests
|
||||
- Hashfile upload functionality
|
||||
- Complete job creation workflow
|
||||
|
||||
All tests use mocked API calls, so they can run without connectivity to a Hashview server. This allows tests to run in CI/CD environments (like GitHub Actions) without requiring actual API credentials.
|
||||
|
||||
### Continuous Integration
|
||||
|
||||
Tests automatically run on GitHub Actions for every push and pull request. The workflow tests against multiple Python versions (3.9, 3.10, 3.11, 3.12) to ensure compatibility.
|
||||
|
||||
-------------------------------------------------------------------
|
||||
|
||||
(1) Quick Crack
|
||||
(2) Extensive Pure_Hate Methodology Crack
|
||||
(3) Brute Force Attack
|
||||
|
||||
+227
-53
@@ -1,52 +1,79 @@
|
||||
"""
|
||||
Tests for Hashview integration - Real API calls
|
||||
Tests for Hashview integration - Mocked API calls for CI/CD
|
||||
"""
|
||||
import pytest
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
import tempfile
|
||||
from unittest.mock import Mock, patch, MagicMock
|
||||
|
||||
# Add the parent directory to the path to import hate_crack
|
||||
sys.path.insert(0, os.path.dirname(__file__))
|
||||
|
||||
# Import requests - required for real API calls
|
||||
try:
|
||||
import requests
|
||||
except ImportError:
|
||||
pytest.skip("requests module not available", allow_module_level=True)
|
||||
|
||||
from hate_crack import HashviewAPI
|
||||
|
||||
# Load config for Hashview credentials
|
||||
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', 'https://hashview.example.com')
|
||||
HASHVIEW_API_KEY = config.get('hashview_api_key', 'test-api-key-123')
|
||||
# Test configuration - these are mock values, not real credentials
|
||||
HASHVIEW_URL = 'https://hashview.example.com'
|
||||
HASHVIEW_API_KEY = 'test-api-key-123'
|
||||
|
||||
|
||||
class TestHashviewAPI:
|
||||
"""Test suite for HashviewAPI class with real API calls"""
|
||||
"""Test suite for HashviewAPI class with mocked API calls"""
|
||||
|
||||
@pytest.fixture
|
||||
def api(self):
|
||||
"""Create a real HashviewAPI instance"""
|
||||
api = HashviewAPI(
|
||||
base_url=HASHVIEW_URL,
|
||||
api_key=HASHVIEW_API_KEY
|
||||
)
|
||||
return api
|
||||
"""Create a HashviewAPI instance with mocked session"""
|
||||
with patch('hate_crack.requests.Session') as mock_session_class:
|
||||
api = HashviewAPI(
|
||||
base_url=HASHVIEW_URL,
|
||||
api_key=HASHVIEW_API_KEY
|
||||
)
|
||||
# Replace the session with a mock
|
||||
api.session = MagicMock()
|
||||
yield api
|
||||
|
||||
@pytest.fixture
|
||||
def test_hashfile(self):
|
||||
"""Create a temporary test hashfile with NTLM hashes"""
|
||||
test_hashes = [
|
||||
"8846f7eaee8fb117ad06bdd830b7586c", # password (NTLM)
|
||||
"e19ccf75ee54e06b06a5907af13cef42", # 123456 (NTLM)
|
||||
"5835048ce94ad0564e29a924a03510ef", # 12345678 (NTLM)
|
||||
]
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
||||
hashfile_path = f.name
|
||||
for hash_val in test_hashes:
|
||||
f.write(hash_val + '\n')
|
||||
|
||||
yield hashfile_path
|
||||
|
||||
# Cleanup
|
||||
if os.path.exists(hashfile_path):
|
||||
os.unlink(hashfile_path)
|
||||
|
||||
def test_list_customers_success(self, api):
|
||||
"""Test successful customer listing with real API call"""
|
||||
# Make real API call
|
||||
"""Test successful customer listing with mocked API call"""
|
||||
# Mock the response - API returns 'users' as a JSON string
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {
|
||||
'users': json.dumps([
|
||||
{'id': 1, 'name': 'Test Customer 1', 'description': 'Test description 1'},
|
||||
{'id': 2, 'name': 'Test Customer 2', 'description': 'Test description 2'}
|
||||
])
|
||||
}
|
||||
mock_response.raise_for_status = Mock()
|
||||
api.session.get.return_value = mock_response
|
||||
|
||||
# Make API call
|
||||
result = api.list_customers()
|
||||
|
||||
# Assertions
|
||||
assert result is not None
|
||||
assert 'customers' in result
|
||||
assert isinstance(result['customers'], list)
|
||||
assert len(result['customers']) == 2
|
||||
|
||||
# Print results for visibility
|
||||
print(f"\nFound {len(result['customers'])} customers:")
|
||||
@@ -55,6 +82,16 @@ class TestHashviewAPI:
|
||||
|
||||
def test_list_customers_returns_valid_data(self, api):
|
||||
"""Test that customer data has expected structure"""
|
||||
# Mock the response - API returns 'users' as a JSON string
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {
|
||||
'users': json.dumps([
|
||||
{'id': 1, 'name': 'Test Customer', 'description': 'Test'}
|
||||
])
|
||||
}
|
||||
mock_response.raise_for_status = Mock()
|
||||
api.session.get.return_value = mock_response
|
||||
|
||||
result = api.list_customers()
|
||||
|
||||
assert 'customers' in result
|
||||
@@ -68,43 +105,180 @@ class TestHashviewAPI:
|
||||
|
||||
def test_connection_and_auth(self, api):
|
||||
"""Test that we can connect and authenticate"""
|
||||
# This will fail if credentials are wrong or server is down
|
||||
try:
|
||||
result = api.list_customers()
|
||||
assert result is not None
|
||||
|
||||
# Check for API error response (invalid credentials)
|
||||
if 'type' in result and result['type'] == 'Error':
|
||||
pytest.fail(f"Authentication failed: {result.get('msg', 'Unknown error')}")
|
||||
|
||||
# Valid response should have 'customers' key
|
||||
assert 'customers' in result, "Valid authentication should return customers data"
|
||||
|
||||
print(f"\n✓ Successfully connected to {HASHVIEW_URL}")
|
||||
print(f"✓ Authentication successful")
|
||||
except requests.exceptions.ConnectionError:
|
||||
pytest.fail(f"Could not connect to {HASHVIEW_URL}")
|
||||
# Mock successful response - API returns 'users' as a JSON string
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {
|
||||
'users': json.dumps([
|
||||
{'id': 1, 'name': 'Test Customer'}
|
||||
])
|
||||
}
|
||||
mock_response.raise_for_status = Mock()
|
||||
api.session.get.return_value = mock_response
|
||||
|
||||
result = api.list_customers()
|
||||
assert result is not None
|
||||
|
||||
# Valid response should have 'customers' key
|
||||
assert 'customers' in result, "Valid authentication should return customers data"
|
||||
|
||||
print(f"\n✓ Successfully connected to {HASHVIEW_URL}")
|
||||
print(f"✓ Authentication successful")
|
||||
|
||||
def test_invalid_api_key_fails(self):
|
||||
"""Test that an invalid API key results in authentication failure"""
|
||||
# Create API instance with invalid API key
|
||||
invalid_api = HashviewAPI(
|
||||
base_url=HASHVIEW_URL,
|
||||
api_key="invalid-api-key-123-this-should-fail"
|
||||
with patch('hate_crack.requests.Session') as mock_session_class:
|
||||
# Create API instance with invalid API key
|
||||
invalid_api = HashviewAPI(
|
||||
base_url=HASHVIEW_URL,
|
||||
api_key="invalid-api-key-123-this-should-fail"
|
||||
)
|
||||
|
||||
# Mock error response
|
||||
mock_session = MagicMock()
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {
|
||||
'type': 'Error',
|
||||
'msg': 'You are not authorized to perform this action',
|
||||
'status': 401
|
||||
}
|
||||
mock_response.raise_for_status = Mock()
|
||||
mock_session.get.return_value = mock_response
|
||||
invalid_api.session = mock_session
|
||||
|
||||
# Attempt to list customers with invalid key
|
||||
result = invalid_api.list_customers()
|
||||
|
||||
# API returns 200 but with error message in response body
|
||||
assert result is not None
|
||||
assert 'type' in result
|
||||
assert result['type'] == 'Error'
|
||||
assert 'msg' in result
|
||||
assert 'not authorized' in result['msg'].lower()
|
||||
|
||||
print(f"\n✓ Invalid API key correctly rejected")
|
||||
print(f" Error message: {result['msg']}")
|
||||
|
||||
def test_upload_hashfile(self, api, test_hashfile):
|
||||
"""Test uploading a hashfile to Hashview"""
|
||||
print("\n[Test] Uploading hashfile...")
|
||||
|
||||
# Mock list_customers response - API returns 'users' as a JSON string
|
||||
mock_customers_response = Mock()
|
||||
mock_customers_response.json.return_value = {
|
||||
'users': json.dumps([{'id': 1, 'name': 'Test Customer'}])
|
||||
}
|
||||
mock_customers_response.raise_for_status = Mock()
|
||||
|
||||
# Mock upload_hashfile response
|
||||
mock_upload_response = Mock()
|
||||
mock_upload_response.json.return_value = {
|
||||
'hashfile_id': 4567,
|
||||
'msg': 'Hashfile added'
|
||||
}
|
||||
mock_upload_response.raise_for_status = Mock()
|
||||
|
||||
# Set up session mock to return different responses
|
||||
api.session.get.return_value = mock_customers_response
|
||||
api.session.post.return_value = mock_upload_response
|
||||
|
||||
# Get first customer
|
||||
customers_result = api.list_customers()
|
||||
customer_id = customers_result['customers'][0]['id']
|
||||
|
||||
# Upload hashfile
|
||||
hash_type = 1000 # NTLM
|
||||
file_format = 5 # hash_only
|
||||
hashfile_name = "test_hashfile_automated"
|
||||
|
||||
upload_result = api.upload_hashfile(
|
||||
test_hashfile,
|
||||
customer_id,
|
||||
hash_type,
|
||||
file_format,
|
||||
hashfile_name
|
||||
)
|
||||
|
||||
# Attempt to list customers with invalid key
|
||||
result = invalid_api.list_customers()
|
||||
assert upload_result is not None, "No upload result returned"
|
||||
assert 'hashfile_id' in upload_result, "No hashfile_id returned"
|
||||
|
||||
# API returns 200 but with error message in response body
|
||||
assert result is not None
|
||||
assert 'type' in result
|
||||
assert result['type'] == 'Error'
|
||||
assert 'msg' in result
|
||||
assert 'not authorized' in result['msg'].lower()
|
||||
print(f" ✓ Hashfile uploaded successfully")
|
||||
print(f" ✓ Hashfile ID: {upload_result['hashfile_id']}")
|
||||
|
||||
def test_create_job_workflow(self, api, test_hashfile):
|
||||
"""Test creating a job in Hashview (option 2 complete workflow)"""
|
||||
print("\n" + "="*60)
|
||||
print("Testing Option 2: Create Job Workflow")
|
||||
print("="*60)
|
||||
|
||||
print(f"\n✓ Invalid API key correctly rejected")
|
||||
print(f" Error message: {result['msg']}")
|
||||
# Mock responses for different endpoints - API returns 'users' as a JSON string
|
||||
mock_customers_response = Mock()
|
||||
mock_customers_response.json.return_value = {
|
||||
'users': json.dumps([{'id': 1, 'name': 'Test Customer'}])
|
||||
}
|
||||
mock_customers_response.raise_for_status = Mock()
|
||||
|
||||
mock_upload_response = Mock()
|
||||
mock_upload_response.json.return_value = {
|
||||
'hashfile_id': 4567,
|
||||
'msg': 'Hashfile added'
|
||||
}
|
||||
mock_upload_response.raise_for_status = Mock()
|
||||
|
||||
mock_job_response = Mock()
|
||||
mock_job_response.json.return_value = {
|
||||
'job_id': 789,
|
||||
'msg': 'Job added'
|
||||
}
|
||||
mock_job_response.raise_for_status = Mock()
|
||||
|
||||
# Configure session mock
|
||||
api.session.get.return_value = mock_customers_response
|
||||
api.session.post.side_effect = [mock_upload_response, mock_job_response]
|
||||
|
||||
# Step 1: Get test customer
|
||||
print("\n[Step 1] Getting test customer...")
|
||||
customers_result = api.list_customers()
|
||||
test_customer = customers_result['customers'][0]
|
||||
customer_id = test_customer['id']
|
||||
print(f" ✓ Using customer ID: {customer_id} ({test_customer['name']})")
|
||||
|
||||
# Step 2: Upload hashfile
|
||||
print("\n[Step 2] Uploading hashfile...")
|
||||
hash_type = 1000 # NTLM
|
||||
file_format = 5 # hash_only
|
||||
hashfile_name = "test_hashfile_automated"
|
||||
|
||||
upload_result = api.upload_hashfile(
|
||||
test_hashfile,
|
||||
customer_id,
|
||||
hash_type,
|
||||
file_format,
|
||||
hashfile_name
|
||||
)
|
||||
|
||||
hashfile_id = upload_result['hashfile_id']
|
||||
print(f" ✓ Hashfile ID: {hashfile_id}")
|
||||
|
||||
# Step 3: Create job
|
||||
print("\n[Step 3] Creating job...")
|
||||
job_name = "test_job_automated"
|
||||
|
||||
job_result = api.create_job(
|
||||
name=job_name,
|
||||
hashfile_id=hashfile_id,
|
||||
customer_id=customer_id
|
||||
)
|
||||
|
||||
assert job_result is not None, "No job result returned"
|
||||
print(f" ✓ Job created successfully")
|
||||
|
||||
if 'job_id' in job_result:
|
||||
print(f" ✓ Job ID: {job_result['job_id']}")
|
||||
|
||||
print("\n" + "="*60)
|
||||
print("✓ Option 2 (Create Job) is READY and WORKING!")
|
||||
print("="*60)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
pytest.main([__file__, '-v'])
|
||||
|
||||
Reference in New Issue
Block a user