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

Update hashview API flow and tests

Browse Source
This commit is contained in:
Justin Bollinger
2026-02-04 15:48:19 -05:00
parent 6f9f4879e1
commit bbaa40958a
4 changed files with 204 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hate_crack/api.py
+28 -16
View File
@@ -117,15 +117,7 @@ def register_torrent_cleanup():
atexit.register(cleanup_torrent_files)
_TORRENT_CLEANUP_REGISTERED = True
def fetch_all_weakpass_wordlists_multithreaded(total_pages=67, threads=10, output_file="weakpass_wordlists.json"):
if os.path.isfile(output_file):
try:
mtime = os.path.getmtime(output_file)
if (time.time() - mtime) < 24 * 60 * 60:
print(f"[i] Using cached wordlist file: {output_file}")
return
except Exception:
pass
def fetch_all_weakpass_wordlists_multithreaded(total_pages=67, threads=10):
wordlists = []
lock = threading.Lock()
q = Queue()
@@ -188,9 +180,7 @@ def fetch_all_weakpass_wordlists_multithreaded(total_pages=67, threads=10, outpu
unique_wordlists.append(wl)
seen.add(wl['name'])
with open(output_file, "w", encoding="utf-8") as f:
json.dump(unique_wordlists, f, indent=2)
print(f"Saved {len(unique_wordlists)} wordlists to {output_file}")
return unique_wordlists
def download_torrent_file(torrent_url, save_dir=None, wordlist_id=None):
register_torrent_cleanup()
@@ -348,12 +338,10 @@ def download_torrent_file(torrent_url, save_dir=None, wordlist_id=None):
return local_filename
def weakpass_wordlist_menu(rank=-1):
fetch_all_weakpass_wordlists_multithreaded()
try:
with open("weakpass_wordlists.json", "r", encoding="utf-8") as f:
all_wordlists = json.load(f)
all_wordlists = fetch_all_weakpass_wordlists_multithreaded()
except Exception as e:
print(f"Failed to load local wordlist cache: {e}")
print(f"Failed to fetch wordlists: {e}")
return
if rank == 0:
filtered_wordlists = all_wordlists
@@ -708,6 +696,30 @@ class HashviewAPI:
return resp.json()
return payload
def stop_job(self, job_id):
url = f"{self.base_url}/v1/jobs/stop/{job_id}"
resp = self.session.get(url)
resp.raise_for_status()
return resp.json()
def delete_job(self, job_id):
url = f"{self.base_url}/v1/jobs/delete/{job_id}"
resp = self.session.get(url)
resp.raise_for_status()
return resp.json()
def start_job(self, job_id, priority=3, limit_recovered=False):
url = f"{self.base_url}/v1/jobs/start/{job_id}"
params = {}
priority = int(priority)
if priority < 1 or priority > 5:
raise ValueError("priority must be an int between 1 and 5")
params["priority"] = priority
params["limit_recovered"] = bool(limit_recovered)
resp = self.session.get(url, params=params)
resp.raise_for_status()
return resp.json()
def download_left_hashes(self, customer_id, hashfile_id, output_file=None):
import sys
url = f"{self.base_url}/v1/hashfiles/{hashfile_id}/left"
hate_crack/main.py
+5 -1
View File
@@ -2000,7 +2000,11 @@ def hashview_api():
# 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'])
stop_after_one = input("Stop after a single result? (y/N): ").strip().upper() == 'Y'
start_result = api_harness.start_job(
job_result['job_id'],
limit_recovered=stop_after_one,
)
print(f"\n✓ Success: {start_result.get('msg', 'Job started')}")
except Exception as e:
print(f"\n✗ Error creating job: {str(e)}")
tests/test_hashview.py
+39 -18
View File
@@ -22,6 +22,23 @@ HASHVIEW_API_KEY = 'test-api-key-123'
class TestHashviewAPI:
"""Test suite for HashviewAPI class with mocked API calls"""
def _get_hashview_config(self):
env_url = os.environ.get('HASHVIEW_URL')
env_key = os.environ.get('HASHVIEW_API_KEY')
if env_url and env_key:
return env_url, env_key
config_path = os.path.join(os.path.dirname(__file__), '..', 'config.json')
try:
with open(config_path) as f:
config = json.load(f)
url = config.get('hashview_url')
key = config.get('hashview_api_key')
if url and key:
return url, key
except Exception:
pass
return env_url, env_key
@pytest.fixture
def api(self):
@@ -55,8 +72,7 @@ class TestHashviewAPI:
def test_list_hashfiles_success(self, api):
"""Test successful hashfile listing with real API if possible, else mock."""
hashview_url = os.environ.get('HASHVIEW_URL')
hashview_api_key = os.environ.get('HASHVIEW_API_KEY')
hashview_url, hashview_api_key = self._get_hashview_config()
if hashview_url and hashview_api_key:
real_api = HashviewAPI(hashview_url, hashview_api_key)
result = real_api.list_hashfiles()
@@ -81,8 +97,7 @@ class TestHashviewAPI:
def test_list_hashfiles_empty(self, api):
"""Test hashfile listing returns empty list if no hashfiles (real API if possible)."""
hashview_url = os.environ.get('HASHVIEW_URL')
hashview_api_key = os.environ.get('HASHVIEW_API_KEY')
hashview_url, hashview_api_key = self._get_hashview_config()
if hashview_url and hashview_api_key:
real_api = HashviewAPI(hashview_url, hashview_api_key)
result = real_api.list_hashfiles()
@@ -101,8 +116,7 @@ class TestHashviewAPI:
def test_get_customer_hashfiles(self, api):
"""Test filtering hashfiles by customer_id (real API if possible)."""
hashview_url = os.environ.get('HASHVIEW_URL')
hashview_api_key = os.environ.get('HASHVIEW_API_KEY')
hashview_url, hashview_api_key = self._get_hashview_config()
customer_id = os.environ.get('HASHVIEW_CUSTOMER_ID')
if hashview_url and hashview_api_key and customer_id:
real_api = HashviewAPI(hashview_url, hashview_api_key)
@@ -129,8 +143,7 @@ class TestHashviewAPI:
def test_upload_cracked_hashes_success(self, api, tmp_path):
"""Test uploading cracked hashes with valid lines (real API if possible)."""
hashview_url = os.environ.get('HASHVIEW_URL')
hashview_api_key = os.environ.get('HASHVIEW_API_KEY')
hashview_url, hashview_api_key = self._get_hashview_config()
hash_type = os.environ.get('HASHVIEW_HASH_TYPE', '1000')
if hashview_url and hashview_api_key:
real_api = HashviewAPI(hashview_url, hashview_api_key)
@@ -184,8 +197,7 @@ class TestHashviewAPI:
def test_create_customer_success(self, api):
"""Test creating a customer (real API if possible)."""
hashview_url = os.environ.get('HASHVIEW_URL')
hashview_api_key = os.environ.get('HASHVIEW_API_KEY')
hashview_url, hashview_api_key = self._get_hashview_config()
if hashview_url and hashview_api_key:
real_api = HashviewAPI(hashview_url, hashview_api_key)
try:
@@ -205,8 +217,7 @@ class TestHashviewAPI:
def test_download_left_hashes(self, api, tmp_path):
"""Test downloading left hashes: real API if possible, else mock."""
hashview_url = os.environ.get('HASHVIEW_URL')
hashview_api_key = os.environ.get('HASHVIEW_API_KEY')
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]):
@@ -240,8 +251,7 @@ class TestHashviewAPI:
def test_download_found_hashes(self, api, tmp_path):
"""Test downloading found hashes: real API if possible, else mock."""
hashview_url = os.environ.get('HASHVIEW_URL')
hashview_api_key = os.environ.get('HASHVIEW_API_KEY')
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]):
@@ -277,8 +287,7 @@ class TestHashviewAPI:
def test_download_wordlist(self, api, tmp_path):
"""Test downloading a wordlist: real API if possible, else mock."""
hashview_url = os.environ.get('HASHVIEW_URL')
hashview_api_key = os.environ.get('HASHVIEW_API_KEY')
hashview_url, hashview_api_key = self._get_hashview_config()
wordlist_id = os.environ.get('HASHVIEW_WORDLIST_ID')
if all([hashview_url, hashview_api_key, wordlist_id]):
real_api = HashviewAPI(hashview_url, hashview_api_key)
@@ -386,8 +395,7 @@ class TestHashviewAPI:
def test_create_job_with_new_customer(self, api, test_hashfile):
"""Test creating a new customer and then creating a job (real API if possible)."""
hashview_url = os.environ.get('HASHVIEW_URL')
hashview_api_key = os.environ.get('HASHVIEW_API_KEY')
hashview_url, hashview_api_key = self._get_hashview_config()
hash_type = os.environ.get('HASHVIEW_HASH_TYPE', '1000')
if hashview_url and hashview_api_key:
real_api = HashviewAPI(hashview_url, hashview_api_key)
@@ -419,6 +427,19 @@ class TestHashviewAPI:
assert job_result is not None
if isinstance(job_result, dict):
assert 'job_id' in job_result
job_id = job_result.get('job_id')
try:
real_api.start_job(job_id)
except Exception:
pass
try:
real_api.stop_job(job_id)
except Exception:
pass
try:
real_api.delete_job(job_id)
except Exception:
pass
except Exception as e:
pytest.skip(f"Real API create_job with new customer not allowed: {e}")
else:
tests/test_hashview_cli_subcommands_subprocess.py
+132 -9
View File
@@ -24,8 +24,29 @@ def _config_has_hashview_key():
return False
def _get_hashview_config():
env_url = os.environ.get("HASHVIEW_URL")
env_key = os.environ.get("HASHVIEW_API_KEY")
if env_url and env_key:
return env_url, env_key
config_path = os.path.join(REPO_ROOT, "config.json")
try:
with open(config_path, "r", encoding="utf-8") as f:
data = json.load(f)
url = data.get("hashview_url")
key = data.get("hashview_api_key")
if url and key:
return url, key
except Exception:
pass
return env_url, env_key
def _ensure_customer_one():
api = HashviewAPI(os.environ["HASHVIEW_URL"], os.environ["HASHVIEW_API_KEY"])
url, key = _get_hashview_config()
if not url or not key:
pytest.skip("Missing hashview_url/hashview_api_key in config.json or env.")
api = HashviewAPI(url, key)
try:
customers_result = api.list_customers()
except Exception as exc:
@@ -87,12 +108,15 @@ def test_hashview_subcommands_require_api_key(tmp_path, args):
reason="Set HASHVIEW_TEST_REAL=1 to run live Hashview subprocess tests.",
)
def test_hashview_subcommands_live_downloads():
required = ["HASHVIEW_URL", "HASHVIEW_API_KEY", "HASHVIEW_HASHFILE_ID"]
required = ["HASHVIEW_HASHFILE_ID"]
missing = [key for key in required if not os.environ.get(key)]
if missing:
pytest.skip(f"Missing required env vars: {', '.join(missing)}")
env = {**os.environ, "PYTHONUNBUFFERED": "1"}
url, key = _get_hashview_config()
if not url or not key:
pytest.skip("Missing hashview_url/hashview_api_key in config.json or env.")
env = {**os.environ, "PYTHONUNBUFFERED": "1", "HASHVIEW_URL": url, "HASHVIEW_API_KEY": key}
base_cmd = [sys.executable, HATE_CRACK_SCRIPT, "hashview"]
customer_id = _ensure_customer_one()
@@ -142,12 +166,15 @@ def test_hashview_subcommands_live_downloads():
reason="Set HASHVIEW_TEST_REAL=1 to run live Hashview subprocess tests.",
)
def test_hashview_subcommands_live_upload_hashfile_job(tmp_path):
required = ["HASHVIEW_URL", "HASHVIEW_API_KEY", "HASHVIEW_HASH_TYPE"]
required = ["HASHVIEW_HASH_TYPE"]
missing = [key for key in required if not os.environ.get(key)]
if missing:
pytest.skip(f"Missing required env vars: {', '.join(missing)}")
env = {**os.environ, "PYTHONUNBUFFERED": "1"}
url, key = _get_hashview_config()
if not url or not key:
pytest.skip("Missing hashview_url/hashview_api_key in config.json or env.")
env = {**os.environ, "PYTHONUNBUFFERED": "1", "HASHVIEW_URL": url, "HASHVIEW_API_KEY": key}
base_cmd = [sys.executable, HATE_CRACK_SCRIPT, "hashview"]
customer_id = _ensure_customer_one()
@@ -179,6 +206,36 @@ def test_hashview_subcommands_live_upload_hashfile_job(tmp_path):
assert run.returncode == 0, output
assert ("Hashfile uploaded" in output) or ("Hashfile added" in output)
assert ("Job created" in output) or ("Failed to add job" in output)
if "Job ID:" in output:
job_id = None
for line in output.splitlines():
if line.strip().startswith("Job ID:"):
try:
job_id = int(line.split("Job ID:")[1].strip())
except Exception:
job_id = None
break
if job_id:
try:
from hate_crack.api import HashviewAPI
url, key = _get_hashview_config()
if not url or not key:
return
api = HashviewAPI(url, key)
try:
api.start_job(job_id)
except Exception:
pass
try:
api.stop_job(job_id)
except Exception:
pass
try:
api.delete_job(job_id)
except Exception:
pass
except Exception:
pass
@pytest.mark.skipif(
@@ -186,12 +243,15 @@ def test_hashview_subcommands_live_upload_hashfile_job(tmp_path):
reason="Set HASHVIEW_TEST_REAL=1 to run live Hashview subprocess tests.",
)
def test_hashview_subcommands_live_upload_hashfile_job_pwdump(tmp_path):
required = ["HASHVIEW_URL", "HASHVIEW_API_KEY"]
required = []
missing = [key for key in required if not os.environ.get(key)]
if missing:
pytest.skip(f"Missing required env vars: {', '.join(missing)}")
env = {**os.environ, "PYTHONUNBUFFERED": "1"}
url, key = _get_hashview_config()
if not url or not key:
pytest.skip("Missing hashview_url/hashview_api_key in config.json or env.")
env = {**os.environ, "PYTHONUNBUFFERED": "1", "HASHVIEW_URL": url, "HASHVIEW_API_KEY": key}
base_cmd = [sys.executable, HATE_CRACK_SCRIPT, "hashview"]
customer_id = _ensure_customer_one()
@@ -226,6 +286,36 @@ def test_hashview_subcommands_live_upload_hashfile_job_pwdump(tmp_path):
assert run.returncode == 0, output
assert ("Hashfile uploaded" in output) or ("Hashfile added" in output)
assert ("Job created" in output) or ("Failed to add job" in output)
if "Job ID:" in output:
job_id = None
for line in output.splitlines():
if line.strip().startswith("Job ID:"):
try:
job_id = int(line.split("Job ID:")[1].strip())
except Exception:
job_id = None
break
if job_id:
try:
from hate_crack.api import HashviewAPI
url, key = _get_hashview_config()
if not url or not key:
return
api = HashviewAPI(url, key)
try:
api.start_job(job_id)
except Exception:
pass
try:
api.stop_job(job_id)
except Exception:
pass
try:
api.delete_job(job_id)
except Exception:
pass
except Exception:
pass
@pytest.mark.skipif(
@@ -233,12 +323,15 @@ def test_hashview_subcommands_live_upload_hashfile_job_pwdump(tmp_path):
reason="Set HASHVIEW_TEST_REAL=1 to run live Hashview subprocess tests.",
)
def test_hashview_subcommands_live_upload_hashfile_job_hashonly(tmp_path):
required = ["HASHVIEW_URL", "HASHVIEW_API_KEY", "HASHVIEW_HASH_TYPE"]
required = ["HASHVIEW_HASH_TYPE"]
missing = [key for key in required if not os.environ.get(key)]
if missing:
pytest.skip(f"Missing required env vars: {', '.join(missing)}")
env = {**os.environ, "PYTHONUNBUFFERED": "1"}
url, key = _get_hashview_config()
if not url or not key:
pytest.skip("Missing hashview_url/hashview_api_key in config.json or env.")
env = {**os.environ, "PYTHONUNBUFFERED": "1", "HASHVIEW_URL": url, "HASHVIEW_API_KEY": key}
base_cmd = [sys.executable, HATE_CRACK_SCRIPT, "hashview"]
customer_id = _ensure_customer_one()
@@ -271,3 +364,33 @@ def test_hashview_subcommands_live_upload_hashfile_job_hashonly(tmp_path):
assert run.returncode == 0, output
assert ("Hashfile uploaded" in output) or ("Hashfile added" in output)
assert ("Job created" in output) or ("Failed to add job" in output)
if "Job ID:" in output:
job_id = None
for line in output.splitlines():
if line.strip().startswith("Job ID:"):
try:
job_id = int(line.split("Job ID:")[1].strip())
except Exception:
job_id = None
break
if job_id:
try:
from hate_crack.api import HashviewAPI
url, key = _get_hashview_config()
if not url or not key:
return
api = HashviewAPI(url, key)
try:
api.start_job(job_id)
except Exception:
pass
try:
api.stop_job(job_id)
except Exception:
pass
try:
api.delete_job(job_id)
except Exception:
pass
except Exception:
pass